[ROCKETMQ-355] Client asyncSend is not fully async

client/src/main/java/org/apache/rocketmq/client/ClientConfig.java

@@ -25,6 +25,7 @@
  * Client Common configuration
  */
 public class ClientConfig {
+    public static final String SEND_ASYNC_SEMAPHORE = "com.rocketmq.sendAsyncSemaphore";
     public static final String SEND_MESSAGE_WITH_VIP_CHANNEL_PROPERTY = "com.rocketmq.sendMessageWithVIPChannel";
     private String namesrvAddr = System.getProperty(MixAll.NAMESRV_ADDR_PROPERTY, System.getenv(MixAll.NAMESRV_ADDR_ENV));
     private String clientIP = RemotingUtil.getLocalAddress();
@@ -46,6 +47,8 @@ public class ClientConfig {
     private String unitName;
     private boolean vipChannelEnabled = Boolean.parseBoolean(System.getProperty(SEND_MESSAGE_WITH_VIP_CHANNEL_PROPERTY, "true"));
 
+    private int asyncSendSemaphore = Integer.parseInt(System.getProperty(SEND_ASYNC_SEMAPHORE, "65536"));
+
     private boolean useTLS = TlsSystemConfig.tlsEnable;
 
     public String buildMQClientId() {
@@ -186,6 +189,14 @@ public void setUseTLS(boolean useTLS) {
         this.useTLS = useTLS;
     }
 
+    public int getAsyncSendSemaphore() {
+        return asyncSendSemaphore;
+    }
+
+    public void setAsyncSendSemaphore(int asyncSendSemaphore) {
+        this.asyncSendSemaphore = asyncSendSemaphore;
+    }
+
     @Override
     public String toString() {
         return "ClientConfig [namesrvAddr=" + namesrvAddr + ", clientIP=" + clientIP + ", instanceName=" + instanceName

client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultMQProducerImpl.java

@@ -29,8 +29,11 @@
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.Semaphore;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
+
 import org.apache.rocketmq.common.message.Message;
 import org.apache.rocketmq.common.message.MessageClientIDSetter;
 import org.apache.rocketmq.common.message.MessageExt;
@@ -98,6 +101,8 @@ public class DefaultMQProducerImpl implements MQProducerInner {
 
     private MQFaultStrategy mqFaultStrategy = new MQFaultStrategy();
 
+    private Semaphore asyncSendSemphore = null;
+
     public DefaultMQProducerImpl(final DefaultMQProducer defaultMQProducer) {
         this(defaultMQProducer, null);
     }
@@ -145,6 +150,10 @@ public void start(final boolean startFactory) throws MQClientException {
 
                 this.checkConfig();
 
+                if (asyncSendSemphore == null) {
+                    asyncSendSemphore = new Semaphore(defaultMQProducer.getAsyncSendSemaphore());
+                }
+
                 if (!this.defaultMQProducer.getProducerGroup().equals(MixAll.CLIENT_INNER_PRODUCER_GROUP)) {
                     this.defaultMQProducer.changeInstanceNameToPID();
                 }
@@ -406,20 +415,99 @@ public MessageExt queryMessageByUniqKey(String topic, String uniqKey)
         return this.mQClientFactory.getMQAdminImpl().queryMessageByUniqKey(topic, uniqKey);
     }
 
+    private static class AsyncSendCallback implements SendCallback {
+        private SendCallback realSendCallback;
+        private Semaphore semaphore;
+
+        public AsyncSendCallback(SendCallback sendCallback, Semaphore semaphore) {
+            this.realSendCallback = sendCallback;
+            this.semaphore = semaphore;
+        }
+
+        @Override
+        public void onSuccess(SendResult sendResult) {
+            try {
+                realSendCallback.onSuccess(sendResult);
+            } finally {
+                semaphore.release();
+            }
+        }
+
+        @Override
+        public void onException(Throwable e) {
+            try {
+                realSendCallback.onException(e);
+            } finally {
+                semaphore.release();
+            }
+        }
+    }
+
+    // FIXME: 2018/1/25
+    private ExecutorService getAsyncSendExecutor() {
+        return this.getCallbackExecutor();
+    }
+
+    private void doAsyncSend(Runnable runnable, final SendCallback sendCallback) {
+        try {
+            getAsyncSendExecutor().submit(runnable);
+        } catch (RejectedExecutionException e) {
+            sendCallback.onException(e);
+        }
+    }
+
+    private void asyncHandleException(final SendCallback sendCallback, final Throwable e) {
+        try {
+            getAsyncSendExecutor().submit(new Runnable() {
+                @Override
+                public void run() {
+                    sendCallback.onException(e);
+                }
+            });
+        } catch (RejectedExecutionException e1) {
+            sendCallback.onException(e);
+        }
+    }
+
     /**
      * DEFAULT ASYNC -------------------------------------------------------
      */
-    public void send(Message msg,
-        SendCallback sendCallback) throws MQClientException, RemotingException, InterruptedException {
+    public void send(Message msg, SendCallback sendCallback) {
         send(msg, sendCallback, this.defaultMQProducer.getSendMsgTimeout());
     }
 
-    public void send(Message msg, SendCallback sendCallback, long timeout)
-        throws MQClientException, RemotingException, InterruptedException {
+    public void send(final Message msg, final SendCallback sendCallback, final long timeout) {
         try {
-            this.sendDefaultImpl(msg, CommunicationMode.ASYNC, sendCallback, timeout);
-        } catch (MQBrokerException e) {
-            throw new MQClientException("unknownn exception", e);
+            boolean acquire = asyncSendSemphore.tryAcquire(timeout, TimeUnit.MILLISECONDS);
+            if (acquire) {
+                final AsyncSendCallback asyncSendCallback = new AsyncSendCallback(sendCallback, asyncSendSemphore);
+                doAsyncSend(new Runnable() {
+                    @Override
+                    public void run() {
+                        try {
+                            sendDefaultImpl(msg, CommunicationMode.ASYNC, asyncSendCallback, timeout);
+                        } catch (Exception e) {
+                            handleCallbackException(e, asyncSendCallback);
+                        }
+                    }
+                }, asyncSendCallback);
+            } else {
+                asyncHandleException(sendCallback, new RejectedExecutionException());
+            }
+        } catch (InterruptedException e) {
+            asyncHandleException(sendCallback, e);
+        }
+    }
+
+    private void handleCallbackException(Exception e, SendCallback sendCallback) {
+        if (sendCallback != null) {
+            if (e instanceof MQBrokerException) {
+                sendCallback.onException(new MQClientException("unknown exception", e));
+            } else {
+                sendCallback.onException(e);
+            }
+        } else {
+            log.warn("asyncSend message callback null real exception is " + e.getMessage(), e);
         }
     }
 
@@ -583,11 +671,11 @@ private TopicPublishInfo tryToFindTopicPublishInfo(final String topic) {
     }
 
     private SendResult sendKernelImpl(final Message msg,
-        final MessageQueue mq,
-        final CommunicationMode communicationMode,
-        final SendCallback sendCallback,
-        final TopicPublishInfo topicPublishInfo,
-        final long timeout) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
+                                      final MessageQueue mq,
+                                      final CommunicationMode communicationMode,
+                                      final SendCallback sendCallback,
+                                      final TopicPublishInfo topicPublishInfo,
+                                      final long timeout) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
         String brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(mq.getBrokerName());
         if (null == brokerAddr) {
             tryToFindTopicPublishInfo(mq.getTopic());
@@ -842,32 +930,46 @@ public SendResult send(Message msg, MessageQueue mq, long timeout)
     /**
      * KERNEL ASYNC -------------------------------------------------------
      */
-    public void send(Message msg, MessageQueue mq, SendCallback sendCallback)
-        throws MQClientException, RemotingException, InterruptedException {
+    public void send(Message msg, MessageQueue mq, SendCallback sendCallback) {
         send(msg, mq, sendCallback, this.defaultMQProducer.getSendMsgTimeout());
     }
 
-    public void send(Message msg, MessageQueue mq, SendCallback sendCallback, long timeout)
-        throws MQClientException, RemotingException, InterruptedException {
-        this.makeSureStateOK();
-        Validators.checkMessage(msg, this.defaultMQProducer);
-
-        if (!msg.getTopic().equals(mq.getTopic())) {
-            throw new MQClientException("message's topic not equal mq's topic", null);
-        }
-
+    public void send(final Message msg, final MessageQueue mq, final SendCallback sendCallback, final long timeout) {
         try {
-            this.sendKernelImpl(msg, mq, CommunicationMode.ASYNC, sendCallback, null, timeout);
-        } catch (MQBrokerException e) {
-            throw new MQClientException("unknown exception", e);
+            boolean acquire = asyncSendSemphore.tryAcquire(timeout, TimeUnit.MILLISECONDS);
+            if (acquire) {
+                final AsyncSendCallback asyncSendCallback = new AsyncSendCallback(sendCallback, asyncSendSemphore);
+                doAsyncSend(new Runnable() {
+                    @Override
+                    public void run() {
+                        try {
+
+                            makeSureStateOK();
+                            Validators.checkMessage(msg, defaultMQProducer);
+
+                            if (!msg.getTopic().equals(mq.getTopic())) {
+                                throw new MQClientException("message's topic not equal mq's topic", null);
+                            }
+                            sendKernelImpl(msg, mq, CommunicationMode.ASYNC, asyncSendCallback, null, timeout);
+                        } catch (Exception e) {
+                            handleCallbackException(e, asyncSendCallback);
+                        }
+                    }
+                }, asyncSendCallback);
+
+            } else {
+                asyncHandleException(sendCallback, new RejectedExecutionException());
+            }
+        } catch (InterruptedException e) {
+            asyncHandleException(sendCallback, e);
         }
     }
 
     /**
      * KERNEL ONEWAY -------------------------------------------------------
      */
     public void sendOneway(Message msg,
-        MessageQueue mq) throws MQClientException, RemotingException, InterruptedException {
+                           MessageQueue mq) throws MQClientException, RemotingException, InterruptedException {
         this.makeSureStateOK();
         Validators.checkMessage(msg, this.defaultMQProducer);
 
@@ -923,17 +1025,31 @@ private SendResult sendSelectImpl(
     /**
      * SELECT ASYNC -------------------------------------------------------
      */
-    public void send(Message msg, MessageQueueSelector selector, Object arg, SendCallback sendCallback)
-        throws MQClientException, RemotingException, InterruptedException {
+    public void send(Message msg, MessageQueueSelector selector, Object arg, SendCallback sendCallback) {
         send(msg, selector, arg, sendCallback, this.defaultMQProducer.getSendMsgTimeout());
     }
 
-    public void send(Message msg, MessageQueueSelector selector, Object arg, SendCallback sendCallback, long timeout)
-        throws MQClientException, RemotingException, InterruptedException {
+    public void send(final Message msg, final MessageQueueSelector selector, final Object arg, final SendCallback sendCallback, final long timeout) {
         try {
-            this.sendSelectImpl(msg, selector, arg, CommunicationMode.ASYNC, sendCallback, timeout);
-        } catch (MQBrokerException e) {
-            throw new MQClientException("unknownn exception", e);
+            boolean acquire = asyncSendSemphore.tryAcquire(timeout, TimeUnit.MILLISECONDS);
+            if (acquire) {
+                final AsyncSendCallback asyncSendCallback = new AsyncSendCallback(sendCallback, asyncSendSemphore);
+                doAsyncSend(new Runnable() {
+                    @Override
+                    public void run() {
+                        try {
+                            sendSelectImpl(msg, selector, arg, CommunicationMode.ASYNC, asyncSendCallback, timeout);
+                        } catch (Exception e) {
+                            handleCallbackException(e, asyncSendCallback);
+                        }
+                    }
+                }, asyncSendCallback);
+
+            } else {
+                asyncHandleException(sendCallback, new RejectedExecutionException());
+            }
+        } catch (InterruptedException e) {
+            asyncHandleException(sendCallback, e);
         }
     }
 
@@ -950,7 +1066,7 @@ public void sendOneway(Message msg, MessageQueueSelector selector, Object arg)
     }
 
     public TransactionSendResult sendMessageInTransaction(final Message msg,
-        final LocalTransactionExecuter tranExecuter, final Object arg)
+                                                          final LocalTransactionExecuter tranExecuter, final Object arg)
         throws MQClientException {
         if (null == tranExecuter) {
             throw new MQClientException("tranExecutor is null", null);
@@ -1064,8 +1180,12 @@ public void setCallbackExecutor(final ExecutorService callbackExecutor) {
         this.mQClientFactory.getMQClientAPIImpl().getRemotingClient().setCallbackExecutor(callbackExecutor);
     }
 
+    private ExecutorService getCallbackExecutor() {
+        return this.mQClientFactory.getMQClientAPIImpl().getRemotingClient().getCallbackExecutor();
+    }
+
     public SendResult send(Message msg,
-        long timeout) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
+                           long timeout) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
         return this.sendDefaultImpl(msg, CommunicationMode.SYNC, null, timeout);
     }
 

client/src/main/java/org/apache/rocketmq/client/producer/DefaultMQProducer.java

@@ -250,8 +250,8 @@ public SendResult send(Message msg,
      * @throws InterruptedException if the sending thread is interrupted.
      */
     @Override
-    public void send(Message msg,
-        SendCallback sendCallback) throws MQClientException, RemotingException, InterruptedException {
+    public void send(Message msg, SendCallback sendCallback)
+        throws MQClientException, RemotingException, InterruptedException {
         this.defaultMQProducerImpl.send(msg, sendCallback);
     }
 

client/src/main/java/org/apache/rocketmq/client/producer/MQProducer.java

@@ -70,11 +70,11 @@ SendResult send(final Message msg, final MessageQueueSelector selector, final Ob
         InterruptedException;
 
     void send(final Message msg, final MessageQueueSelector selector, final Object arg,
-        final SendCallback sendCallback) throws MQClientException, RemotingException,
+              final SendCallback sendCallback) throws MQClientException, RemotingException,
         InterruptedException;
 
     void send(final Message msg, final MessageQueueSelector selector, final Object arg,
-        final SendCallback sendCallback, final long timeout) throws MQClientException, RemotingException,
+              final SendCallback sendCallback, final long timeout) throws MQClientException, RemotingException,
         InterruptedException;
 
     void sendOneway(final Message msg, final MessageQueueSelector selector, final Object arg)