SpringCloud Stream操作消息队列

浅浅的花香味﹌ 2023-07-05 14:28 30阅读 0赞

### 概述 ###

![format_png][]

SpringCloud Stream是为微服务提供消息驱动能力的组件，使用就可以操作消息队列。

应用程序通过input和output与stream的binder通信，而binder与消息中间件(MQ)通信。使用stream是对消息中间件的进一步封装，无视不同中间件的API，做到统一封装，可以做到代码层面的无感知，甚至动态切换中间件，不过只目前只支持RabbitMQ和kafka。

### 集成 ###

pom引入依赖，我使用的是rabbitmq，使用kafka就引入kafka

<dependency>
       <groupId>org.springframework.cloud</groupId>
       <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
    </dependency>

配置文件配置mq信息，将消息分组，queue为myMessageInput.order，消息格式为json

spring:
      rabbitmq:
        host: 
        port: 5672
        username: guest
        password: guest
      cloud:
        stream:
          bindings:
            myMessageInput:
              group: order
              content-type: application/json

新建一个Stream接口  StreamClient

package com.viki.order.message;
    
    
    import org.springframework.cloud.stream.annotation.Input;
    import org.springframework.cloud.stream.annotation.Output;
    import org.springframework.messaging.MessageChannel;
    import org.springframework.messaging.SubscribableChannel;
    
    /**
     * @Author Sakura
     * @Date 10/11/2019
     **/
    
    public interface StreamClient {
    
        String INPUT = "myMessageInput";
    
        String OUTPUT = "myMessageOutput";
    
        //接收消息，使用SubscribableChannel 
        @Input(StreamClient.INPUT)
        SubscribableChannel input();
    
        //发送消息，使用MessageChannel 
        @Output(StreamClient.OUTPUT)
        MessageChannel output();
    }

新建消息接收端  StreamReceiver

package com.viki.order.message;
    
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.cloud.stream.annotation.EnableBinding;
    import org.springframework.cloud.stream.annotation.StreamListener;
    import org.springframework.stereotype.Component;
    
    /**
     * @Author Sakura
     * @Date 10/11/2019
     **/
    @Component
    @Slf4j
    @EnableBinding(StreamClient.class) //上边定义的接口文件
    public class StreamReceiver {
    
        @StreamListener(StreamClient.OUTPUT)
        public void processOutput(Object message){
            log.info("StreamReceiver output : {}", message);
        }
    }

新建发送端controller

package com.viki.order.controller;
    
    import com.viki.order.dto.OrderDTO;
    import com.viki.order.message.StreamClient;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.data.domain.jaxb.SpringDataJaxb;
    import org.springframework.messaging.support.MessageBuilder;
    import org.springframework.web.bind.annotation.GetMapping;
    import org.springframework.web.bind.annotation.RestController;
    
    import java.util.Date;
    
    /**
     * @Author Sakura
     * @Date 10/11/2019
     **/
    @RestController
    public class SendMessageController {
    
        @Autowired
        private StreamClient streamClient;
    
        @GetMapping("/sendMessage")
        public void sendMessage(){
            String message = "send to mq: now is " + new Date();
            streamClient.output().send(MessageBuilder.withPayload(message).build());
        }
    
    }

此时StreamClient 调用的output方法，与接收端StreamListener的值应一致

启动项目，mq就会新增一个queue

![20200215174000673.png][]

此时访问 [http://localhost:8080/sendMessage][http_localhost_8080_sendMessage] 接口

就会在控制台收到

![20200215174045466.png][]

[format_png]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9pbWFnZXMyMDE4LmNuYmxvZ3MuY29tL2Jsb2cvMTIwMjYzOC8yMDE4MDUvMTIwMjYzOC0yMDE4MDUyODIwNDQwMDAxMS0xOTk2NTUxMzcxLnBuZw?x-oss-process=image/format,png
[20200215174000673.png]: https://img-blog.csdnimg.cn/20200215174000673.png
[http_localhost_8080_sendMessage]: http://localhost:8081/sendMessage
[20200215174045466.png]: https://img-blog.csdnimg.cn/20200215174045466.png