SpringBoot整合RocketMQ批量发送消息的实现代码

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <parent>
        <artifactId>rocketmq</artifactId>
        <groupId>com.alian</groupId>
        <version>1.0.0-SNAPSHOT</version>
    </parent>
    <modelVersion>4.0.0</modelVersion>

    <artifactId>06-send-batched-message</artifactId>

    <properties>
        <maven.compiler.source>8</maven.compiler.source>
        <maven.compiler.target>8</maven.compiler.target>
    </properties>

    <dependencies>
        <dependency>
            <groupId>com.alian</groupId>
            <artifactId>common-rocketmq-dto</artifactId>
            <version>1.0.0-SNAPSHOT</version>
        </dependency>
    </dependencies>

</project>

server.port=8005

# rocketmq地址
rocketmq.name-server=192.168.0.234:9876
# 默认的生产者组
rocketmq.producer.group=batched_group
# 发送同步消息超时时间
rocketmq.producer.send-message-timeout=3000
# 用于设置在消息发送失败后，生产者是否尝试切换到下一个服务器。设置为 true 表示启用，在发送失败时尝试切换到下一个服务器
rocketmq.producer.retry-next-server=true
# 用于指定消息发送失败时的重试次数
rocketmq.producer.retry-times-when-send-failed=3
# 设置消息压缩的阈值,为0表示禁用消息体的压缩
rocketmq.producer.compress-message-body-threshold=0

	@Autowired
    private RocketMQTemplate rocketMQTemplate;

    @Test
    public void syncSendBatchStringMessagesWithBuilder() {
        String topic = "string_message_topic";
        String message = "超级喜欢Golang语言：";
        List<Message<String>> messageList = new ArrayList<>();
        for (int i = 0; i < 10; i++) {
            Message<String> rocketMessage = MessageBuilder.withPayload(message + i)
                    // 设置消息类型
                    .setHeader(MessageHeaders.CONTENT_TYPE, "text/plain")
                    .build();
            // 加入到列表
            messageList.add(rocketMessage);
        }
        // 使用syncSend发送批量消息
        SendResult sendResult = rocketMQTemplate.syncSend(topic, messageList);
        log.info("同步批量发送普通消息结果：{}",sendResult);
    }

    @Test
    public void asyncSendBatchStringMessageWithBuilder() {
        String topic = "string_message_topic";
        String message = "Alian超级喜欢Golang语言：";
        List<Message<String>> messageList = new ArrayList<>();
        for (int i = 0; i < 10; i++) {
            Message<String> rocketMessage = MessageBuilder.withPayload(message + i)
                    // 设置消息类型
                    .setHeader(MessageHeaders.CONTENT_TYPE, "text/plain")
                    .build();
            // 加入到列表
            messageList.add(rocketMessage);
        }
        // 使用asyncSend发送批量消息
        rocketMQTemplate.asyncSend(topic, messageList, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                // 异步发送成功的回调逻辑
                log.info("异步批量发送普通消息成功: " + sendResult);
            }

            @Override
            public void onException(Throwable e) {
                // 异步发送失败的回调逻辑
                log.info("异步批量发送普通消息失败: " + e.getMessage());
            }
        });
    }

    @Test
    public void syncSendBatchOrderlyStringMessagesWithBuilder() {
        String topic = "ordered_string_message_topic";
        String message = "同步批量发送顺序消息，超级喜欢Go语言：";
        List<Message<String>> messageList = new ArrayList<>();
        for (int i = 0; i < 10; i++) {
            Message<String> rocketMessage = MessageBuilder.withPayload(message + i)
                    // 设置消息类型
                    .setHeader(MessageHeaders.CONTENT_TYPE, "text/plain")
                    .build();
            // 加入到列表
            messageList.add(rocketMessage);
        }
        // 使用syncSendOrderly发送批量顺序消息，消费者线程设置为1
        SendResult sendResult = rocketMQTemplate.syncSendOrderly(topic, messageList, "alian_sync_ordered");
        log.info("批量发送顺序消息发送结果：{}",sendResult);
    }

@Slf4j
@Service
@RocketMQMessageListener(topic = "ordered_string_message_topic", 
						consumerGroup = "ORDERED_GROUP_STRING", 
						consumeMode = ConsumeMode.ORDERLY)
public class StringMessageConsumer implements RocketMQListener<String> {

    @Override
    public void onMessage(String message) {
        log.info("字符串消费者接收到的消息: {}", message);
        // 处理消息的业务逻辑
    }
}

@RocketMQMessageListener(topic = "ordered_string_message_topic", 
						consumerGroup = "CONCURRENT_GROUP_STRING",
						consumeThreadNumber = 1)
public class StringMessageConsumer implements RocketMQListener<String> {

    @Override
    public void onMessage(String message) {
        log.info("字符串消费者接收到的消息: {}", message);
        // 处理消息的业务逻辑
    }
}

	@Test
    public void asyncSendBatchOrderlyStringMessageWithBuilder2() {
        String topic = "ordered_string_message_topic";
        String message = "Alian超级喜欢Golang语言：";
        List<String> messageList = new ArrayList<>();
        for (int i = 0; i < 10; i++) {
            // 加入到列表
            messageList.add(message + i);
        }
        // 使用 asyncSendOrderly 发送批量消息
        rocketMQTemplate.asyncSendOrderly(topic, messageList, "alian_async_ordered", new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                // 异步发送成功的回调逻辑
                log.info("异步消息发送字符串消息成功: " + sendResult);
            }

            @Override
            public void onException(Throwable e) {
                // 异步发送失败的回调逻辑
                log.info("异步消息发送字符串消息失败: " + e.getMessage());
            }
        });
    }

[GROUP_STRING_18] com.alian.ordered.StringMessageConsumer  : 字符串消费者接收到的消息: ["Alian超级喜欢Golang语言：0","Alian超级喜欢Golang语言：1","Alian超级喜欢Golang语言：2","Alian超级喜欢Golang语言：3","Alian超级喜欢Golang语言：4","Alian超级喜欢Golang语言：5","Alian超级喜欢Golang语言：6","Alian超级喜欢Golang语言：7","Alian超级喜欢Golang语言：8","Alian超级喜欢Golang语言：9"]