首页　>　文章列表　>　Springboot中RocketMQ怎么实现广播消息

Springboot中RocketMQ怎么实现广播消息

springboot RocketMQ
500 2023-05-16

Springboot中RocketMQ怎么实现广播消息

RocketMQ消息模式主要有两种：广播模式、集群模式（负载均衡模式）

广播模式是每个消费者，都会消费消息；

负载均衡模式是每一个消费只会被某一个消费者消费一次；

我们业务上一般用的是负载均衡模式，当然一些特殊场景需要用到广播模式，比如发送一个信息到邮箱，手机，站内提示；

我们可以通过@RocketMQMessageListener的messageModel属性值来设置，MessageModel.BROADCASTING是广播模式，MessageModel.CLUSTERING是默认集群负载均衡模式

下面来介绍下 springboot+rockermq 整合实现广播消息

创建Springboot项目，添加rockermq 依赖

<!--rocketMq依赖-->

<dependency>

    <groupId>org.apache.rocketmq</groupId>

    <artifactId>rocketmq-spring-boot-starter</artifactId>

    <version>2.2.1</version>

</dependency>

配置rocketmq

# 端口

server:

port: 8083
# 配置 rocketmq

rocketmq:

name-server: 127.0.0.1:9876

#生产者

producer:

#生产者组名，规定在一个应用里面必须唯一

group: group1

#消息发送的超时时间默认3000ms

send-message-timeout: 3000

#消息达到4096字节的时候，消息就会被压缩。默认 4096

compress-message-body-threshold: 4096

#最大的消息限制，默认为128K

max-message-size: 4194304

#同步消息发送失败重试次数

retry-times-when-send-failed: 3

#在内部发送失败时是否重试其他代理，这个参数在有多个broker时才生效

retry-next-server: true

#异步消息发送失败重试的次数

retry-times-when-send-async-failed: 3

生产端：新建一个 controller 来做消息发送

生产端按正常发送逻辑发送消息即可

package com.example.springbootrocketdemo.controller;

import org.apache.rocketmq.spring.core.RocketMQTemplate;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.web.bind.annotation.RequestMapping;

import org.springframework.web.bind.annotation.RestController;

/**

 * 广播消息

 * @author qzz

 */

@RestController

public class RocketMQBroadCOntroller {

    @Autowired

    private RocketMQTemplate rocketMQTemplate;

    /**

     * 发送广播消息

     */

    @RequestMapping("/testBroadSend")

    public void testSyncSend(){

        //参数一：topic   如果想添加tag,可以使用"topic:tag"的写法

        //参数二：消息内容

        for(int i=0;i<10;i++){

            rocketMQTemplate.convertAndSend("test-topic-broad","test-message"+i);

        }

    }

}

创建两个消费者来消费消息

我们先集群负载均衡测试，加上messageModel=MessageModel.CLUSTERING

消费者1：

package com.example.springbootrocketdemo.config;

import org.apache.rocketmq.spring.annotation.MessageModel;

import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;

import org.apache.rocketmq.spring.core.RocketMQListener;

import org.springframework.stereotype.Service;

/**

 * 广播消息

 * 配置RocketMQ监听

 * MessageModel.CLUSTERING:集群模式

 * MessageModel.BROADCASTING:广播模式

 * @author qzz

 */

@Service

@RocketMQMessageListener(consumerGroup = "test-broad",topic = "test-topic-broad",messageModel = MessageModel.CLUSTERING)

public class RocketMQBroadConsumerListener implements RocketMQListener<String> {

    @Override

    public void onMessage(String s) {

        System.out.println("集群模式 消费者1，消费消息："+s);

    }

}

消费者2：与消费者1在同一个consumerGroup 和 topic

package com.example.springbootrocketdemo.config;

import org.apache.rocketmq.spring.annotation.MessageModel;

import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;

import org.apache.rocketmq.spring.core.RocketMQListener;

import org.springframework.stereotype.Service;

/**

 * 广播消息

 * 配置RocketMQ监听

 * MessageModel.CLUSTERING:集群模式

 * MessageModel.BROADCASTING:广播模式

 * @author qzz

 */

@Service

@RocketMQMessageListener(consumerGroup = "test-broad",topic = "test-topic-broad",messageModel = MessageModel.CLUSTERING)

public class RocketMQBroadConsumerListener2 implements RocketMQListener<String> {

    @Override

    public void onMessage(String s) {

        System.out.println("集群模式 消费者2，消费消息："+s);

    }

}

启动服务，测试集群模式消费

集群模式测试：两个消费者平摊消息

把上面两个消费者的 messageModel 属性值修改成广播模式

消费者1：

package com.example.springbootrocketdemo.config;

import org.apache.rocketmq.spring.annotation.MessageModel;

import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;

import org.apache.rocketmq.spring.core.RocketMQListener;

import org.springframework.stereotype.Service;

/**

 * 广播消息

 * 配置RocketMQ监听

 * MessageModel.CLUSTERING:集群模式

 * MessageModel.BROADCASTING:广播模式

 * @author qzz

 */

@Service

@RocketMQMessageListener(consumerGroup = "test-broad",topic = "test-topic-broad",messageModel = MessageModel.BROADCASTING)

public class RocketMQBroadConsumerListener implements RocketMQListener<String> {

    @Override

    public void onMessage(String s) {

        System.out.println("广播消息1 广播模式，消费消息："+s);

    }

}

消费者2：与消费者1在同一个consumerGroup 和 topic

package com.example.springbootrocketdemo.config;

import org.apache.rocketmq.spring.annotation.MessageModel;

import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;

import org.apache.rocketmq.spring.core.RocketMQListener;

import org.springframework.stereotype.Service;

/**

 * 广播消息

 * 配置RocketMQ监听

 * MessageModel.CLUSTERING:集群模式

 * MessageModel.BROADCASTING:广播模式

 * @author qzz

 */

@Service

@RocketMQMessageListener(consumerGroup = "test-broad",topic = "test-topic-broad",messageModel = MessageModel.BROADCASTING)

public class RocketMQBroadConsumerListener2 implements RocketMQListener<String> {

    @Override

    public void onMessage(String s) {

        System.out.println("广播消息2 广播模式，消费消息："+s);

    }

}