org.apache.rocketmq.remoting.exception.RemotingTooMuchRequestException: invokeAsyncImpl调用太快

Question

我在使用Rocket MQ(v4.1.0-孵化)客户端时遇到严重问题，如下所示：

2017-10-16 16:18:12:457[ERROR][SimpleProducer$1.onException(SimpleProducer.java:44)] - send message to mq fail:
org.apache.rocketmq.remoting.exception.RemotingTooMuchRequestException: invokeAsyncImpl invoke too fast
    at org.apache.rocketmq.remoting.netty.NettyRemotingAbstract.invokeAsyncImpl(NettyRemotingAbstract.java:422)
    at org.apache.rocketmq.remoting.netty.NettyRemotingClient.invokeAsync(NettyRemotingClient.java:488)
    at org.apache.rocketmq.client.impl.MQClientAPIImpl.sendMessageAsync(MQClientAPIImpl.java:368)
    at org.apache.rocketmq.client.impl.MQClientAPIImpl.onExceptionImpl(MQClientAPIImpl.java:455)
    at org.apache.rocketmq.client.impl.MQClientAPIImpl.access$100(MQClientAPIImpl.java:156)
    at org.apache.rocketmq.client.impl.MQClientAPIImpl$1.operationComplete(MQClientAPIImpl.java:417)
    at org.apache.rocketmq.remoting.netty.ResponseFuture.executeInvokeCallback(ResponseFuture.java:51)
    at org.apache.rocketmq.remoting.netty.NettyRemotingAbstract$2.run(NettyRemotingAbstract.java:275)
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
    at java.util.concurrent.FutureTask.run(FutureTask.java:266)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)

我不知道如何修复它，即使我谷歌了很多，没有任何正确的答案。

下面是我以异步方式编写的生成器代码：

public class SimpleProducer {

static final Logger logger = LoggerFactory.getLogger(SimpleProducer.class);
static AtomicInteger total = new AtomicInteger(0);
private final static CountDownLatch mCountDownLatch = new CountDownLatch(1);
public static void main(String[] args){
    logger.info("Bootstrap start...");
    DefaultMQProducer producer = new DefaultMQProducer("Producer");
    producer.setNamesrvAddr("192.168.137.112:9876");
    try {
        producer.start();
        producer.setRetryTimesWhenSendAsyncFailed(3);
        long start = System.currentTimeMillis();
        for (int i = 0; i < 10000000; i++)
            try {
                {
                    Message msg = new Message("newTopic",
                            "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));
                   producer.send(msg, new SendCallback(){
                       public void onSuccess(SendResult sendResult) {
                           if (sendResult.getSendStatus().equals(SendStatus.SEND_OK)) {
                               //MsgSendResponseCounter.factory.getInstance().onSuccess();
                               logger.info("Succeeded send {} message, total {}", sendResult.getMsgId(), total.getAndIncrement());
                           }
                       }
                       public void onException(Throwable ee) {
                           logger.error("send message to mq fail:", ee);
                       }
                   });
                }

            } catch (Exception e) {
                e.printStackTrace();
            }
        mCountDownLatch.await();
        long end = System.currentTimeMillis();
        logger.info("Finished 10000000 msgs! elapsed time {} in all", (end - start)/1000);
    } catch (Exception e) {
        e.printStackTrace();
    }finally{
        producer.shutdown();
    }
}

}

Answer 1

答案其实很简单，你的发送速度太快了，以至于达到了流量控制阈值。

当您发送aysnc消息时，客户端将尝试获取许可证，在收到broker响应后，它将释放许可证。

由于您正在以异步方式发送消息，而producer.send()将以非常快的速度返回，并且您继续在for循环中发送消息，而没有任何休眠毫秒，这将面临问题。

只需在调用producer.send或使用不需要SendCallBack的同步发送方法后放入Thread.sleep(100