RocketMQ源码：producer 异步发送(容错机制)

快来打我* 2024-03-25 21:18 43阅读 0赞

## 1. 前言 ##

异步发送消息其实很多地方与同步发送一样，不过有一点是你在编程的时候需要提供`SendCallback` 对象，用来`发送响应来的时候进行回调使用`，我们知道同步发送是等待broker响应到来，然后将响应往上返回，这个异步调用就是响应来的时候，对你提供的回调对象进行调用，你这个回调对象可以写一些自己的逻辑等等。

## 2. 简单使用 ##

在源码解析之前我们要先看一下异步发送消息是怎样编程的: org.apache.rocketmq.example.simple.AsyncProducer:

public static void main(
        String[] args) throws MQClientException, InterruptedException, UnsupportedEncodingException {
    
        DefaultMQProducer producer = new DefaultMQProducer("Jodie_Daily_test");
        producer.start();
        producer.setRetryTimesWhenSendAsyncFailed(0);
    
        int messageCount = 100;
        final CountDownLatch countDownLatch = new CountDownLatch(messageCount);
        for (int i = 0; i < messageCount; i++) {
            try {
                final int index = i;
                Message msg = new Message("Jodie_topic_1023",
                    "TagA",
                    "OrderID188",
                    "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));
                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();
                    }
                });
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        countDownLatch.await(5, TimeUnit.SECONDS);
        producer.shutdown();
    }
    复制代码

这里需要我们提供一个`SendCallback` 对象用来响应来的时候回调，其中异常的时候会调用这里面的`onException`方法， 成功的时候调用`onSuccess`方法，执行相应的逻辑。

## 3. 源码解析 ##

好了，现在开始源码解析：

public void send(Message msg,
        SendCallback sendCallback) throws MQClientException, RemotingException, InterruptedException {
        msg.setTopic(withNamespace(msg.getTopic()));
        this.defaultMQProducerImpl.send(msg, sendCallback);
    }
    复制代码

首先是调用了`defaultMQProducerImpl` 的`send`方法，并且将`msg` 与`callback`传进去:

public void send(Message msg,
        SendCallback sendCallback) throws MQClientException, RemotingException, InterruptedException {
        send(msg, sendCallback, this.defaultMQProducer.getSendMsgTimeout());
    }
    复制代码

这里又调用了下它的重载方法，然后将发送超时时间传进去，默认的一个发送超时时间是`3s`。

public void send(final Message msg, final SendCallback sendCallback, final long timeout)
        throws MQClientException, RemotingException, InterruptedException {
        final long beginStartTime = System.currentTimeMillis();
        // 异步线程池
        ExecutorService executor = this.getAsyncSenderExecutor();
        try {
            executor.submit(new Runnable() {
                @Override
                public void run() {
                    long costTime = System.currentTimeMillis() - beginStartTime;
                    if (timeout > costTime) {
                        try {
                            // todo
                            sendDefaultImpl(msg, CommunicationMode.ASYNC, sendCallback, timeout - costTime);
                        } catch (Exception e) {
                            sendCallback.onException(e);
                        }
                    } else {
                        sendCallback.onException(
                            new RemotingTooMuchRequestException("DEFAULT ASYNC send call timeout"));
                    }
                }
    
            });
        } 
        ...
    }
    复制代码

这里需要注意的是，它将这个发送消息的任务交给了一个`异步发送线程池`，然后在任务中是调用了`sendDefaultImpl` 方法，然后通信方式是`异步CommunicationMode.ASYNC`，这里我们需要看下这个线程池的一些参数，因为关乎我们以后的调优:

public ExecutorService getAsyncSenderExecutor() {
        return null == asyncSenderExecutor ? defaultAsyncSenderExecutor : asyncSenderExecutor;
    }
    复制代码

这里先是判断`asyncSenderExecutor` 这个线程池是不是null，其实咱们这里它是null的（因为我们没有指定线程池，不过你编程的时候是可以指定的，使用`setAsyncSenderExecutor()`这个方法就可以设置了），就使用`defaultAsyncSenderExecutor`线程池，这个`defaultAsyncSenderExecutor` 线程池是在`defaultMQProducerImpl` 类构造方法创建的的，我们可以看下:

public DefaultMQProducerImpl(final DefaultMQProducer defaultMQProducer, RPCHook rpcHook) {
        this.defaultMQProducer = defaultMQProducer;
        this.rpcHook = rpcHook;
    
        // 5w大小的链表队列，异步发送线程池队列
        this.asyncSenderThreadPoolQueue = new LinkedBlockingQueue<Runnable>(50000);
        // 默认的异步发送线程池
        this.defaultAsyncSenderExecutor = new ThreadPoolExecutor(
            Runtime.getRuntime().availableProcessors(),// 核心线程池数，cpu核数
            Runtime.getRuntime().availableProcessors(), // 最大线程池数，cpu核数
            1000 * 60,
            TimeUnit.MILLISECONDS,
            this.asyncSenderThreadPoolQueue,
            new ThreadFactory() {
                private AtomicInteger threadIndex = new AtomicInteger(0);
    
                @Override
                public Thread newThread(Runnable r) {
                    return new Thread(r, "AsyncSenderExecutor_" + this.th