跳至主要內容

08-SpringBoot整合RocketMQ

Ms.Zyh2026/9/27大约 8 分钟12-MQ常用

一，整合SpringBoot

创建SpringBoot项目，引入jar包

<dependency>
	<groupId>org.apache.rocketmq</groupId>
	<artifactId>rocketmq-spring-boot-starter</artifactId>
	<version>2.1.1</version>
</dependency>

引入配置信息：

# NameServer地址
rocketmq.name-server = 1.15.141.218:9876
# 默认的消息生产者组
rocketmq.producer.group = springBootGroup

1.1 消息的发送

1.1.1 单向发送消息
@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringRocketTest {
    @Resource
    private RocketMQTemplate rocketMQTemplate;
	@Test
    public void sendMessageTest(){
        // 单项发送消息
        rocketMQTemplate.sendOneWay("springBootTopic","Hello, World!");
        
    }
}
1.1.2 同步发送消息
@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringRocketTest {
    @Resource
    private RocketMQTemplate rocketMQTemplate;
	@Test
    public void sendMessageTest(){
        // 同步发送消息
        SendResult sendResult = rocketMQTemplate.syncSend(springTopic, "Hello, World!");
    }
}
1.1.3 异步发送消息
@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringRocketTest {
    @Resource
    private RocketMQTemplate rocketMQTemplate;
	@Test
    public void sendMessageTest(){
         rocketMQTemplate.asyncSend("springBootTopic", "Hello, World!", new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.println("发送成功！");
            }

            @Override
            public void onException(Throwable throwable) {
                System.out.println("发送失败！");
            }
        });
    }
}
1.1.4 sendAndReceive方法
@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringRocketTest {
    @Resource
    private RocketMQTemplate rocketMQTemplate;
	@Test
    public void sendMessageTest(){
         // 同步发送消息并且返回一个String类型的结果。
        String replyString = rocketMQTemplate.sendAndReceive("springBootTopic", "request string", String.class);
     
        
        // 同步发送消息并且返回一个Byte数组类型的结果。
        byte[] replyBytes = rocketMQTemplate.sendAndReceive("springBootTopic", MessageBuilder.withPayload("request byte[]").build(), byte[].class, 3000);
       
         // 同步发送一个带hash参数的请求(排序消息)，并返回一个User类型的结果
        User requestUser = new User().setUserAge((byte) 9).setUserName("requestUserName");
        User replyUser = rocketMQTemplate.sendAndReceive("springBootTopic", requestUser, User.class, "order-id");
    
        // 同步发送一个带延迟级别的消息(延迟消息)，并返回一个泛型结果
        ProductWithPayload<String> replyGenericObject = rocketMQTemplate.sendAndReceive("springBootTopic", "request generic",
                new TypeReference<ProductWithPayload<String>>() {
                }.getType(), 30000, 2);
        
        // 异步发送消息，返回String类型结果。
        rocketMQTemplate.sendAndReceive("springBootTopic", "request string", new RocketMQLocalRequestCallback<String>() {
            @Override public void onSuccess(String message) {
                System.out.printf("send %s and receive %s %n", "request string", message);
            }

            @Override public void onException(Throwable e) {
                e.printStackTrace();
            }
        });
        // 异步发送消息，并返回一个User类型的结果。
        rocketMQTemplate.sendAndReceive("springBootTopic", new User().setUserAge((byte) 9).setUserName("requestUserName"), new RocketMQLocalRequestCallback<User>() {
            @Override public void onSuccess(User message) {
               
            }
            @Override public void onException(Throwable e) {
                e.printStackTrace();
            }
        }, 5000);
    }
}

1.2 消息的接收

@Component
@RocketMQMessageListener(consumerGroup = "springBootConsumerGroup", topic = "springBootTopic",consumeMode= ConsumeMode.CONCURRENTLY,messageModel = MessageModel.BROADCASTING)
// consumerGroup: 消费者组；
// topic：订阅的主题；consumeMode：
// ConsumeMode(同时模式),ORDERLY(顺序模式)
// messageModel：CLUSTERING(集群模式模式),BROADCASTING(广播模式)
public class SpringConsumer implements RocketMQListener<String> {
    @Override
    public void onMessage(String message) {
        System.out.println("Received message : "+ message);
    }
}

1.3 顺序消息

消息的生产者：

@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringRocketTest {
    @Resource
    private RocketMQTemplate rocketMQTemplate;
	@Test
    public void sendMessageTest(){
        // 参数一：topic   如果想添加tag,可以使用"topic:tag"的写法
        // 参数二：消息内容
        // 参数三：hashKey 用来计算决定消息发送到哪个消息队列
        
        // 单向
        rocketMQTemplate.sendOneWayOrderly("springBootTopic","Hello, World!","hashKey"); 
        // 同步
        SendResult sendResult = rocketMQTemplate.syncSendOrderly("springBootTopic", "Hello, World!","hashKey"); 
        // 异步
        rocketMQTemplate.asyncSendOrderly("springBootTopic", "Hello, World!", "hashKey", new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.println("发送成功！");
            }

            @Override
            public void onException(Throwable throwable) {
                System.out.println("发送失败！");
            }
        });
    }
}

消息的消费者：consumeMode= ConsumeMode.ORDERLY

@Component
@RocketMQMessageListener(consumerGroup = "springBootConsumerGroup", topic = "springBootTopic",consumeMode= ConsumeMode.ORDERLY) 
// consumeMode：ConsumeMode(同时模式),ORDERLY(顺序模式)
public class SpringConsumer implements RocketMQListener<String> {
    @Override
    public void onMessage(String message) {
        System.out.println("Received message : "+ message);
    }
}

1.4 广播消息

在集群状态 MessageModel.CLUSTERING下，每一条消息只会被同一个消费者组中的一个实例消费到，这跟 kafka和rabbitMQ的集群模式是一样的，而广播模式MessageModel.BROADCASTING则是把消息发给了所有订阅了对应主题的消费者，而不管消费者是不是同一个消费者组。

消息的生产者和普通的没什么区别，消息的消费者如下：

@Component
@RocketMQMessageListener(consumerGroup = "springBootConsumerGroup", topic = "springBootTopic",consumeMode= ConsumeMode.CONCURRENTLY,messageModel = MessageModel.BROADCASTING)
// messageModel：CLUSTERING(集群模式模式),BROADCASTING(广播模式)
public class SpringConsumer implements RocketMQListener<String> {
    @Override
    public void onMessage(String message) {
        System.out.println("Received message : "+ message);
    }
}

1.5 批量消息

批量发送消息能显著提高传递消息的性能。限制是这些批量消息应该有相同的topic，而且不能是延时消息。此外，这一批消息的总大小不应超过4MB，如果超过可以有2种处理方案：

  • 将消息进行切割成多个小于4M的内容进行发送

  • 修改4M的限制改成更大，可以设置Producer的maxMessageSize属性，修改配置文件中的maxMessageSize属性。

@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringRocketTest {
    @Resource
    private RocketMQTemplate rocketMQTemplate;
	@Test
    public void sendMessageTest(){
        sList<Message<String>> messageList = new ArrayList<>();
        for(int i=0;i<10;i++){
            messageList.add(MessageBuilder.withPayload("批量消息"+(i+1)).build());
        }
        //参数一：topic 如果想添加tag,可以使用"topic:tag"的写法
        //参数二：消息内容
        SendResult sendResult = rocketMQTemplate.syncSend("springBootTopic",messageList);
        System.out.println(sendResult);
    }
}

如果发送的消息大于4M：

class ListSplitter implements Iterator<List<Message>> {
    private int sizeLimit = 1000 * 1000;
    private final List<Message> messages;
    private int currIndex;

    public ListSplitter(List<Message> messages) {
        this.messages = messages;
    }

    @Override
    public boolean hasNext() {
        return currIndex < messages.size();
    }

    @Override
    public List<Message> next() {
        int nextIndex = currIndex;
        int totalSize = 0;
        for (; nextIndex < messages.size(); nextIndex++) {
            Message message = messages.get(nextIndex);
            int tmpSize = message.getTopic().length() + message.getBody().length;
            Map<String, String> properties = message.getProperties();
            for (Map.Entry<String, String> entry : properties.entrySet()) {
                tmpSize += entry.getKey().length() + entry.getValue().length();
            }
            tmpSize = tmpSize + 20; 
            if (tmpSize > sizeLimit) {
                if (nextIndex - currIndex == 0) {
                    nextIndex++;
                }
                break;
            }
            if (tmpSize + totalSize > sizeLimit) {
                break;
            } else {
                totalSize += tmpSize;
            }

        }
        List<Message> subList = messages.subList(currIndex, nextIndex);
        currIndex = nextIndex;
        return subList;
    }

    @Override
    public void remove() {
        throw new UnsupportedOperationException("Not allowed to remove");
    }
}
@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringRocketTest {
    @Resource
    private RocketMQTemplate rocketMQTemplate;
	@Test
    public void sendMessageTest(){
       List<Message> messageList = new ArrayList<>();
        for(int i=0;i<1000;i++){
            byte[] bytes = (("批量消息"+i).getBytes(CharsetUtil.UTF_8));
            messageList.add(new Message("message-topic-batch","tag"+i,bytes));
        }
        // 切割消息,把大的消息分裂传给你若干个小的消息
        ListSplitter splitter = new ListSplitter(messageList);
        while(splitter.hasNext()){
            List<Message> listItem = splitter.next();
            // 发送消息
            //参数一：topic 如果想添加tag,可以使用"topic:tag"的写法
            //参数二：消息内容
            SendResult sendResult = rocketMQTemplate.syncSend("springBootTopic",messageList,6000);
            System.out.println(sendResult);
        }
    }
}

1.6 过滤消息

1.6.1 TagFilter

消息的生产者：

@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringRocketTest {
    @Resource
    private RocketMQTemplate rocketMQTemplate;
	@Test
    public void sendMessageTest(){
       String[] tags = new String[] {"TagA", "TagB", "TagC"};
        for (int i = 0; i < 15; i++) {
            String tag = tags[i % tags.length];
            rocketMQTemplate.syncSend("TagFilterTopic:"+tag,tag+"-Hello world");
        }
    }
}

消息的消费者：

@Component
@RocketMQMessageListener(consumerGroup = "TagFilterConsumerGroup", topic = "TagFilterTopic"
        ,consumeMode= ConsumeMode.CONCURRENTLY
        ,messageModel = MessageModel.CLUSTERING
        ,selectorExpression = "TagA || TagC"
        ,selectorType = SelectorType.TAG
)
// selectorExpression: tag表达式；selectorType：表达式类型TAG和SQL92
public class SpringConsumer implements RocketMQListener<String> {
    @Override
    public void onMessage(String message) {
        System.out.println("收到消息内容 "+ message);
    }
}

控制台输出：

收到消息内容 TagA-Hello world
收到消息内容 TagC-Hello world
收到消息内容 TagC-Hello world
收到消息内容 TagC-Hello world
收到消息内容 TagA-Hello world
收到消息内容 TagC-Hello world
收到消息内容 TagA-Hello world
收到消息内容 TagA-Hello world
收到消息内容 TagC-Hello world
收到消息内容 TagA-Hello world

可以看出输出收到的消息是TagA和TagC，TagB是没有接收到的。

1.6.2 SqlFilter

消息的生产者：

@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringRocketTest {
    @Resource
    private RocketMQTemplate rocketMQTemplate;
	@Test
    public void sendMessageTest(){
        String[] tags = new String[] {"TagA", "TagB", "TagC"};
        for (int i = 0; i < 15; i++) {
            String tag = tags[i % tags.length];
            Map<String,Object> headers = new HashMap<>();//设置消息的属性（header）信息
            headers.put("otherProperty", String.valueOf(i));
            rocketMQTemplate.convertAndSend("SqlFilterTopic:"+tag,tag+" - Hello RocketMQ - " + i,headers);
        }
    }
}

消息的消费者：

@Component
@RocketMQMessageListener(consumerGroup = "SqlFilterConsumerGroup", topic = "SqlFilterTopic"
        ,consumeMode= ConsumeMode.CONCURRENTLY
        ,messageModel = MessageModel.BROADCASTING
        ,selectorExpression = "(TAGS is not null and TAGS in ('TagA', 'TagB'))" +
        "and (otherProperty is not null and otherProperty between 0 and 3)"
        ,selectorType = SelectorType.SQL92
)
// selectorExpression: tag表达式；selectorType：表达式类型TAG和SQL92
public class SpringConsumer implements RocketMQListener<String> {
    @Override
    public void onMessage(String message) {
        System.out.println("收到消息内容 "+ message);
    }
}

控制台输出：

收到消息内容 TagB - Hello RocketMQ - 1
收到消息内容 TagA - Hello RocketMQ - 3
收到消息内容 TagA - Hello RocketMQ - 0

1.7 延迟消息

延迟消息实现的效果就是在调用producer.send方法后，消息并不会立即发送出去，而是会等一 段时间再发送出去。这是RocketMQ特有的一个功能。 那会延迟多久呢？延迟时间的设置就是在Message消息对象上设置一个延迟级别 message.setDelayTimeLevel(3);开源版本的RocketMQ中，对延迟消息并不支持任意时间的延迟设定(商业版本中支持)，而是只支 持18个固定的延迟级别，1到18分别对应1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h。

@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringRocketTest {
    @Resource
    private RocketMQTemplate rocketMQTemplate;
	@Test
    public void sendMessageTest(){
      for (int i = 0; i < 100; i++) {
            // 参数一：topi，参数二：消息，参数三：超时时间，参数四：延迟级别
            rocketMQTemplate.syncSend("ScheduledTopic",MessageBuilder.withPayload("Hello"+i).build(),2000,3);
        }
    }
}

1.8 事务消息

事务消息主要和消息的生产者有关：

@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringRocketTest {
    @Resource
    private RocketMQTemplate rocketMQTemplate;
	@Test
    public void sendMessageTest(){
        String[] tags = new String[] {"TagA", "TagB", "TagC", "TagD", "TagE"};
        for (int i = 0; i < 10; i++) {
            String tag =  tags[i % tags.length];
            rocketMQTemplate.sendMessageInTransaction("TransactionTopic:"+tag,MessageBuilder.withPayload(tag + " Hello RocketMQ " + i).build(),null);
        }
    }
}

TransactionListener事务监听器：

@RocketMQTransactionListener
public class TransactionMsgListener implements RocketMQLocalTransactionListener {
    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object o) {
        String tag = (String) message.getHeaders().get("rocketmq_TAGS");
        if(StringUtils.contains(tag,"TagA")){
            return RocketMQLocalTransactionState.COMMIT; // 返回COMMIT状态的消息会立即被消费者消费到
        }else if(StringUtils.contains(tag,"TagB")){
            return RocketMQLocalTransactionState.ROLLBACK; // 返回ROLLBACK状态的消息会被丢弃。
        }else{
            return RocketMQLocalTransactionState.UNKNOWN; // 返回UNKNOWN状态的消息会由Broker过一段时间再来回查事务的状态。
        }
    }

    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message message) {
        String tag = (String) message.getHeaders().get("rocketmq_TAGS");
        if(StringUtils.contains(tag,"TagC")){
            return RocketMQLocalTransactionState.COMMIT; // 返回COMMIT状态的消息会立即被消费者消费到
        }else if(StringUtils.contains(tag,"TagD")){
            return RocketMQLocalTransactionState.ROLLBACK; // 返回ROLLBACK状态的消息会被丢弃。
        }else{
            return RocketMQLocalTransactionState.UNKNOWN; // 返回UNKNOWN状态的消息会由Broker过一段时间再来回查事务的状态。
        }
    }
}

消息的消费者：

@Component
@RocketMQMessageListener(consumerGroup = "TransactionConsumerGroup", topic = "TransactionTopic",consumeMode= ConsumeMode.CONCURRENTLY,messageModel = MessageModel.BROADCASTING)
public class SpringConsumer implements RocketMQListener<String> {
    @Override
    public void onMessage(String message) {
        System.out.println("收到消息内容 "+ message);
    }
}

控制台输出：

收到消息内容 TagA Hello RocketMQ 0
收到消息内容 TagA Hello RocketMQ 5
收到消息内容 TagC Hello RocketMQ 2
收到消息内容 TagC Hello RocketMQ 7

会先输出TagA的内容，过一会再出输出TagC的内容，符合预期场景。

最近更新
贡献者: zyh1020
豫ICP备2022026603号
Copyright © 2026 Ms.Zyh