开发笔记:springcloudstream

org.springframework.cloud
spring-cloud-starter-stream-rabbit

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;
@EnableBinding(Sink.class)
public class SinkReceiver {
private static Logger logger= LoggerFactory.getLogger(SinkReceiver.class);
@StreamListener(Sink.INPUT)
public void receive(Object payload){
logger.info("Received: "+payload);
}
}

spring.application.name=stream-hello
spring.rabbitmq.host=10.202.203.29
spring.rabbitmq.port=5672
spring.rabbitmq.username=springcloud
spring.rabbitmq.password=123456

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.context.annotation.Bean;
import org.springframework.integration.annotation.InboundChannelAdapter;
import org.springframework.integration.annotation.Poller;
import org.springframework.integration.core.MessageSource;
import org.springframework.messaging.support.GenericMessage;
import java.util.Date;
@EnableBinding(value = {Sink.class})
public class SinkSender {
private static Logger logger= LoggerFactory.getLogger(SinkSender.class);
@Bean
@InboundChannelAdapter(value = Sink.INPUT,poller = @Poller(fixedDelay = "2000"))
public MessageSource timerMessageSource(){
return ()-> new GenericMessage<>(new Date());
}
}

@Transformer(inputChannel = Sink.INPUT,outputChannel = Sink.INPUT)
public Object transform(Date message){
return new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(message);
}

public class User {
private Integer id;
private String name;
private Integer age;
public Integer getId() {
return id;
}
public void setId(Integer id) {
this.id = id;
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public Integer getAge() {
return age;
}
public void setAge(Integer age) {
this.age = age;
}
}

@Bean
@InboundChannelAdapter(value = Sink.INPUT,poller = @Poller(fixedDelay = "2000"))
public MessageSource timerMessageSource(){
return ()->new GenericMessage<>("{\\"id\\":1,\\"name\\":\\"tom\\",\\"age\\":20}");
}

@ServiceActivator(inputChannel = Sink.INPUT)
public void receive(User user){
logger.info("Received: "+user);
}
@Transformer(inputChannel = Sink.INPUT,outputChannel = Sink.INPUT)
public User transform(String message) throws Exception {
ObjectMapper objectMapper=new ObjectMapper();
User user=objectMapper.readValue(message,User.class);
return user;
}

@StreamListener(Sink.INPUT)
public void receive(User user){
logger.info("Received: "+user);
}

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.Processor;
import org.springframework.messaging.handler.annotation.SendTo;
@EnableBinding(value = {Processor.class})
public class App1 {
private static Logger logger= LoggerFactory.getLogger(App1.class);
@StreamListener(Processor.INPUT)
@SendTo(Processor.OUTPUT)
public Object receiveFromInput(Object payload){
logger.info("Received: "+payload);
return "From Input Channel Return - "+payload;
}
}

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.Processor;
import org.springframework.context.annotation.Bean;
import org.springframework.integration.annotation.InboundChannelAdapter;
import org.springframework.integration.annotation.Poller;
import org.springframework.integration.core.MessageSource;
import org.springframework.messaging.support.GenericMessage;
import java.util.Date;
@EnableBinding(value = {Processor.class})
public class App2 {
private static Logger logger= LoggerFactory.getLogger(App2.class);
@Bean
@InboundChannelAdapter(value = Processor.OUTPUT,poller = @Poller(fixedDelay = "2000"))
public MessageSource timeMessageSource(){
return ()->new GenericMessage<>(new Date());
}
@StreamListener(Processor.INPUT)
public void receiveFromOutput(Object payload){
logger.info("Received: "+payload);
}
}

spring.rabbitmq.host=10.202.203.29
spring.rabbitmq.port=5672
spring.rabbitmq.username=springcloud
spring.rabbitmq.password=123456
spring.cloud.stream.bindings.input.destination=output
spring.cloud.stream.bindings.output.destination=input
server.port=8001