我在使用 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();
}
}
}
最佳答案
答案其实很简单，您发送的速度太快，达到了流量控制阈值。
当您发送 aysnc 消息时，客户端将尝试获取许可证，在收到代理响应后，它将释放许可证。
由于您以异步方式发送消息，其中 Producer.send() 将非常快地返回，并且您在 for 循环中继续发送消息，而没有任何 sleep 毫，这将面临问题。
只需在调用 Producer.send 之后放置Thread.sleep(100)或使用不需要SendCallBack的同步发送方法
关于java - org.apache.rocketmq.remoting.exception.RemotingTooMuchRequestException : invokeAsyncImpl invoke too fast，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/46767541/