#yyds干货盘点# springcloud整合stream实现同一通道根据消息内容分发不同的消费逻辑

Posted 2022-01-22 灰太狼_cxh

springcloud整合stream实现同一通道根据消息内容分发不同的消费逻辑，在​​#yyds干货盘点# springcloud整合stream消费自己生产的消息​​基础代码上进行修改。

1.代码实现：

修改类功能

@RestController
public class TestController 

    @Autowired
    private TestTopic testTopic;

    @GetMapping("/sendMessage")
    public String messageWithMQ(@RequestParam String message) 
        testTopic.output().send(MessageBuilder.withPayload(message).build());
        return "ok";
    

    @GetMapping("/sendMessage2")
    public String sendMessage2(@RequestParam String message) 
        testTopic.output().send(MessageBuilder.withPayload(message).setHeader("version", "1.0").build());
        testTopic.output().send(MessageBuilder.withPayload(message).setHeader("version", "2.0").build());
        return "ok";
    

@Slf4j
@Component
public class TestListener 

    @StreamListener(TestTopic.INPUT)
    public void receive(String payload) 
        log.info("Received2: " + payload);
    

    @StreamListener(value = TestTopic.INPUT, condition = "headers[version]==1.0")
    public void receiveV1(String payload, @Header("version") String version) 
        log.info("Received v1 : " + payload + ", " + version);
    

    @StreamListener(value = TestTopic.INPUT, condition = "headers[version]==2.0")
    public void receiveV2(String payload, @Header("version") String version) 
        log.info("Received v2 : " + payload + ", " + version);
    

2.实现效果：

启动rabbitmq, 项目stream, 打开浏览器​​http://localhost:8080/sendMessage2?message=hello-cxh​​

查看控制台输出：