RocketMQ的可靠性传输

引用消息队列时，如何保证消息的可靠传输是一大难点，本文推荐了一个尽最大可能去避免消息丢失的解决方案。

整体

分析：

需确保一发一存一消费这些过程均无消息丢失

利用ACK机制保证每个阶段需要执行的操作成功后，再往下一个阶段推动（放行）

实现：

Producer——>Broker

• 发送方式

  • 同步发送

    • Producer向broker发送消息，会阻塞当前线程等待broker响应结果

    public class SyncProducer {
    	public static void main(String[] args) throws Exception {
        	// 实例化消息生产者Producer
            DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
        	// 设置NameServer的地址
    	    	producer.setNamesrvAddr("localhost:9876");
        	// 启动Producer实例
            producer.start();
        	for (int i = 0; i < 100; i++) {
        	    // 创建消息，并指定Topic，Tag和消息体
        	    Message msg = new Message("TopicTest" /* Topic */,
            	"TagA" /* Tag */,
            	("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
            	);
            	// 发送消息到一个Broker
                SendResult sendResult = producer.send(msg);
                // 通过sendResult返回消息是否成功送达
                System.out.printf("%s%n", sendResult);
        	}
        	// 如果不再发送消息，关闭Producer实例。
        	producer.shutdown();
        }
    }
    

  • 异步发送

    • Producer首先构建一个向broker发送消息的任务，把该任务提交给线程池，等执行完该任务时，回调用户自定义的回调函数，执行处理结果

    public class AsyncProducer {
    	public static void main(String[] args) throws Exception {
        	// 实例化消息生产者Producer
            DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
        	// 设置NameServer的地址
            producer.setNamesrvAddr("localhost:9876");
        	// 启动Producer实例
            producer.start();
            producer.setRetryTimesWhenSendAsyncFailed(0);
    	
    	int messageCount = 100;
            // 根据消息数量实例化倒计时计算器
    	final CountDownLatch2 countDownLatch = new CountDownLatch2(messageCount);
        	for (int i = 0; i < messageCount; i++) {
                    final int index = i;
                	// 创建消息，并指定Topic，Tag和消息体
                    Message msg = new Message("TopicTest",
                        "TagA",
                        "OrderID188",
                        "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));
                    // SendCallback接收异步返回结果的回调
                    producer.send(msg, new SendCallback() {
                        @Override
                        public void onSuccess(SendResult sendResult) {
                            countDownLatch.countDown();
                            System.out.printf("%-10d OK %s %n", index,
                                sendResult.getMsgId());
                        }
                        @Override
                        public void onException(Throwable e) {
                            countDownLatch.countDown();
          	                System.out.printf("%-10d Exception %s %n", index, e);
          	                e.printStackTrace();
                        }
                	});
        	}
    	// 等待5s
    	countDownLatch.await(5, TimeUnit.SECONDS);
        	// 如果不再发送消息，关闭Producer实例。
        	producer.shutdown();
        }
    }
    

  • Oneway

    • Oneway方式只负责发送请求，不等待应答，Producer只负责把请求发出去，不会处理响应结果

    public class OnewayProducer {
    	public static void main(String[] args) throws Exception{
        	// 实例化消息生产者Producer
            DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
        	// 设置NameServer的地址
            producer.setNamesrvAddr("localhost:9876");
        	// 启动Producer实例
            producer.start();
        	for (int i = 0; i < 100; i++) {
            	// 创建消息，并指定Topic，Tag和消息体
            	Message msg = new Message("TopicTest" /* Topic */,
                    "TagA" /* Tag */,
                    ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
            	);
            	// 发送单向消息，没有任何返回结果
            	producer.sendOneway(msg);
    
        	}
        	// 如果不再发送消息，关闭Producer实例。
        	producer.shutdown();
        }
    }