springcloudstream+rabbitmq小记

此篇主要记录spring-cloud stream&＃xff0c;对rabbitmq的安装使用不做累述。

1. 创建stream-receiver作为消费者
   pom.xml&＃xff1a;

   
cloud-streamcom.cherrish1.0-SNAPSHOT4.0.0stream-receiverorg.springframework.cloudspring-cloud-stream-binder-rabbit

   application.properties&＃xff1a;
    

   spring.application.name&＃61;customer
server.port&＃61;7889spring.rabbitmq.host&＃61;192.168.1.17
spring.rabbitmq.port&＃61;5672
spring.rabbitmq.username&＃61;admin
spring.rabbitmq.password&＃61;adminspring.cloud.stream.bindings.input.destination&＃61;sink-channel
spring.cloud.stream.bindings.output.destination&＃61;sink-channel#不指定该输出通道无法接收消息

   Java 代码&＃xff1a;
    

   /**********************************CustomerApp.java*********************************/
package com.cherrish;import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;/*** &＃64;author cherrishccl* &＃64;time 2018-08-31 14:53* &＃64;name CustomerApp* &＃64;desc:*/
&＃64;SpringBootApplication
public class CustomerApp {private static Logger logger &＃61; LoggerFactory.getLogger(CustomerApp.class);public static void main(String[] args) {SpringApplication.run(CustomerApp.class, args);}}
/**********************************SinkReceiver.java*********************************/
package com.cherrish;import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.messaging.Message;import java.util.concurrent.atomic.AtomicInteger;/*** &＃64;author cherrish* &＃64;time 2018-08-31 14:39* &＃64;name SinkReceiver* &＃64;desc:*/
&＃64;EnableBinding(value &＃61; Sink.class)
public class SinkReceiver {private static final AtomicInteger NUM &＃61; new AtomicInteger(0);private static Logger log &＃61; LoggerFactory.getLogger(SinkReceiver.class);&＃64;StreamListener(Sink.INPUT)public void receive(Message payload) {log.info(NUM.getAndIncrement() &＃43; " Received : " &＃43; payload.getPayload());}
}

    

2. 创建stream-sender作为生产者
   pom.xml&＃xff1a;
    

   
cloud-streamcom.cherrish1.0-SNAPSHOT4.0.0stream-senderorg.springframework.cloudspring-cloud-stream-binder-rabbit

   application.properties&＃xff1a;
    

   spring.application.name&＃61;producer
server.port&＃61;7888spring.rabbitmq.host&＃61;192.168.1.195
spring.rabbitmq.port&＃61;5672
spring.rabbitmq.username&＃61;admin
spring.rabbitmq.password&＃61;adminspring.cloud.stream.bindings.input.destination&＃61;sink-channel
spring.cloud.stream.bindings.output.destination&＃61;sink-channel

   
   Java代码&＃xff1a;

   /**********************************ProducerApp.java*********************************/
package com.cherrish;import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.scheduling.annotation.EnableScheduling;/*** &＃64;author cherrish* &＃64;time 2018-08-31 14:52* &＃64;name ProducerApp* &＃64;desc:*/
&＃64;EnableScheduling
&＃64;SpringBootApplication
public class ProducerApp {private static Logger logger &＃61; LoggerFactory.getLogger(ProducerApp.class);public static void main(String[] args) {SpringApplication.run(ProducerApp.class, args);}
}
/**********************************SinkSender.java*********************************/
package com.cherrish;import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.messaging.Source;
import org.springframework.integration.annotation.InboundChannelAdapter;
import org.springframework.integration.annotation.Poller;import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.atomic.AtomicInteger;/*** &＃64;author cherrish* &＃64;time 2018-08-30 16:16* &＃64;name SinkSender* &＃64;desc:*/
&＃64;EnableBinding(value &＃61; {Source.class})
public class SinkSender {private static Logger log &＃61; LoggerFactory.getLogger(SinkSender.class);private final static AtomicInteger NUM &＃61; new AtomicInteger(0);&＃64;InboundChannelAdapter(value &＃61; Source.OUTPUT,poller &＃61; &＃64;Poller(fixedRate &＃61; "3000"))public String timerMessageSource() {String format &＃61; new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date());log.info(NUM.getAndIncrement() &＃43; " Send message : " &＃43; format);return format;}&＃64;AutowiredSource source;public void send(String message){source.output().send(org.springframework.integration.support.MessageBuilder.withPayload(message).build());}
}
/**********************************ScheduleTimer.java*********************************/package com.cherrish;import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;import java.util.Date;/*** &＃64;author cherrish* &＃64;time 2018-09-03 10:58* &＃64;name ScheduleTimer* &＃64;desc:*/
&＃64;Component
public class ScheduleTimer {&＃64;AutowiredSinkSender sender;&＃64;Scheduled(fixedRate &＃61; 3000)public void send(){sender.send("定时调用发送消息" &＃43; new Date());}
}

   ---------------------------
   父pom.xml&＃xff1a;
    

   
4.0.0com.cherrish1.0-SNAPSHOTcloud-streampomstream-senderstream-receiverFinchley.RELEASEUTF-81.81.81.82.0.4.RELEASEtrueorg.springframework.bootspring-boot-starter-weborg.springframework.cloudspring-cloud-dependencies${spring-cloud.version}pomimportorg.springframework.bootspring-boot-dependencies${spring-boot.version}pomimport