1

我在使用 Rocket MQ（v4.1.0-incubating）客户端时遇到了关键问题，如下所示：

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();
    }
}

}

4

1 回答 1

1

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

当你发送aysnc消息时，客户端会尝试获取一个permit，在收到broker响应后，它会释放permit。

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

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

于 2017-11-09T07:23:18.590 回答