<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>1.5.8.RELEASE</version>
</parent>

<dependencyManagement>
<dependencies>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-dependencies</artifactId>
<version>Ditmars.RELEASE</version>
<type>pom</type>
<scope>import</scope>
</dependency>
</dependencies>
</dependencyManagement>

<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter</artifactId>
<exclusions>
<exclusion>
<groupId>ch.qos.logback</groupId>
<artifactId>logback-classic</artifactId>
</exclusion>
<exclusion>
<groupId>ch.qos.logback</groupId>
<artifactId>logback-core</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-stream-kafka</artifactId>
</dependency>
</dependencies>

<exclusions>
<exclusion>
<groupId>ch.qos.logback</groupId>
<artifactId>logback-classic</artifactId>
</exclusion>
<exclusion>
<groupId>ch.qos.logback</groupId>
<artifactId>logback-core</artifactId>
</exclusion>
</exclusions>

@EnableBinding(value = { Processor.class })
@SpringBootApplication
public class App {

public static void main(String[] args) {
ConfigurableApplicationContext context = SpringApplication.run(App.class);
// ×¢²á´¦Àíº¯Êý
System.out.println("×¢²á½á¹û£º" + setHander(context));
// ·¢ËÍÏûÏ¢
System.out.println("·¢Ëͽá¹û£º" + write(context));
}

// ·¢ËÍÏûÏ¢
public static boolean write(ConfigurableApplicationContext context) {
Service service = context.getBean(Service.class);
return service.write("¹·×ÓÔÚÂð?");
}

// ×¢²á½ÓÊÕµ½ÏûϢʱµÄ´¦Àíº¯Êý
public static boolean setHander(ConfigurableApplicationContext context) {
Service service = context.getBean(Service.class);
return service.subscribe(result -> {
System.out.print("¹·×ÓÊÕµ½ÏûÏ¢£º" + result.getPayload());
});
}
}

@Component
public class Service {

@Autowired
private Processor processor;

public boolean write(String data) {
return processor.output().send(MessageBuilder

.withPayload(data).build());
}

public boolean subscribe(MessageHandler handler) {
return processor.input().subscribe(handler);
}
}

@Autowired
private Processor processor;

processor.output().send(MessageBuilder

.withPayload(data).build());

spring.cloud.stream.bindings.input.destination=test
spring.cloud.stream.bindings.output.destination=test

spring.cloud.stream.bindings.<channelName>.<key>=value

public interface Sink {

String INPUT = "input";

@Input(Sink.INPUT)
SubscribableChannel input();

}

public interface Source {

String OUTPUT = "output";

@Output(Source.OUTPUT)
MessageChannel output();

}

public interface Processor extends

Source, Sink {

}

@Input
public SubscribableChannel yyy();

@Input("zzz")
public SubscribableChannel yyy();

@EnableBinding(value={Sink.class,Source.class})

@Component
public class SendingBean {

private Source source;

@Autowired
public SendingBean(Source source) {
this.source = source;
}

public void sayHello(String name) {
source.output().send(MessageBuilder.with

Payload(name).build());
}
}

@Component
public class SendingBean {

private MessageChannel output;

@Autowired
public SendingBean(MessageChannel output) {
this.output = output;
}

public void sayHello(String name) {
output.send(MessageBuilder.withPay

load(name).build());
}
}

@Component
public class SendingBean {

private MessageChannel output;

@Autowired
public SendingBean(@Qualifier("customOutput") MessageChannel output) {
this.output = output;
}

public void sayHello(String name) {
this.output.send(MessageBuilder.with

Payload(name).build());
}
}

@EnableBinding(Source.class)
public class TimerSource {

@Value("${format}")
private String format;

@Bean
@InboundChannelAdapter(value = Source.OUTPUT, poller = @Poller(fixedDelay = "${fixedDelay}", maxMessagesPerPoll = "1"))
public MessageSource<String> timerMessageSource() {
return () -> new GenericMessage<>(new SimpleDateFormat(format).format(new Date()));
}
}

@EnableBinding(value = { Processor.class })
@SpringBootApplication
public class App {

public static void main(String[] args) {
ConfigurableApplicationContext context = SpringApplication.run(App.class);
// ·¢ËÍÏûÏ¢
System.out.println("·¢Ëͽá¹û£º" + write(context));
}

@StreamListener(Sink.INPUT)
public void handler(String message) {
System.out.print("¹·×ÓÊÕµ½ÏûÏ¢£º" + message);
}

// ·¢ËÍÏûÏ¢
public static boolean write(ConfigurableApplicationContext context) {
Service service = context.getBean(Service.class);
return service.write("¹·×ÓÔÚÂð?");
}
}

public static boolean setHander(ConfigurableApplicationContext context) {
Service service = context.getBean(Service.class);
return service.subscribe(result -> {
System.out.print("¹·×ÓÊÕµ½ÏûÏ¢£º" + result.getPayload());
});
}

@StreamListener(target=Sink.INPUT)
public void handler1(Message message,@Header(name="contentType") Object header) {
System.out.print("¹·×ÓÊÕµ½messageÏûÏ¢£º" + message.getMessage());
System.out.print("ÏûÏ¢header£º" + header);
}

@StreamListener(target=Sink.INPUT)
public void handler1(Message message,@Headers Map<String,Object> header) {
System.out.print("¹·×ÓÊÕµ½messageÏûÏ¢£º" + message.getMessage());
System.out.print("ÏûÏ¢header£º" + header);
}

@EnableBinding(Processor.class)
public class TransformProcessor {

@Autowired
VotingService votingService;

@StreamListener(Processor.INPUT)
@SendTo(Processor.OUTPUT)
public VoteResult handle(Vote vote) {
return votingService.record(vote);
}
}

//Ïȶ¨ÒåÁ½¸öDTO
public class Message {
private String message;
private Integer all;

public String getMessage() {
return message;
}

public void setMessage(String message) {
this.message = message;
}

public Integer getAll() {
return all;
}

public void setAll(Integer all) {
this.all = all;
}

}

public class Error {

private String error;

public String getError() {
return error;
}

public void setError(String error) {
this.error = error;
}

}

@Component
public class Service {

@Autowired
private Processor processor;

public boolean write(Object data) {
return processor.output().send(
MessageBuilder.withPayload(data)

.setHeader("contentType", data.getClass().getSimpleName()).build());
}
}

@EnableBinding(value = { Processor.class })
@SpringBootApplication
public class App {

public static void main(String[] args) {
ConfigurableApplicationContext context = SpringApplication.run(App.class);
// ·¢ËÍÏûÏ¢
Message message = new Message();
message.setAll(200);
message.setMessage("¹·×ÓÔÚÂð£¿");

Error error = new Error();
error.setError("´íÎóºô»½£¡");
write(context, message);
write(context, error);
}

@StreamListener(target = Sink.INPUT, condition = "headers['contentType']=='Message'")
public void handler1(@Payload Message message, @Header("contentType") String header) {
System.out.println("¹·×ÓÊÕµ½messageÏûÏ¢1£º" + message.getMessage());
}

@StreamListener(target = Sink.INPUT, condition = "headers['contentType']=='Error'")
public void handler2(Error message) {
System.out.print("¹·×ÓÊÕµ½errorÏûÏ¢2£º" + message.getError());
}

@StreamListener(target = Sink.INPUT, condition = "headers['contentType']=='Message'")
public void handler3(@Payload Message message, @Header("contentType") String header) {
System.out.println("¹·×ÓÊÕµ½messageÏûÏ¢3£º" + message.getMessage());
}

// ·¢ËÍÏûÏ¢
public static boolean write(ConfigurableApplicationContext context, Object data) {
Service service = context.getBean(Service.class);
return service.write(data);
}
}

¹·×ÓÊÕµ½messageÏûÏ¢1£º¹·×ÓÔÚÂð£¿
¹·×ÓÊÕµ½messageÏûÏ¢3£º¹·×ÓÔÚÂð£¿
¹·×ÓÊÕµ½errorÏûÏ¢2£º´íÎóºô»½£¡

//Source
@SpringBootApplication
@EnableBinding(Source.class)
public class SourceApplication {

@InboundChannelAdapter(value = Source.OUTPUT)
public String timerMessageSource() {
return new SimpleDateFormat().format(new Date());
}
}
//Processor
@SpringBootApplication
@EnableBinding(Processor.class)
public class ProcessorApplication {

@Transformer(inputChannel=Sink.INPUT

,outputChannel=Source.OUTPUT)
public String loggerSink(String payload) {
return payload.toUpperCase();
}
}
//Sink
@SpringBootApplication
@EnableBinding(Sink.class)
public class SinkApplication {

@StreamListener(Sink.INPUT)
public void loggerSink(Object payload) {
System.out.println("Received: " + payload);
}
}

@SpringBootApplication
public class App {

public static void main(String[] args) {
new AggregateApplicationBuilder().from(Source

Application.class).args("--fixedDelay=5000")
.via(ProcessorApplication.class).to(SinkApp

lication.class).args("--debug=true").run(args);
}
}

public static void main(String[] args) {
new AggregateApplicationBuilder()
.from(SourceApplication.class).namespace("from").

args("--fixedDelay=20000")
.via(ProcessorApplication.class).namespace("via")
.to(SinkApplication.class).namespace("to").args("--debug=true").run(args);
}

//»ñÈ¡´«ÈëµÄ²ÎÊý
@Value("${fixedDelay:null}")
private String args;

@InboundChannelAdapter(value = Source.OUTPUT)
public String timerMessageSource() {
//Êä³ö²ÎÊý
System.out.println("Source get args:"+args);
return new SimpleDateFormat().format(new Date());
}

java -jar stream-aggregation.jar

Source get args:20000
Processor get args:null
Sink get args:null
Received: 17-12-14 ÏÂÎç5:43

java -jar stream.jar via --fixedDelay=200

Source get args:20000
Processor get args:200
Sink get args:null
Received: 17-12-14 ÏÂÎç5:46

spring.cloud.stream.bindings.input.destination=test
spring.cloud.stream.bindings.output.destination=test

spring.cloud.stream.bindings.ͨµÀÃû³Æ.binder=BinderÃû³Æ

spring.cloud.stream.defaultBinder=BinderÃû³Æ

rabbit:\org.springframework.cloud.stream

.binder.rabbit.config.RabbitServiceAuto

Configuration

spring.cloud.stream.instanceCount

spring.cloud.stream.instanceIndex

spring.cloud.stream.dynamicDesinations

spring.cloud.stream.defaultBinder

spring.cloud.stream.overrideCloudConnectors

spring.cloud.stream.bindings.<channelName>

.<property>=<value>

spring.cloud.stream.default.<property>=<value>

spring.cloud.stream.channelName.

destinaction=destinaction1,destinaction2