SpringCloudStream函数式编程模型

生产者每隔一秒发送一次消息

但实际生产中，更多是由业务场景触发，故不能使用该模式进行配置，SpringCloud还提供了另外一种StreamBridge模式

官方文档介绍

[Sending arbitrary data to an output (e.g. Foreign event-driven sources)] docs.spring.io/spring-cloud-stream/docs/3.1.0/reference/html/spring-cloud-stream.html#_sending_arbitrary_data_to_an_output_e_g_foreign_event_driven_sources

There are cases where the actual source of data may be coming from the external (foreign) system that is not a binder. For example, the source of the data may be a classic REST endpoint. How do we bridge such source with the functional mechanism used by spring-cloud-stream?

Spring Cloud Stream provides two mechanisms, so let’s look at them in more details

Here, for both samples we’ll use a standard MVC endpoint method called delegateToSupplier bound to the root web context, delegating incoming requests to stream via two different mechanisms - imperative (via StreamBridge) and reactive (via EmitterProcessor).

大意是，实际的数据源可以是外部事件驱动，例如通过Rest接口

官方样例

[Using StreamBridge] docs.spring.io/spring-cloud-stream/docs/3.1.0/reference/html/spring-cloud-stream.html#_using_streambridge

@SpringBootApplication
@Controller
public class WebSourceApplication {

	public static void main(String[] args) {
		SpringApplication.run(WebSourceApplication.class, "--spring.cloud.stream.source=toStream");
	}

	@Autowired
	private StreamBridge streamBridge;

	@RequestMapping
	@ResponseStatus(HttpStatus.ACCEPTED)
	public void delegateToSupplier(@RequestBody String body) {
		System.out.println("Sending " + body);
		streamBridge.send("toStream-out-0", body);
	}
}

Here we autowire a StreamBridge bean which allows us to send data to an output binding effectively bridging non-stream application with spring-cloud-stream. Note that preceding example does not have any source functions defined (e.g., Supplier bean) leaving the framework with no trigger to create source bindings, which would be typical for cases where configuration contains function beans. So to trigger the creation of source binding we use spring.cloud.stream.source property where you can declare the name of your sources. The provided name will be used as a trigger to create a source binding. So in the preceding example the name of the output binding will be toStream-out-0 which is consistent with the binding naming convention used by functions . You can use ; to signify multiple sources (e.g., --spring.cloud.stream.source=foo;bar)

@SpringBootApplication
public class ProducerApplication {

    public static void main(String[] args) {
 SpringApplication.run(ProducerApplication.class, args);
    }
}


@RestController
public class Producer {

    @Autowired
    private StreamBridge streamBridge;

    @RequestMapping("/send1")
    public String send1() {
 streamBridge.send("source1-out-0", new Date());
 return "success1";
    }

    @RequestMapping("/send2")
    public String send2() {
 streamBridge.send("source2-out-0", LocalDateTime.now().format(DateTimeFormatter.ISO_LOCAL_DATE_TIME));
 return "success2";
    }

}

spring:
  rabbitmq:
    host: 192.168.133.128
    username: dev-user
    password: devpassword

  cloud:
    stream:
      bindings:
 source1-out-0:
   destination: test2
 source2-out-0:
   destination: test3
    function:
      definition: source1;source2


server:
  port: 8083

@SpringBootApplication
public class ConsumerApplication {

    public static void main(String[] args) {
 SpringApplication.run(ConsumerApplication.class, args);
    }

    @Bean
    public Consumer sink1() {
 return System.out::println;
    }

    @Bean
    public Consumer sink2() {
 return System.out::println;
    }
}

spring:
  rabbitmq:
    host: 192.168.133.128
    username: dev-user
    password: devpassword

  cloud:
    stream:
      bindings:
 sink1-in-0:
   destination: test2
 sink2-in-0:
   destination: test3
    function:
      definition: sink1;sink2

server:
  port: 8081