<?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
# 默認(rèn)的生產(chǎn)者組
rocketmq.producer.group=batched_group
# 發(fā)送同步消息超時時間
rocketmq.producer.send-message-timeout=3000
# 用于設(shè)置在消息發(fā)送失敗后，生產(chǎn)者是否嘗試切換到下一個服務(wù)器。設(shè)置為 true 表示啟用，在發(fā)送失敗時嘗試切換到下一個服務(wù)器
rocketmq.producer.retry-next-server=true
# 用于指定消息發(fā)送失敗時的重試次數(shù)
rocketmq.producer.retry-times-when-send-failed=3
# 設(shè)置消息壓縮的閾值,為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)
                    // 設(shè)置消息類型
                    .setHeader(MessageHeaders.CONTENT_TYPE, "text/plain")
                    .build();
            // 加入到列表
            messageList.add(rocketMessage);
        }
        // 使用syncSend發(fā)送批量消息
        SendResult sendResult = rocketMQTemplate.syncSend(topic, messageList);
        log.info("同步批量發(fā)送普通消息結(jié)果：{}",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)
                    // 設(shè)置消息類型
                    .setHeader(MessageHeaders.CONTENT_TYPE, "text/plain")
                    .build();
            // 加入到列表
            messageList.add(rocketMessage);
        }
        // 使用asyncSend發(fā)送批量消息
        rocketMQTemplate.asyncSend(topic, messageList, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                // 異步發(fā)送成功的回調(diào)邏輯
                log.info("異步批量發(fā)送普通消息成功: " + sendResult);
            }

            @Override
            public void onException(Throwable e) {
                // 異步發(fā)送失敗的回調(diào)邏輯
                log.info("異步批量發(fā)送普通消息失敗: " + e.getMessage());
            }
        });
    }

    @Test
    public void syncSendBatchOrderlyStringMessagesWithBuilder() {
        String topic = "ordered_string_message_topic";
        String message = "同步批量發(fā)送順序消息，超級喜歡Go語言：";
        List<Message<String>> messageList = new ArrayList<>();
        for (int i = 0; i < 10; i++) {
            Message<String> rocketMessage = MessageBuilder.withPayload(message + i)
                    // 設(shè)置消息類型
                    .setHeader(MessageHeaders.CONTENT_TYPE, "text/plain")
                    .build();
            // 加入到列表
            messageList.add(rocketMessage);
        }
        // 使用syncSendOrderly發(fā)送批量順序消息，消費者線程設(shè)置為1
        SendResult sendResult = rocketMQTemplate.syncSendOrderly(topic, messageList, "alian_sync_ordered");
        log.info("批量發(fā)送順序消息發(fā)送結(jié)果：{}",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);
        // 處理消息的業(yè)務(wù)邏輯
    }
}

@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);
        // 處理消息的業(yè)務(wù)邏輯
    }
}

	@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 發(fā)送批量消息
        rocketMQTemplate.asyncSendOrderly(topic, messageList, "alian_async_ordered", new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                // 異步發(fā)送成功的回調(diào)邏輯
                log.info("異步消息發(fā)送字符串消息成功: " + sendResult);
            }

            @Override
            public void onException(Throwable e) {
                // 異步發(fā)送失敗的回調(diào)邏輯
                log.info("異步消息發(fā)送字符串消息失敗: " + 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"]