RocketMQ发送和接收方式详解

RocketMQ有几种发送方式

RocketMQ 提供了几种不同的消息发送方式，以满足不同场景下的需求。这些发送方式主要包括：同步发送（Synchronous）：这是最常见的一种发送方式，客户端发送消息后，会等待服务器端的响应。只有当消息完全被服务器接收，客户端才会继续执行。这种方式适用于对可靠性要求较高的场景。异步发送（Asynchronous）：客户端在发送消息后不会立即等待服务器的响应，而是通过一个回调接口来处理服务器的响应，这样可以提高消息发送的吞吐量。这种方式适用于对响应时间要求较敏感的场景。单向发送（One-way）：这种方式是最轻量级的一种发送方式，客户端只负责发送消息，不等待服务器的响应，也不关心消息是否被服务器接收。这种方式的可靠性最低，但吞吐量最高，适用于对可靠性要求不高的日志收集等场景。顺序发送（Ordered）：RocketMQ 还支持顺序发送消息，确保特定主题的消息能按照发送的顺序被消费。这对于需要保证消息顺序的业务场景（如订单处理）非常重要。**顺序必须是同步的**这些发送方式可以根据具体的业务需求和场景来选择使用，以达到最佳的性能和可靠性平衡。

不同发送方式的调用方式或者不同配置

在使用 RocketMQ 进行消息发送时，可以根据不同的发送需求选择合适的发送方式。以下是不同发送方式的基本调用方式或配置示例：

1. 同步发送（Synchronous）
同步发送是指消息发送方将消息发送给服务器后，会等待服务器的响应，确认消息已经被接收后才继续执行后续操作。

// 实例化消息生产者Producer
DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName");
// 设置NameServer地址
producer.setNamesrvAddr("nameserver1:9876");
// 启动Producer实例
producer.start();// 创建消息实例，指定topic，tag和消息体
Message msg = new Message("TopicTest", "TagA", ("Hello RocketMQ").getBytes(RemotingHelper.DEFAULT_CHARSET));// 调用send方法发送消息，并获取发送结果
SendResult sendResult = producer.send(msg);// 打印发送结果
System.out.printf("%s%n", sendResult);// 当不再发送消息时，关闭Producer实例
producer.shutdown();

2. 异步发送（Asynchronous）
异步发送是指消息发送方发送消息后，不会立即等待服务器的响应，而是提供一个回调接口，服务器响应时通过回调接口来通知发送方。

// 实例化消息生产者Producer
DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName");
// 设置NameServer地址
producer.setNamesrvAddr("nameserver1:9876");
// 启动Producer实例
producer.start();// 创建消息实例，指定topic，tag和消息体
Message msg = new Message("TopicTest", "TagA", ("Hello RocketMQ").getBytes(RemotingHelper.DEFAULT_CHARSET));// 异步发送消息
producer.send(msg, new SendCallback() {@Overridepublic void onSuccess(SendResult sendResult) {// 消息发送成功的回调System.out.printf("%-10d OK %s %n", index, sendResult.getMsgId());}@Overridepublic void onException(Throwable e) {// 消息发送失败的回调System.out.printf("%-10d Exception %s %n", index, e);e.printStackTrace();}
});// 关闭Producer实例
producer.shutdown();

3. 单向发送（One-way）
单向发送是指消息发送方只负责发送消息，不等待服务器的响应也不关心消息是否被服务器接收，适用于对可靠性要求不高的场景。

// 实例化消息生产者Producer
DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName");
// 设置NameServer地址
producer.setNamesrvAddr("nameserver1:9876");
// 启动Producer实例
producer.start();// 创建消息实例，指定topic，tag和消息体
Message msg = new Message("TopicTest", "TagA", ("Hello RocketMQ").getBytes(RemotingHelper.DEFAULT_CHARSET));// 单向发送消息，没有返回结果
producer.sendOneway(msg);// 关闭Producer实例
producer.shutdown();

4. 顺序发送（Ordered）
顺序发送保证了特定主题的消息能按照发送的顺序来消费。

// 实例化消息生产者Producer
DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName");
// 设置NameServer地址
producer.setNamesrvAddr("nameserver1:9876");
// 启动Producer实例
producer.start();String[] tags = new String[]{"TagA", "TagB", "TagC"};
// 订单列表
List<OrderStep> orderList = new ArrayList<OrderStep>();
orderList.add(new OrderStep(15103111039L, "创建"));
orderList.add(new OrderStep(15103111065L, "支付"));
orderList.add(new OrderStep(15103111039L, "完成"));for (int i = 0; i < 10; i++) {// 加入订单的时间戳来模拟简单的订单IDString body = orderList.get(i % orderList.size()) + " 时间：" + System.currentTimeMillis();Message msg = new Message("TopicTest", tags[i % tags.length], "KEY" + i, body.getBytes());SendResult sendResult = producer.send(msg, new MessageQueueSelector() {@Overridepublic MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {Long id = (Long) arg;  // 根据订单id选择发送queuelong index = id % mqs.size();return mqs.get((int) index);}}, orderList.get(i % orderList.size()).getOrderId());  // 订单idSystem.out.println(String.format("SendResult status:%s, queueId:%d, body:%s",sendResult.getSendStatus(),sendResult.getMessageQueue().getQueueId(),body));
}// 关闭Producer实例
producer.shutdown();

每种发送方式都有其适用的场景，可以根据实际的业务需求选择使用。在实际开发中，需要根据具体的业务逻辑对示例代码进行适当的调整和优化。

不同接收方式的调用方式或者不同配置

1. 集群消费（Clustering）
集群消费模式下，同一个消费者组中的多个消费者实例共同消费主题下的消息，每条消息只会被消费一次。

// 实例化消费者
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("ConsumerGroupName");
// 设置NameServer地址
consumer.setNamesrvAddr("nameserver1:9876");
// 订阅一个或多个Topic，以及Tag来过滤需要消费的消息
consumer.subscribe("TopicTest", "*");
// 注册回调实现类来处理从broker拉取回来的消息
consumer.registerMessageListener(new MessageListenerConcurrently() {@Overridepublic ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {for (MessageExt msg : msgs) {// 消费消息System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), new String(msg.getBody()));}// 标记该消息已经被成功消费return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}
});
// 启动消费者实例
consumer.start();

2. 广播消费（Broadcasting）
广播消费模式下，消息会被消费者组中的每个消费者都消费一次。

// 实例化消费者，并设置消费模式为广播模式
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("ConsumerGroupName");
consumer.setMessageModel(MessageModel.BROADCASTING);
consumer.setNamesrvAddr("nameserver1:9876");
consumer.subscribe("TopicTest", "*");
consumer.registerMessageListener(new MessageListenerConcurrently() {@Overridepublic ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {for (MessageExt msg : msgs) {System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), new String(msg.getBody()));}return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}
});
consumer.start();

3. 顺序消费（Orderly）
顺序消费保证了特定主题的消息能按照发送的顺序来消费。

DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("ConsumerGroupName");
consumer.setNamesrvAddr("nameserver1:9876");
consumer.subscribe("TopicTest", "*");
consumer.registerMessageListener(new MessageListenerOrderly() {@Overridepublic ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {for (MessageExt msg : msgs) {System.out.printf("%s Receive New Messages: %s, QueueId: %d %n", Thread.currentThread().getName(), new String(msg.getBody()), msg.getQueueId());}return ConsumeOrderlyStatus.SUCCESS;}
});
consumer.start();

4. 延迟消费
延迟消费不是通过消费者的特定设置来实现的，而是在发送消息时设置消息的延迟等级。

DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName");
producer.setNamesrvAddr("nameserver1:9876");
producer.start();
Message msg = new Message("TopicTest", "TagA", "OrderID188", "Hello world".getBytes());
// 设置延迟等级3，这会使消息延迟10s再被消费
msg.setDelayTimeLevel(3);
SendResult sendResult = producer.send(msg);
producer.shutdown();

接收延迟消息的代码与普通消息的接收方式相同，不需要特殊配置。

5. 重试和死信队列
对于处理失败的消息，RocketMQ 会自动重试，无需特别配置。如果重试次数达到上限仍然失败，消息会被转移到死信队列。消费死信队列中的消息需要订阅特定的Topic（%DLQ%+消费者组名）。

DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("ConsumerGroupName");
consumer.setNamesrvAddr("nameserver1:9876");
// 订阅死信队列
consumer.subscribe("%DLQ%ConsumerGroupName", "*");
consumer.registerMessageListener(new MessageListenerConcurrently() {@Overridepublic ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {for (MessageExt msg : msgs) {System.out.printf("DLQ Message: %s %n", new String(msg.getBody()));}return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}
});
consumer.start();