【Spring Cloud Stream】发送和接收消息

【前置文章】

【Spring Cloud Stream】入门，并简单的配置Consumer进行消息接收

【官方文档】

【本文内容】
本文基于Spring Cloud Stream 3.2.4版本，用RabbitMQ作为消息中间件。

使用Supplier接口发送消息：
- 新建一个bean返回type为Supplier，方法为hello()，会自动创建exchange=hello-out-0
- 默认情况下每1秒钟由Spring框架触发1次。
- 修改默认值：spring.cloud.stream.poller.fixedDelay=5000
- Reactive风格编程，Supplier<Flux<String>> hello()，Flux.just(String)，只会触发一次。
- Reactive风格编程，配合Thread.sleep和Schedulers.elastic()，可实现每1秒发送1次消息。
- Reactive风格编程，使用@PollableBean轻松实现每1秒发送1次消息。
主动发送消息到一个output
- 使用StreamBridge发送消息到exchange=fooout-in-0，配合spring.cloud.stream.output-bindings=fooout使用。
- 使用StreamBridge发送消息到动态destination中，如myDestination，没有该exchange则新建，有则跳过新建。
- 使用ChannelInterceptor拦截消息，配合@GlobalChannelInterceptor使用。
Functional组合：spring.cloud.function.definition=toUpperCase|wrapInQuotes
多个input/output参数的Functions：参考官方文档
Batch Consumers：Consumer<List<Person>> batchPerson()
Batch Producers：Function<String, List<Message<String>>> batch()
Spring IntegrationFlow作为functions：参考官方文档

1. 使用`Supplier`接口发送消息

基于java8的三个接口：

Comsumer相当于接收消息（即消费消息），所以只有input。
Function相关于接收消息送处理后再发送给下游，所以有input/output。
Supplier则是发送消息给下游，即生产者，配置上只有output。

Comsumer和Function在前置文章中有介绍，本章节介绍Supplier。

首先，添加一个Supplier的bean：

    @Bean
    public Supplier<String> hello() {
        return () -> "hello world";
    }

在RabbitMQ console中可以看到创建了exchange：

image.png

并没有queue创建，因为我们的app只需要负责将消息发送给exchange，至于下游的接收，则需要下游app自己处理。

值得注意的是，不管是Function或是Consumer，因为都有input的配置，即消息从上游发送后，当前app中的方法就会被trigger（即方法的触发依赖说消息的到达）。

但Supplier作为发送方，需要谁来触发呢？毕竟它没有绑定到任何一个input上。默认情况下，Spring框架会在启动的时候进行触发，并且每秒会触发一次。

为了验证上述结论，我们可以在项目中再额外定义一个Comsumer用来接收上述Supplier发送的消息：

    @Bean
    public Consumer<String> helloConsumer() {
        return str -> {
            log.info("Received: " + str);
        };
    }

在前置文章#4.5中讲到，如果有1个以上的functional bean，那么就需要显示定义出bean的名字，并且我们将Consumer的input和Supplier的output，指到了同一个exchange中：

spring.cloud.function.definition=helloConsumer;hello
spring.cloud.stream.bindings.helloConsumer-in-0.destination=hello
spring.cloud.stream.bindings.hello-out-0.destination=hello

启动程序后，可以看到在Consumer端源源不断的接收到来自Supplier的消息，而且间隔就是1秒：

image.png

默认的参数可以修改，详细看：https://docs.spring.io/spring-cloud-stream/docs/3.2.4/reference/html/spring-cloud-stream.html#_polling_configuration_properties
比如我们加上spring.cloud.stream.poller.fixedDelay=5000，那么Supplier消息就会每隔5秒发送一次。

接下来，我们尝试使用reactive的编程风格来编写Supplier的bean，下面的这个方法，其实只会被调用一次：

    @Bean
    public Supplier<Flux<String>> hello() {
        return () -> Flux.just("hello from reactive style");
    }

这个比较好测试，我们直接在exchange=hello上面手动binding一个queue=my-queue：

image.png

启动程序后发现，只能接收到一条message，就再也没有新的message了。

官方给了reactive想要实现每隔一秒发送消息的示例，这里使用了Schedulers以及让这个Thread sleep 1秒：

@Bean
    public Supplier<Flux<String>> hello() {
        return () -> Flux.fromStream(Stream.generate(new Supplier<String>() {
            @Override
            public String get() {
                try {
                    Thread.sleep(1000L);
                } catch (InterruptedException e) {
                    // ignore
                }
                return "Hello from Supplier";
            }
        })).subscribeOn(Schedulers.elastic()).share();
    }

那么，有没有一种办法，可以用reactive的编程风格来编写Supplier并且用更为简单的方式实现每隔1秒发送一次消息呢？（为什么要执着于每隔多久发送一次，这里的use case可以是我们需要源源不断的从一个数据源中主动拉取数据，处理好之后再发送给下游）。

答案是有的，可以使用@PollableBean，以下的代码也会每隔1秒钟发送一次消息：

    @PollableBean
    public Supplier<Flux<String>> hello() {
        return () -> Flux.just("hello");
    }

2. 主动发送消息到一个output

很多时候我们并不能使用Supplier来处理消息，主要原因是有些驱动并不能在启动时就设定好，而是有可能是通过REST api来发送一个消息，对于这种use case，Spring Cloud Stream提供了两种方式：

2.1 方式1：使用`StreamBridge`发送消息到output

@RestController
public class StreamBridgeController {
    @Autowired
    private StreamBridge streamBridge;

    @GetMapping("fooout")
    public String delegateToSupplier() {
        streamBridge.send("fooout-out-0", "hello from streamBridge");
        return "success";
    }
}

这时候就用到了上面#4.6的配置了，即没有绑定到具体的function的时候，需要配置spring.cloud.stream.input/output-binings属性

spring.cloud.stream.output-bindings=fooout

启动项目后，自动创建exchange：

image.png

为了测试我们手动创建一个queue=my-queue，并绑定到这个exchange上：

image.png

调用API：http://localhost:8080/fooout后，可以接收到消息：

image.png

在这里我们自动装载了一个StreamBridge bean，使得可以发送消息到指定的output binding上。另外streamBridge.send(..)发送的消息是一个Object对象，即不仅可以发送String，还可以发送一个对象作为message，默认的conversion是json格式，比如我们尝试发送一个Student对象：

        streamBridge.send("fooout-out-0", new Student(1, "test"));

image.png

2.2 方式2：使用`StreamBridge`发送到动态的destination中

我们尝试发送消息到exchange=myDestination中，但不同于上个例子，这里的myDestination没有在application.properties中配置，因此，这样的exchange name会被当作是动态的destination（目的地）。

如果exchange=myDestination不存在，则会先创建出来，如果存在则跳过创建。

    @GetMapping("dynamic")
    public String dynamicDestination() {
        streamBridge.send("myDestination", "hello dynamic destination.");
        return "success";
    }

动态新增的exchange：

image.png

2.3 `StreamBridge.send(...)`其它方法：

boolean send(String bindingName, Object data, MimeType outputContentType)：默认的content type是application/json，但在send的时候可以指定content type。
boolean send(String bindingName, @Nullable String binderName, Object data)：Spring Cloud Stream支持多个binder的case，比如从Kafka收消息再发送到RabbitMQ。通过StreamBridge发送时，可以指定相应的binderName。

2.4 使用拦截器拦截`StreamBridge`

StreamBridge的send方法内部用的是MessageChannel来建立output binding，MessageChannel位于spring-messaging.jar中，即这是Spring messaging中的代码，Jms用的也是这套，并不是Spring Cloud Stream独有。

关于MessageChannel的Interceptor，这篇文章更为详细：https://dzone.com/articles/spring-cloud-stream-channel-interceptor

MessageChannel可以用ChannelInterceptor来拦截。这个接口有6个方法，基本涵盖了接收消息的各个阶段：

preSend(...)
postSend(...)
afterSendCompletion(...)
preReceive(...)
postReceive(...)
afterReceiveCompletion(...)

先新建一个ChannelInterceptor的bean，Spring Cloud Stream并不会注入这个interceptor，除非这个bean加上@GlobalChannelInterceptor(patterns = "*")。

    @Bean
    @GlobalChannelInterceptor(patterns = "*")
    public ChannelInterceptor customInterceptor() {
        return new ChannelInterceptor() {
            @Override
            public Message<?> preSend(Message<?> message, MessageChannel channel) {
                String bytes = new String((byte[]) message.getPayload(), StandardCharsets.UTF_8);
                log.info("interceptor message: {}", bytes);
                return message;
            }
        };
    }

【测试】
访问上述#2.1的地址：http://localhost:8080/fooout，控制台日志：

2023-01-24 12:23:57.308 INFO 25053 --- [nio-8080-exec-1] com.config.MessagInterceptor : interceptor message: {"id":1,"name":"test"}

访问上述#2.2的地址：http://localhost:8080/dynamic，控制台日志：

2023-01-24 12:24:45.136 INFO 25053 --- [nio-8080-exec-2] com.config.MessagInterceptor : interceptor message: hello dynamic destination.

上述的@GlobalChannelInterceptor(patterns = "*")会拦截所有的binding，如果只想拦截#2.1的fooout-out-0的binding，那么可以把patterns改成：@GlobalChannelInterceptor(patterns = "fooout-*")。

3. Functional组合

使用functional编程还可以将一系列的functional组合起来使用。例如有两个Function bean：

    @Bean
    public Function<String, String> toUpperCase() {
        return s -> s.toUpperCase();
    }

    @Bean
    public Function<String, String> wrapInQuotes() {
        return s -> "\"" + s + "\"";
    }

超过1个以上的functional我们需要配置spring.cloud.function.definition，在前置文章#4.5中有介绍，可以使用;隔开。

但如果我们用|隔开，就表示functional的组合，即消息从toUpperCase接收，处理（这个方法的处理结果是把消息变成大写），然后再发送给下游，下游是wrapInQuotes方法（这个方法是把消息加上双引号），然后再发送给下游：

spring.cloud.function.definition=toUpperCase|wrapInQuotes

【测试】

image.png

我们测试的开始端是RabbitMQ console --> 在exchange=toUpperCasewrapInQuotes-in-0中发送消息 --> 消息被toUpperCase()接收 --> 转大写后交给下一个functional处理（这里也是通过发消息），即wrapInQuotes() --> 给消息加上双引号 --> 发送消息给toUpperCasewrapInQuotes-out-0 --> 为了测试方便，我们新建一个my-queue，来接收消息。

首先发送消息：

image.png

再看到消息拦截器preSend方法发送的消息：

image.png

最后my-queue收到的消息，可以看到转成了大写后又加了括号：

image.png

如果觉得名字太长，可以按前置文章的#4.2章，可以先给binding标下name：

spring.cloud.function.definition=toUpperCase|wrapInQuotes
spring.cloud.stream.function.bindings.toUpperCase|wrapInQuotes-in-0=input
spring.cloud.stream.bindings.input.destination=my-input

这样原先的exchange=toUpperCasewrapInQuotes-in-0就会变为my-input。

4. 多个input/output参数的Functions

为什么会有这种需求？

大数据：你的数据来源可能不止一个地方。
数据聚合：有时候需要我们将来自不同stream的数据进行聚合。

具体查看官方文档：https://docs.spring.io/spring-cloud-stream/docs/3.2.4/reference/html/spring-cloud-stream.html#_functions_with_multiple_input_and_output_arguments

5. Batch Consumers/Producers

5.1 Batch Consumers

具体查看官方文档：

图来自文章，虽然文章用的是中间件是solace，但图表达的很好：

image.png

首先，定义一个bean为Consumer的方法，可以批量接收Person，并打印出来：

    @Bean
    public Consumer<List<Person>> batchPerson() {
        return persons -> {
            log.info("Received " + persons.size());
            persons.forEach(p -> log.info("received: " + p));
        };
    }

需要相应的配置：

spring.cloud.stream.bindings.batchPerson-in-0.consumer.batch-mode=true

spring.cloud.stream.rabbit.bindings.batchPerson-in-0.consumer.enable-batching=true
spring.cloud.stream.rabbit.bindings.batchPerson-in-0.consumer.batch-size=10
spring.cloud.stream.rabbit.bindings.batchPerson-in-0.consumer.receive-timeout=200

【测试】
为了方便测试，写了测试代码：

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @GetMapping("batch-consumer")
    public String batch() {
        rabbitTemplate.convertAndSend("batchPerson-in-0", "#", "{\"name\":\"aaa\"}");
        rabbitTemplate.convertAndSend("batchPerson-in-0", "#", "{\"name\":\"bbb\"}");
        return "success";
    }

浏览器调用http://localhost:8080/batch-consumer后，日志打印：

image.png

5.2 Batch Producers

image.png

首先，创建producer bean，该方法会接收1个字符串，然后再分发出4条消息：

@Bean
    public Function<String, List<Message<String>>> batch() {
        return p -> {
            List<Message<String>> list = new ArrayList<>();
            list.add(MessageBuilder.withPayload(p + ":1").build());
            list.add(MessageBuilder.withPayload(p + ":2").build());
            list.add(MessageBuilder.withPayload(p + ":3").build());
            list.add(MessageBuilder.withPayload(p + ":4").build());
            return list;
        };
    }

【测试】
为了方便测试，我们写一个Consumer来接收上述批量Producer的结果：

    @Bean
    public Consumer<String> batchConsumer() {
        return str -> log.info("received: " + str);
    }

超过1个functional时，需要显性指定名字，以及我们把batchConsumer的in指定为上述batch producer的下游：

spring.cloud.function.definition=batch;batchConsumer
spring.cloud.stream.bindings.batchConsumer-in-0.destination=batch-out-0

在RabbitMQ console，exchange=batch-in-0中发布1条消息：

image.png

batchConsumer端收到了4条消息：

image.png

6. Spring IntegrationFlow作为functions

Spring Cloud Stream和Spring Integration集成，详细参考官方文档：https://docs.spring.io/spring-cloud-stream/docs/3.2.4/reference/html/spring-cloud-stream.html#_spring_integration_flow_as_functions

【Spring Cloud Stream】发送和接收消息

1. 使用Supplier接口发送消息

2. 主动发送消息到一个output

2.1 方式1：使用StreamBridge发送消息到output

2.2 方式2：使用StreamBridge发送到动态的destination中

2.3 StreamBridge.send(...)其它方法：

2.4 使用拦截器拦截StreamBridge

3. Functional组合

4. 多个input/output参数的Functions

5. Batch Consumers/Producers

5.1 Batch Consumers

5.2 Batch Producers

6. Spring IntegrationFlow作为functions

推荐阅读更多精彩内容

1. 使用`Supplier`接口发送消息

2.1 方式1：使用`StreamBridge`发送消息到output

2.2 方式2：使用`StreamBridge`发送到动态的destination中

2.3 `StreamBridge.send(...)`其它方法：

2.4 使用拦截器拦截`StreamBridge`