SpringCloud之Stream-4.Stream实现消息广

创建消息Producer服务，配置消息主题

这里新建了一个controller，controller里面有一个接口，这个接口是用来生产消息的，然后最后面是一个方法，这个方法用来消费消息

package com.example.streamsample;

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.stream.function.StreamBridge;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.function.Consumer;

@RestController
@RequestMapping
@Configuration
@Slf4j
public class Controller {

    private final StreamBridge streamBridge;

    @Autowired
    public Controller(StreamBridge streamBridge) {
        this.streamBridge = streamBridge;
    }

    @GetMapping("/streamTest")
    public void streamTest(String msg){
        streamBridge.send("mytopic", msg); // 向kafka发送消息
    }

    // 消费者，bean名称必须与配置文件中的bindingName一致
    @Bean
    public Consumer<String> receiver() {
        return msg -> log.info("receive msg:{}", msg);
    }
}

在stream 3.0之后，就如之前帖子讲的，我们可以使用方法来直接消费消息：

    // 消费者，bean名称必须与配置文件中的bindingName一致
    @Bean
    public Consumer<String> receiver() {
        return msg -> log.info("receive msg:{}", msg);
    }

而如上，topic就默认为 receiver-in-0*****

而在下面的配置中就是把这个信道绑定到mytopic，这样的话，生产者发送到的主题就跟消费的主题一致了。

spring.cloud.stream.bindings.receiver-in-0.destination: mytopic

启动多个Consumer节点测试消息广播

然后我们可以以不同的端口号来启动两次这个服务

image.png

注意把这个并行打开就可以同时启动两次这个服务了。

image.png

RabbitMQ界面查看广播组（Exchanges）

然后我们可以调用postman来生产一条消息，然后分别到两个服务的控制台看，发现都接受到了这个topic发出的消息了：

image.png
image.png