在一些业务场景下，我们希望使用 Producer 批量发送消息，提高发送性能。在 RocketMQTemplate 中，提供了一个方法方法批量发送消息的方法。代码如下：

// RocketMQTemplate.java

public <T extends Message> SendResult syncSend(String destination, Collection<T> messages, long timeout) {
    // ... 省略具体代码实现
}

通过方法参数 destination 可知，必须发送相同 Topic 的消息。
要注意方法参数 messages ，每个集合的元素必须是 Spring Messaging 定义的 Message 消息。RocketMQTemplate 重载了非常多的 #syncSend(...) 方法，一定要小心哟。
通过方法名可知，这个是同步批量发送消息。

有一点要注意，虽然是批量发送多条消息，但是是以所有消息加起来的大小，不能超过消息的最大大小的限制，而不是按照单条计算。所以，一次性发送的消息特别多，还是需要分批的进行批量发送。

一、快速开始

1.1 Demo02Message

创建 [Demo02Message]消息类，提供给当前示例使用。代码如下：

package com.ebadagang.springboot.rocketmq.message;

/**
 * 示例 02 的 Message 消息
 */
public class Demo02Message {

    public static final String TOPIC = "DEMO_02";

    /**
     * 编号
     */
    private Integer id;

    public Demo02Message setId(Integer id) {
        this.id = id;
        return this;
    }

    public Integer getId() {
        return id;
    }

    @Override
    public String toString() {
        return "Demo02Message{" +
                "id=" + id +
                '}';
    }

}

1.2 Demo02Producer

它会使用 RocketMQTemplate 实现批量发送消息。代码如下：

package com.ebadagang.springboot.rocketmq.producer;

import com.ebadagang.springboot.rocketmq.message.Demo02Message;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;

@Component
public class Demo02Producer {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    public SendResult sendBatch(Collection<Integer> ids) {
        // <1>  创建多条 Demo02Message 消息
        List<Message> messages = new ArrayList<>(ids.size());
        for (Integer id : ids) {
            // 创建 Demo02Message 消息
            Demo02Message message = new Demo02Message().setId(id);
            // 构建 Spring Messaging 定义的 Message 消息
            messages.add(MessageBuilder.withPayload(message).build());
        }
        // 同步批量发送消息
        return rocketMQTemplate.syncSend(Demo02Message.TOPIC, messages, 30 * 1000L);
    }

}

注意，在 <1> 处，我们就创建了 Spring Messaging 定义的 Message 消息的数组，用于下面使用 RocketMQTemplate 批量发送消息。

1.3 Demo02Consumer

package com.ebadagang.springboot.rocketmq.consumer;

import com.ebadagang.springboot.rocketmq.message.Demo02Message;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

@Component
//@RocketMQMessageListener(
//        topic = Demo02Message.TOPIC,
//        consumerGroup = "demo02-consumer-group-" + Demo02Message.TOPIC
//)
public class Demo02Consumer implements RocketMQListener<Demo02Message> {

    private Logger logger = LoggerFactory.getLogger(getClass());

    @Override
    public void onMessage(Demo02Message message) {
        logger.info("[onMessage][线程编号:{} 消息内容：{}]", Thread.currentThread().getId(), message);
    }

}

1.4 简单测试

创建 [Demo02ProducerTest]测试类，编写一个单元测试方法，调用 Demo02Producer 批量发送消息。代码如下：

package com.ebadagang.springboot.rocketmq.producer;

import com.ebadagang.springboot.rocketmq.Application;
import org.apache.rocketmq.client.producer.SendResult;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

import java.util.Arrays;
import java.util.List;
import java.util.concurrent.CountDownLatch;

@RunWith(SpringRunner.class)
@SpringBootTest(classes = Application.class)
public class Demo02ProducerTest {

    private Logger logger = LoggerFactory.getLogger(getClass());

    @Autowired
    private Demo02Producer producer;

    @Test
    public void testSendBatch() throws InterruptedException {
        List<Integer> ids = Arrays.asList(1, 2, 3, 4, 5);
        SendResult result = producer.sendBatch(ids);
        logger.info("[testSendBatch][发送编号：[{}] 发送结果：[{}]]", ids, result);

        // 阻塞等待，保证消费
        new CountDownLatch(1).await();
    }

}

我们来执行 #testSendBatch() 方法，测试批量发送消息。控制台输出如下：

# Producer 批量发送5条消息成功
2020-08-04 17:10:35.808  INFO 10116 --- [           main] c.e.s.r.producer.Demo02ProducerTest      : [testSendBatch][发送编号：[[1, 2, 3, 4, 5]] 发送结果：[SendResult [sendStatus=SEND_OK, msgId=240884E30114A66731EF8A0EAAFD768A278418B4AAC213229D910002,240884E30114A66731EF8A0EAAFD768A278418B4AAC213229D910006,240884E30114A66731EF8A0EAAFD768A278418B4AAC213229D910007,240884E30114A66731EF8A0EAAFD768A278418B4AAC213229D910008,240884E30114A66731EF8A0EAAFD768A278418B4AAC213229D910009, offsetMsgId=6585E30D00002A9F0000000000033A18,6585E30D00002A9F0000000000033B2C,6585E30D00002A9F0000000000033C40,6585E30D00002A9F0000000000033D54,6585E30D00002A9F0000000000033E68, messageQueue=MessageQueue [topic=DEMO_02, brokerName=broker-a, queueId=3], queueOffset=5]]]

# 5条消息，被逐条消费
2020-08-04 17:10:35.808  INFO 10116 --- [MessageThread_8] c.e.s.rocketmq.consumer.Demo02Consumer   : [onMessage][线程编号:178 消息内容：Demo02Message{id=3}]
2020-08-04 17:10:35.808  INFO 10116 --- [MessageThread_6] c.e.s.rocketmq.consumer.Demo02Consumer   : [onMessage][线程编号:176 消息内容：Demo02Message{id=1}]
2020-08-04 17:10:35.808  INFO 10116 --- [MessageThread_7] c.e.s.rocketmq.consumer.Demo02Consumer   : [onMessage][线程编号:177 消息内容：Demo02Message{id=2}]
2020-08-04 17:10:35.823  INFO 10116 --- [essageThread_10] c.e.s.rocketmq.consumer.Demo02Consumer   : [onMessage][线程编号:180 消息内容：Demo02Message{id=5}]
2020-08-04 17:10:35.823  INFO 10116 --- [MessageThread_9] c.e.s.rocketmq.consumer.Demo02Consumer   : [onMessage][线程编号:179 消息内容：Demo02Message{id=4}]

可以看到5条消息，被 Demo02Consumer 并发消费完成。

底线

本文源代码使用 Apache License 2.0开源许可协议，这里是本文源码Gitee地址，可通过命令git clone+地址下载代码到本地，也可直接点击链接通过浏览器方式查看源代码。

RocketMQ 与 Spring Boot整合(二、批量发送消息）