#yyds干货盘点# springcloud整合stream消费自己生产的消息

Posted 2022-01-21 灰太狼_cxh

springcloud整合stream消费自己生产的消息，在​​#yyds干货盘点# springcloud整合stream,rabbitmq实现消息驱动功能 ​​基础代码上进行修改。

1.代码实现：

添加依赖

<dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
        </dependency>

yml配置

spring:
  cloud:
    stream:
      bindings:
        example-topic-input:
          destination: default.messages
          binder: cxh-topic
        example-topic-output:
          destination: default.messages
          binder: cxh-topic
      binders:
        cxh-topic:
          type: rabbit
          environment:
            spring:
              rabbitmq:
                host: localhost
                port: 5672
                username: guest
                password: guest

添加类

@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";
    

public interface TestTopic 

    String OUTPUT = "example-topic-output";
    String INPUT = "example-topic-input";

    @Output(OUTPUT)
    MessageChannel output();

    @Input(INPUT)
    SubscribableChannel input();

@Slf4j
@Component
public class TestListener 

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

@EnableBinding(TestTopic.class)
@SpringBootApplication
public class StreamApplication 

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

2.实现效果：

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

查看控制台输出：

com.cxh.stream.msg.TestListener          : Received2: hello-cxh