RocketMQ收发消息实践

今天药忘吃喽~ 2023-01-21 10:17 152阅读 0赞

### 文章目录 ###

*   *   *  （一）发送消息-生产者
         *   *  【1】添加依赖
             *  【2】设置配置文件
             *  【3】使用Binder发送消息
             *  【4】设置多个消息通道
             *  【5】测试消息发送
         *  （二）接收消息-消费者
         *   *  【1】添加依赖
             *  【2】设置配置文件
             *  【3】定义多通道接收
             *  【4】修改配置文件
             *  【5】定义消息的接收
         *  （三）MQ的生产和接收

### （一）发送消息-生产者 ###

Spring Cloud Alibaba已集成RocketMQ，使用Spring Cloud Stream对RocketMQ发送和接收消息

#### 【1】添加依赖 ####

<dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-web</artifactId>
            </dependency>
            <dependency>
                <groupId>org.springframework.cloud</groupId>
                <artifactId>spring-cloud-stream-binder-rocketmq</artifactId>
                <version>0.9.0.RELEASE</version>
            </dependency>

#### 【2】设置配置文件 ####

name-server指定RockerMQ的NameServer地址，将指定名称为output的Binding的消息发送到TopicTest

#### 【3】使用Binder发送消息 ####

@RestController
    public class SendController { 
        
        @Autowired
        Source source;
        @GetMapping("/send")
        public  String  sendMsg(String message){ 
            MessageBuilder<String> messageBuilder = MessageBuilder.withPayload(message);
            Message<String> mqMessage = messageBuilder.build();
            boolean send = source.output().send(mqMessage);
            return "消息发送"+send;
        }
    }

@SpringBootApplication
    @EnableBinding({ Source.class})
    public class DemoApplication { 
    
        public static void main(String[] args) { 
            SpringApplication.run(DemoApplication.class, args);
        }
    }

说明：@EnableBinding(\{Source.class\})表示绑定配置文件中名称为output的消息通道Binding，Source类中定义的消息通道名称为output，  
GET http://localhost/send?message=XXX  
执行上面请求发送消息的MQ中

#### 【4】设置多个消息通道 ####

上面应用的是单个消息通道，在工作中肯定需要多个不同的消息通道划分业务，需要自定义消息通道的名称，具体的操作是通过自定义类似Source类，修改通道名称好相关配置即可

定义描述通道信息的接口

/** * 定义用户消息通道 * @author : zhangyu */
    public interface UserSource { 
    
        String OUTPUT="userOutput";
    
        @Output(UserSource.OUTPUT)
        MessageChannel output();
    }

启动类启用该通道  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L09jdG9wdXMyMQ_size_16_color_FFFFFF_t_70]  
设置配置文件  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L09jdG9wdXMyMQ_size_16_color_FFFFFF_t_70 1]

server.port=8081
    spring.cloud.stream.rocketmq.binder.name-server=192.168.208.32:9876
    spring.cloud.stream.bindings.output.destination=TopicTest
    spring.cloud.stream.rocketmq.bindings.output.producer.group=demo-group
    
    
    
    spring.cloud.stream.bindings.userOutput.destination=TopicUser
    spring.cloud.stream.rocketmq.bindings.userOutput.producer.group=user-group

#### 【5】测试消息发送 ####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L09jdG9wdXMyMQ_size_16_color_FFFFFF_t_70 2]

代码

@Autowired
        UserSource userSource;
        @GetMapping("/user/send")
        public  String  sendMsgOfUser(String message){ 
            MessageBuilder<String> messageBuilder = MessageBuilder.withPayload(message);
            Message<String> mqMessage = messageBuilder.build();
            boolean send = userSource.output().send(mqMessage);
            return "消息发送"+send;
        }

### （二）接收消息-消费者 ###

上面我们已经能够生产消息，并可以根据不同的业务发送到匹配的通道上去，下面介绍如何接受消息

#### 【1】添加依赖 ####

创建一个新的SpringBoot项目，添加如下依赖

<dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter</artifactId>
            </dependency>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-web</artifactId>
            </dependency>
            <dependency>
                <groupId>org.springframework.cloud</groupId>
                <artifactId>spring-cloud-stream-binder-rocketmq</artifactId>
                <version>0.9.0.RELEASE</version>
            </dependency>

#### 【2】设置配置文件 ####

说明：name-server指定MQ的NameServer地址，destination指定Topic名称，指定名称为input的Binding接受TopicInput的消息

#### 【3】定义多通道接收 ####

上面的生产者我们构建了多个通道的消息，那么消息者也需要支持接收多通道的消息，自定义方式如下，这里创建接口InputChannel，描述需要接收消息的通道

public interface InputChannel { 
    
        String USER_INPUT="userInput";
        String ORDER_INPUT="orderInput";
    
    
        @Input(InputChannel.USER_INPUT)
        SubscribableChannel userInput();
    
        @Input(InputChannel.ORDER_INPUT)
        SubscribableChannel orderInput();
    
    
    }

这里自定义了俩个通道，一个userInput也就是本次生产消费MQ测试的通道，orderInput是一个暂时未使用的通道。

#### 【4】修改配置文件 ####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L09jdG9wdXMyMQ_size_16_color_FFFFFF_t_70 3]

server.port=8082
    spring.cloud.stream.rocketmq.binder.name-server=192.168.208.32:9876
    spring.cloud.stream.bindings.input.destination=TopicInput
    spring.cloud.stream.bindings.input.group=input-group
    
    
    
    spring.cloud.stream.bindings.userInput.destination=TopicUser
    spring.cloud.stream.bindings.userInput.group=user-group
    
    spring.cloud.stream.bindings.orderInput.destination=TopicOrder
    spring.cloud.stream.bindings.orderInput.group=order-group

#### 【5】定义消息的接收 ####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L09jdG9wdXMyMQ_size_16_color_FFFFFF_t_70 4]

@SpringBootApplication
    @EnableBinding({ Sink.class,InputChannel.class})
    public class ConsumerApplication { 
    
        public static void main(String[] args) { 
            SpringApplication.run(ConsumerApplication.class, args);
        }
    
    
        @StreamListener(value = InputChannel.USER_INPUT)
        public void receive(String receiveMsg) { 
            System.out.println("接收消息==》" + receiveMsg);
        }
    
    }

### （三）MQ的生产和接收 ###

启动俩个项目，访问下面接口  
http://localhost:8081/user/send?message=Alice

![在这里插入图片描述][20210515142851280.png]  
![在这里插入图片描述][20210515142855411.png]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L09jdG9wdXMyMQ_size_16_color_FFFFFF_t_70]: /images/20221020/a00fa3fc63d34a5fae4eb97af007cd29.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L09jdG9wdXMyMQ_size_16_color_FFFFFF_t_70 1]: /images/20221020/cf75bdeea57f48d28f52027a7ff6b966.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L09jdG9wdXMyMQ_size_16_color_FFFFFF_t_70 2]: /images/20221020/9b71f71a213c40edb51766129176c18d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L09jdG9wdXMyMQ_size_16_color_FFFFFF_t_70 3]: /images/20221020/f4adacb5a74c4ee2a60c86f542901f22.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L09jdG9wdXMyMQ_size_16_color_FFFFFF_t_70 4]: /images/20221020/15e2b7c6795f41688ee18be99de29e6d.png
[20210515142851280.png]: /images/20221020/ea0bd322f1d24a34a6ceefafc6743bcd.png
[20210515142855411.png]: /images/20221020/a77298d6979d45d99d5a39193333d494.png