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"]