RocketMQ 消息队列

public class ProducerExample {
    public static void main(String[] args) throws Exception {
        // 创建生产者
        DefaultMQProducer producer = new DefaultMQProducer("producer_group");
        producer.setNamesrvAddr("localhost:9876");
        producer.start();

        // 发送消息
        for (int i = 0; i < 10; i++) {
            Message msg = new Message("TopicTest", "TagA",
                ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
            SendResult sendResult = producer.send(msg);
            System.out.printf("%s%n", sendResult);
        }

        producer.shutdown();
    }
}

public class ConsumerExample {
    public static void main(String[] args) throws Exception {
        // 创建消费者
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer_group");
        consumer.setNamesrvAddr("localhost:9876");
        consumer.subscribe("TopicTest", "*");

        // 注册消息监听器
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(
                List<MessageExt> messages,
                ConsumeConcurrentlyContext context) {
                for (MessageExt message : messages) {
                    System.out.printf("Receive message: %s%n",
                        new String(message.getBody()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        consumer.start();
        System.out.printf("Consumer Started.%n");
    }
}

public class SyncProducer {
    public void sendSyncMessage() throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("sync_producer_group");
        producer.setNamesrvAddr("localhost:9876");
        producer.start();

        Message msg = new Message("TopicTest", "TagA",
            "Hello RocketMQ".getBytes(RemotingHelper.DEFAULT_CHARSET));

        // 同步发送，等待发送结果
        SendResult sendResult = producer.send(msg);
        System.out.printf("Send result: %s%n", sendResult);

        producer.shutdown();
    }
}

public class AsyncProducer {
    public void sendAsyncMessage() throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("async_producer_group");
        producer.setNamesrvAddr("localhost:9876");
        producer.start();

        Message msg = new Message("TopicTest", "TagA",
            "Hello RocketMQ".getBytes(RemotingHelper.DEFAULT_CHARSET));

        // 异步发送，不等待发送结果
        producer.send(msg, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.printf("Send success: %s%n", sendResult);
            }

            @Override
            public void onException(Throwable e) {
                System.out.printf("Send failed: %s%n", e.getMessage());
            }
        });

        producer.shutdown();
    }
}

public class OnewayProducer {
    public void sendOnewayMessage() throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("oneway_producer_group");
        producer.setNamesrvAddr("localhost:9876");
        producer.start();

        Message msg = new Message("TopicTest", "TagA",
            "Hello RocketMQ".getBytes(RemotingHelper.DEFAULT_CHARSET));

        // 单向发送，不关心发送结果
        producer.sendOneway(msg);

        producer.shutdown();
    }
}

public class ClusterConsumer {
    public void consumeMessage() throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("cluster_consumer_group");
        consumer.setNamesrvAddr("localhost:9876");
        consumer.subscribe("TopicTest", "*");

        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(
                List<MessageExt> messages,
                ConsumeConcurrentlyContext context) {
                // 集群消费：每条消息只会被一个消费者消费
                for (MessageExt message : messages) {
                    System.out.printf("Receive message: %s%n",
                        new String(message.getBody()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        consumer.start();
    }
}

public class BroadcastConsumer {
    public void consumeMessage() throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("broadcast_consumer_group");
        consumer.setNamesrvAddr("localhost:9876");
        consumer.setMessageModel(MessageModel.BROADCASTING); // 设置为广播模式
        consumer.subscribe("TopicTest", "*");

        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(
                List<MessageExt> messages,
                ConsumeConcurrentlyContext context) {
                // 广播消费：每条消息会被所有消费者消费
                for (MessageExt message : messages) {
                    System.out.printf("Receive message: %s%n",
                        new String(message.getBody()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        consumer.start();
    }
}

public class OrderedProducer {
    public void sendOrderedMessage() throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("ordered_producer_group");
        producer.setNamesrvAddr("localhost:9876");
        producer.start();

        // 全局顺序消息：所有消息按发送顺序消费
        for (int i = 0; i < 10; i++) {
            Message msg = new Message("OrderedTopic", "TagA",
                ("Ordered message " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));

            // 发送到同一个队列保证顺序
            SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
                @Override
                public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                    // 选择第一个队列
                    return mqs.get(0);
                }
            }, null);

            System.out.printf("Send result: %s%n", sendResult);
        }

        producer.shutdown();
    }
}

public class PartitionOrderedProducer {
    public void sendPartitionOrderedMessage() throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("partition_ordered_producer_group");
        producer.setNamesrvAddr("localhost:9876");
        producer.start();

        // 分区顺序消息：同一分区的消息按顺序消费
        for (int i = 0; i < 10; i++) {
            Message msg = new Message("PartitionOrderedTopic", "TagA",
                ("Partition ordered message " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));

            // 根据业务ID选择队列
            String orderId = "ORDER_" + (i % 3); // 3个分区
            SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
                @Override
                public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                    String orderId = (String) arg;
                    int index = Math.abs(orderId.hashCode()) % mqs.size();
                    return mqs.get(index);
                }
            }, orderId);

            System.out.printf("Send result: %s%n", sendResult);
        }

        producer.shutdown();
    }
}

public class TransactionProducer {
    public void sendTransactionMessage() throws Exception {
        TransactionMQProducer producer = new TransactionMQProducer("transaction_producer_group");
        producer.setNamesrvAddr("localhost:9876");

        // 设置事务监听器
        producer.setTransactionListener(new TransactionListener() {
            @Override
            public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
                // 执行本地事务
                try {
                    // 模拟本地事务处理
                    System.out.println("执行本地事务: " + new String(msg.getBody()));
                    // 返回提交状态
                    return LocalTransactionState.COMMIT_MESSAGE;
                } catch (Exception e) {
                    // 返回回滚状态
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                }
            }

            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt msg) {
                // 检查本地事务状态
                return LocalTransactionState.COMMIT_MESSAGE;
            }
        });

        producer.start();

        // 发送事务消息
        Message msg = new Message("TransactionTopic", "TagA",
            "Transaction message".getBytes(RemotingHelper.DEFAULT_CHARSET));
        SendResult sendResult = producer.sendMessageInTransaction(msg, null);
        System.out.printf("Transaction send result: %s%n", sendResult);

        producer.shutdown();
    }
}

public class TagFilterConsumer {
    public void consumeWithTagFilter() throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("tag_filter_consumer_group");
        consumer.setNamesrvAddr("localhost:9876");

        // 只消费 TagA 和 TagB 的消息
        consumer.subscribe("FilterTopic", "TagA || TagB");

        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(
                List<MessageExt> messages,
                ConsumeConcurrentlyContext context) {
                for (MessageExt message : messages) {
                    System.out.printf("Receive message: %s, Tag: %s%n",
                        new String(message.getBody()), message.getTags());
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        consumer.start();
    }
}

public class SQLFilterConsumer {
    public void consumeWithSQLFilter() throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("sql_filter_consumer_group");
        consumer.setNamesrvAddr("localhost:9876");

        // 使用 SQL92 语法过滤消息
        consumer.subscribe("SQLFilterTopic",
            MessageSelector.bySql("a > 5 AND b = 'test'"));

        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(
                List<MessageExt> messages,
                ConsumeConcurrentlyContext context) {
                for (MessageExt message : messages) {
                    System.out.printf("Receive message: %s%n",
                        new String(message.getBody()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        consumer.start();
    }
}

public class MessageBacktrackConsumer {
    public void consumeFromTimestamp() throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("backtrack_consumer_group");
        consumer.setNamesrvAddr("localhost:9876");
        consumer.subscribe("BacktrackTopic", "*");

        // 设置消费起始时间（回溯到指定时间）
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_TIMESTAMP);
        consumer.setConsumeTimestamp("20231201080000"); // 格式：yyyyMMddHHmmss

        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(
                List<MessageExt> messages,
                ConsumeConcurrentlyContext context) {
                for (MessageExt message : messages) {
                    System.out.printf("Receive message: %s, StoreTime: %s%n",
                        new String(message.getBody()),
                        new Date(message.getStoreTimestamp()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        consumer.start();
    }
}

public class OptimizedProducer {
    public void optimizedSend() throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("optimized_producer_group");
        producer.setNamesrvAddr("localhost:9876");

        // 性能优化配置
        producer.setCompressMsgBodyOverHowmuch(4096); // 消息压缩阈值
        producer.setMaxMessageSize(4 * 1024 * 1024); // 最大消息大小
        producer.setSendMsgTimeout(3000); // 发送超时时间
        producer.setRetryTimesWhenSendFailed(2); // 发送失败重试次数
        producer.setRetryTimesWhenSendAsyncFailed(2); // 异步发送失败重试次数

        producer.start();

        // 批量发送消息
        List<Message> messages = new ArrayList<>();
        for (int i = 0; i < 100; i++) {
            Message msg = new Message("BatchTopic", "TagA",
                ("Batch message " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
            messages.add(msg);
        }

        // 批量发送
        SendResult sendResult = producer.send(messages);
        System.out.printf("Batch send result: %s%n", sendResult);

        producer.shutdown();
    }
}

public class OptimizedConsumer {
    public void optimizedConsume() throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("optimized_consumer_group");
        consumer.setNamesrvAddr("localhost:9876");
        consumer.subscribe("OptimizedTopic", "*");

        // 性能优化配置
        consumer.setConsumeThreadMin(20); // 最小消费线程数
        consumer.setConsumeThreadMax(64); // 最大消费线程数
        consumer.setConsumeMessageBatchMaxSize(32); // 批量消费消息数量
        consumer.setPullBatchSize(32); // 批量拉取消息数量
        consumer.setPullInterval(0); // 拉取间隔，0表示不间隔

        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(
                List<MessageExt> messages,
                ConsumeConcurrentlyContext context) {
                // 批量处理消息
                for (MessageExt message : messages) {
                    System.out.printf("Receive message: %s%n",
                        new String(message.getBody()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        consumer.start();
    }
}

public class MonitoringConfig {
    public void setupMonitoring() {
        // 设置监控日志
        System.setProperty("rocketmq.client.logUseSlf4j", "true");
        System.setProperty("rocketmq.client.logLevel", "INFO");
        System.setProperty("rocketmq.client.logRoot", "/opt/logs/rocketmq");

        // 设置统计开关
        System.setProperty("rocketmq.client.statistics", "true");
    }
}

public class FailureHandlingConsumer {
    public void handleConsumeFailure() throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("failure_handling_consumer_group");
        consumer.setNamesrvAddr("localhost:9876");
        consumer.subscribe("FailureTopic", "*");

        // 设置重试次数
        consumer.setMaxReconsumeTimes(3);

        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(
                List<MessageExt> messages,
                ConsumeConcurrentlyContext context) {
                for (MessageExt message : messages) {
                    try {
                        // 业务处理逻辑
                        processMessage(message);
                        System.out.printf("Process message success: %s%n",
                            new String(message.getBody()));
                    } catch (Exception e) {
                        System.err.printf("Process message failed: %s, error: %s%n",
                            new String(message.getBody()), e.getMessage());

                        // 根据重试次数决定是否继续重试
                        if (message.getReconsumeTimes() >= 3) {
                            // 超过重试次数，记录到死信队列或人工处理
                            handleDeadLetterMessage(message);
                            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                        } else {
                            // 返回重试状态
                            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                        }
                    }
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        consumer.start();
    }

    private void processMessage(MessageExt message) throws Exception {
        // 模拟业务处理
        if (Math.random() < 0.3) {
            throw new RuntimeException("模拟处理失败");
        }
    }

    private void handleDeadLetterMessage(MessageExt message) {
        // 处理死信消息
        System.out.printf("Handle dead letter message: %s%n",
            new String(message.getBody()));
    }
}

// 1. 检查消息存储时间
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);

// 2. 增加消息存储时间
// 在 broker.conf 中配置
fileReservedTime=72  // 消息保留72小时

// 3. 使用同步发送确保消息到达
SendResult result = producer.send(msg);
if (result.getSendStatus() != SendStatus.SEND_OK) {
    // 发送失败处理
}

// 1. 实现幂等性处理
public class IdempotentConsumer {
    private Set<String> processedMessages = new ConcurrentHashMap<>();

    public ConsumeConcurrentlyStatus consumeMessage(
            List<MessageExt> messages,
            ConsumeConcurrentlyContext context) {
        for (MessageExt message : messages) {
            String messageId = message.getMsgId();

            // 检查消息是否已处理
            if (processedMessages.contains(messageId)) {
                System.out.println("Message already processed: " + messageId);
                continue;
            }

            // 处理消息
            processMessage(message);

            // 记录已处理的消息
            processedMessages.add(messageId);
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
}

// 1. 增加消费者实例
// 2. 优化消费逻辑
// 3. 调整消费参数
consumer.setConsumeThreadMin(50);
consumer.setConsumeThreadMax(100);
consumer.setConsumeMessageBatchMaxSize(64);

// 4. 使用批量消费
consumer.registerMessageListener(new MessageListenerConcurrently() {
    @Override
    public ConsumeConcurrentlyStatus consumeMessage(
            List<MessageExt> messages,
            ConsumeConcurrentlyContext context) {
        // 批量处理消息
        List<CompletableFuture<Void>> futures = new ArrayList<>();

        for (MessageExt message : messages) {
            CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
                processMessage(message);
            });
            futures.add(future);
        }

        // 等待所有消息处理完成
        CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();

        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
});