From 67545d976c608211eeb07feca0b9e39b5f04c2b1 Mon Sep 17 00:00:00 2001 From: thisisbaozi Date: Sun, 1 Jan 2017 10:52:47 +0800 Subject: [PATCH 1/6] add new way to send message in asynchronous way --- .../rocketmq/client/impl/MQClientAPIImpl.java | 65 ++-- .../impl/producer/DefaultMQProducerImpl.java | 56 ++-- .../impl/producer/DefaultSendPromise.java | 281 ++++++++++++++++++ .../client/impl/producer/SendPromise.java | 11 + .../client/producer/DefaultMQProducer.java | 7 + .../rocketmq/client/producer/MQProducer.java | 4 + .../rocketmq/client/producer/SendFuture.java | 21 ++ .../rocketmq/client/TestSendPromise.java | 65 ++++ 8 files changed, 464 insertions(+), 46 deletions(-) create mode 100644 client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultSendPromise.java create mode 100644 client/src/main/java/org/apache/rocketmq/client/impl/producer/SendPromise.java create mode 100644 client/src/main/java/org/apache/rocketmq/client/producer/SendFuture.java create mode 100644 client/src/test/java/org/apache/rocketmq/client/TestSendPromise.java diff --git a/client/src/main/java/org/apache/rocketmq/client/impl/MQClientAPIImpl.java b/client/src/main/java/org/apache/rocketmq/client/impl/MQClientAPIImpl.java index 12580c14977..8c1c94ec605 100644 --- a/client/src/main/java/org/apache/rocketmq/client/impl/MQClientAPIImpl.java +++ b/client/src/main/java/org/apache/rocketmq/client/impl/MQClientAPIImpl.java @@ -37,6 +37,7 @@ import org.apache.rocketmq.client.impl.consumer.PullResultExt; import org.apache.rocketmq.client.impl.factory.MQClientInstance; import org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl; +import org.apache.rocketmq.client.impl.producer.SendPromise; import org.apache.rocketmq.client.impl.producer.TopicPublishInfo; import org.apache.rocketmq.client.log.ClientLogger; import org.apache.rocketmq.client.producer.SendCallback; @@ -287,7 +288,7 @@ public SendResult sendMessage(// final SendMessageContext context, // 7 final DefaultMQProducerImpl producer // 8 ) throws RemotingException, MQBrokerException, InterruptedException { - return sendMessage(addr, brokerName, msg, requestHeader, timeoutMillis, communicationMode, null, null, null, 0, context, producer); + return sendMessage(addr, brokerName, msg, requestHeader, timeoutMillis, communicationMode, null, null, null, 0, context, null, producer); } public SendResult sendMessage(// @@ -302,6 +303,7 @@ public SendResult sendMessage(// final MQClientInstance instance, // 9 final int retryTimesWhenSendFailed, // 10 final SendMessageContext context, // 11 + final SendPromise promise, final DefaultMQProducerImpl producer // 12 ) throws RemotingException, MQBrokerException, InterruptedException { RemotingCommand request = null; @@ -321,7 +323,7 @@ public SendResult sendMessage(// case ASYNC: final AtomicInteger times = new AtomicInteger(); this.sendMessageAsync(addr, brokerName, msg, timeoutMillis, request, sendCallback, topicPublishInfo, instance, - retryTimesWhenSendFailed, times, context, producer); + retryTimesWhenSendFailed, times, context, promise, producer); return null; case SYNC: return this.sendMessageSync(addr, brokerName, msg, timeoutMillis, request); @@ -345,6 +347,34 @@ private SendResult sendMessageSync(// return this.processSendResponse(brokerName, msg, response); } + public static void safeInvoke(SendCallback callback, SendPromise promise, SendResult result) { + try { + if (callback != null) { + callback.onSuccess(result); + } else if (promise != null) { + promise.complete(result); + } + } catch (Throwable cause) { + if (log.isDebugEnabled()) { + log.error("Caught unknown exception {} while invoking callback", cause.getClass().getName(), cause); + } + } + } + + public static void safeReport(SendCallback callback, SendPromise promise, Throwable cause) { + try { + if (callback != null) { + callback.onException(cause); + } else if (promise != null) { + promise.report(cause); + } + } catch (Throwable cause1) { + if (log.isDebugEnabled()) { + log.error("Caught unknown exception {} while invoking callback", cause1.getClass().getName(), cause1); + } + } + } + private void sendMessageAsync(// final String addr, // final String brokerName, // @@ -357,6 +387,7 @@ private void sendMessageAsync(// final int retryTimesWhenSendFailed, // final AtomicInteger times, // final SendMessageContext context, // + final SendPromise promise, final DefaultMQProducerImpl producer // ) throws InterruptedException, RemotingException { this.remotingClient.invokeAsync(addr, request, timeoutMillis, new InvokeCallback() { @@ -388,32 +419,29 @@ public void operationComplete(ResponseFuture responseFuture) { context.getProducer().executeSendMessageHookAfter(context); } - try { - sendCallback.onSuccess(sendResult); - } catch (Throwable e) { - } + safeInvoke(sendCallback, promise, sendResult); producer.updateFaultItem(brokerName, System.currentTimeMillis() - responseFuture.getBeginTimestamp(), false); } catch (Exception e) { producer.updateFaultItem(brokerName, System.currentTimeMillis() - responseFuture.getBeginTimestamp(), true); onExceptionImpl(brokerName, msg, 0L, request, sendCallback, topicPublishInfo, instance, - retryTimesWhenSendFailed, times, e, context, false, producer); + retryTimesWhenSendFailed, times, e, context, false, promise, producer); } } else { producer.updateFaultItem(brokerName, System.currentTimeMillis() - responseFuture.getBeginTimestamp(), true); if (!responseFuture.isSendRequestOK()) { MQClientException ex = new MQClientException("send request failed", responseFuture.getCause()); onExceptionImpl(brokerName, msg, 0L, request, sendCallback, topicPublishInfo, instance, - retryTimesWhenSendFailed, times, ex, context, true, producer); + retryTimesWhenSendFailed, times, ex, context, true, promise, producer); } else if (responseFuture.isTimeout()) { MQClientException ex = new MQClientException("wait response timeout " + responseFuture.getTimeoutMillis() + "ms", responseFuture.getCause()); onExceptionImpl(brokerName, msg, 0L, request, sendCallback, topicPublishInfo, instance, - retryTimesWhenSendFailed, times, ex, context, true, producer); + retryTimesWhenSendFailed, times, ex, context, true, promise, producer); } else { MQClientException ex = new MQClientException("unknow reseaon", responseFuture.getCause()); onExceptionImpl(brokerName, msg, 0L, request, sendCallback, topicPublishInfo, instance, - retryTimesWhenSendFailed, times, ex, context, true, producer); + retryTimesWhenSendFailed, times, ex, context, true, promise, producer); } } } @@ -432,6 +460,7 @@ private void onExceptionImpl(final String brokerName, // final Exception e, // final SendMessageContext context, // final boolean needRetry, // + final SendPromise promise, final DefaultMQProducerImpl producer // 12 ) { int tmp = curTimes.incrementAndGet(); @@ -443,31 +472,29 @@ private void onExceptionImpl(final String brokerName, // try { request.setOpaque(RemotingCommand.createNewRequestId()); sendMessageAsync(addr, tmpmq.getBrokerName(), msg, timeoutMillis, request, sendCallback, topicPublishInfo, instance, - timesTotal, curTimes, context, producer); + timesTotal, curTimes, context, promise, producer); } catch (InterruptedException e1) { onExceptionImpl(tmpmq.getBrokerName(), msg, timeoutMillis, request, sendCallback, topicPublishInfo, instance, timesTotal, curTimes, e1, - context, false, producer); + context, false, promise, producer); } catch (RemotingConnectException e1) { producer.updateFaultItem(brokerName, 3000, true); onExceptionImpl(tmpmq.getBrokerName(), msg, timeoutMillis, request, sendCallback, topicPublishInfo, instance, timesTotal, curTimes, e1, - context, true, producer); + context, true, promise, producer); } catch (RemotingTooMuchRequestException e1) { onExceptionImpl(tmpmq.getBrokerName(), msg, timeoutMillis, request, sendCallback, topicPublishInfo, instance, timesTotal, curTimes, e1, - context, false, producer); + context, false, promise, producer); } catch (RemotingException e1) { producer.updateFaultItem(brokerName, 3000, true); onExceptionImpl(tmpmq.getBrokerName(), msg, timeoutMillis, request, sendCallback, topicPublishInfo, instance, timesTotal, curTimes, e1, - context, true, producer); + context, true, promise, producer); } } else { if (context != null) { context.setException(e); context.getProducer().executeSendMessageHookAfter(context); } - try { - sendCallback.onException(e); - } catch (Exception ignored) { - } + + safeReport(sendCallback, promise, e); } } diff --git a/client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultMQProducerImpl.java b/client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultMQProducerImpl.java index 8e819792c4c..76ad744b21e 100644 --- a/client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultMQProducerImpl.java +++ b/client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultMQProducerImpl.java @@ -24,12 +24,8 @@ import java.util.List; import java.util.Random; import java.util.Set; -import java.util.concurrent.BlockingQueue; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.LinkedBlockingQueue; -import java.util.concurrent.ThreadPoolExecutor; -import java.util.concurrent.TimeUnit; +import java.util.concurrent.*; + import org.apache.rocketmq.client.QueryResult; import org.apache.rocketmq.client.Validators; import org.apache.rocketmq.client.common.ClientErrorCode; @@ -44,16 +40,7 @@ import org.apache.rocketmq.client.impl.factory.MQClientInstance; import org.apache.rocketmq.client.latency.MQFaultStrategy; import org.apache.rocketmq.client.log.ClientLogger; -import org.apache.rocketmq.client.producer.DefaultMQProducer; -import org.apache.rocketmq.client.producer.LocalTransactionExecuter; -import org.apache.rocketmq.client.producer.LocalTransactionState; -import org.apache.rocketmq.client.producer.MessageQueueSelector; -import org.apache.rocketmq.client.producer.SendCallback; -import org.apache.rocketmq.client.producer.SendResult; -import org.apache.rocketmq.client.producer.SendStatus; -import org.apache.rocketmq.client.producer.TransactionCheckListener; -import org.apache.rocketmq.client.producer.TransactionMQProducer; -import org.apache.rocketmq.client.producer.TransactionSendResult; +import org.apache.rocketmq.client.producer.*; import org.apache.rocketmq.common.MixAll; import org.apache.rocketmq.common.ServiceState; import org.apache.rocketmq.common.UtilAll; @@ -412,12 +399,25 @@ public void send(Message msg, SendCallback sendCallback) throws MQClientExceptio public void send(Message msg, SendCallback sendCallback, long timeout) throws MQClientException, RemotingException, InterruptedException { try { - this.sendDefaultImpl(msg, CommunicationMode.ASYNC, sendCallback, timeout); + this.sendDefaultImpl(msg, CommunicationMode.ASYNC, sendCallback, timeout, null); } catch (MQBrokerException e) { throw new MQClientException("unknownn exception", e); } } + public SendFuture send(Message message, Executor executor, long timeout) throws MQClientException, RemotingException { + SendPromise promise = new DefaultSendPromise(executor); + try { + this.sendDefaultImpl(message, CommunicationMode.ASYNC, null, timeout, promise); + } catch (MQBrokerException e) { + throw new MQClientException("unknownn exception", e); + } catch (InterruptedException e) { + // ignore + } + + return promise; + } + public MessageQueue selectOneMessageQueue(final TopicPublishInfo tpInfo, final String lastBrokerName) { return this.mqFaultStrategy.selectOneMessageQueue(tpInfo, lastBrokerName); } @@ -430,7 +430,8 @@ private SendResult sendDefaultImpl(// Message msg, // final CommunicationMode communicationMode, // final SendCallback sendCallback, // - final long timeout// + final long timeout, + final SendPromise promise ) throws MQClientException, RemotingException, MQBrokerException, InterruptedException { this.makeSureStateOK(); Validators.checkMessage(msg, this.defaultMQProducer); @@ -455,7 +456,7 @@ private SendResult sendDefaultImpl(// brokersSent[times] = mq.getBrokerName(); try { beginTimestampPrev = System.currentTimeMillis(); - sendResult = this.sendKernelImpl(msg, mq, communicationMode, sendCallback, topicPublishInfo, timeout); + sendResult = this.sendKernelImpl(msg, mq, communicationMode, sendCallback, topicPublishInfo, timeout, promise); endTimestamp = System.currentTimeMillis(); this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, false); switch (communicationMode) { @@ -582,7 +583,7 @@ private SendResult sendKernelImpl(final Message msg, // final CommunicationMode communicationMode, // final SendCallback sendCallback, // final TopicPublishInfo topicPublishInfo, // - final long timeout) throws MQClientException, RemotingException, MQBrokerException, InterruptedException { + final long timeout, SendPromise promise) throws MQClientException, RemotingException, MQBrokerException, InterruptedException { String brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(mq.getBrokerName()); if (null == brokerAddr) { tryToFindTopicPublishInfo(mq.getTopic()); @@ -680,7 +681,8 @@ private SendResult sendKernelImpl(final Message msg, // topicPublishInfo, // 8 this.mQClientFactory, // 9 this.defaultMQProducer.getRetryTimesWhenSendAsyncFailed(), // 10 - context, // + context, + promise, this); break; case ONEWAY: @@ -801,7 +803,7 @@ public void executeSendMessageHookAfter(final SendMessageContext context) { */ public void sendOneway(Message msg) throws MQClientException, RemotingException, InterruptedException { try { - this.sendDefaultImpl(msg, CommunicationMode.ONEWAY, null, this.defaultMQProducer.getSendMsgTimeout()); + this.sendDefaultImpl(msg, CommunicationMode.ONEWAY, null, this.defaultMQProducer.getSendMsgTimeout(), null); } catch (MQBrokerException e) { throw new MQClientException("unknown exception", e); } @@ -824,7 +826,7 @@ public SendResult send(Message msg, MessageQueue mq, long timeout) throw new MQClientException("message's topic not equal mq's topic", null); } - return this.sendKernelImpl(msg, mq, CommunicationMode.SYNC, null, null, timeout); + return this.sendKernelImpl(msg, mq, CommunicationMode.SYNC, null, null, timeout, null); } /** @@ -845,7 +847,7 @@ public void send(Message msg, MessageQueue mq, SendCallback sendCallback, long t } try { - this.sendKernelImpl(msg, mq, CommunicationMode.ASYNC, sendCallback, null, timeout); + this.sendKernelImpl(msg, mq, CommunicationMode.ASYNC, sendCallback, null, timeout, null); } catch (MQBrokerException e) { throw new MQClientException("unknown exception", e); } @@ -859,7 +861,7 @@ public void sendOneway(Message msg, MessageQueue mq) throws MQClientException, R Validators.checkMessage(msg, this.defaultMQProducer); try { - this.sendKernelImpl(msg, mq, CommunicationMode.ONEWAY, null, null, this.defaultMQProducer.getSendMsgTimeout()); + this.sendKernelImpl(msg, mq, CommunicationMode.ONEWAY, null, null, this.defaultMQProducer.getSendMsgTimeout(), null); } catch (MQBrokerException e) { throw new MQClientException("unknown exception", e); } @@ -898,7 +900,7 @@ private SendResult sendSelectImpl(// } if (mq != null) { - return this.sendKernelImpl(msg, mq, communicationMode, sendCallback, null, timeout); + return this.sendKernelImpl(msg, mq, communicationMode, sendCallback, null, timeout, null); } else { throw new MQClientException("select message queue return null.", null); } @@ -1046,7 +1048,7 @@ public void endTransaction(// } public SendResult send(Message msg, long timeout) throws MQClientException, RemotingException, MQBrokerException, InterruptedException { - return this.sendDefaultImpl(msg, CommunicationMode.SYNC, null, timeout); + return this.sendDefaultImpl(msg, CommunicationMode.SYNC, null, timeout, null); } public ConcurrentHashMap getTopicPublishInfoTable() { diff --git a/client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultSendPromise.java b/client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultSendPromise.java new file mode 100644 index 00000000000..8c55a1370c0 --- /dev/null +++ b/client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultSendPromise.java @@ -0,0 +1,281 @@ +package org.apache.rocketmq.client.impl.producer; + +import org.apache.rocketmq.client.log.ClientLogger; +import org.apache.rocketmq.client.producer.SendCallback; +import org.apache.rocketmq.client.producer.SendResult; +import org.apache.rocketmq.client.producer.SendFuture; +import org.slf4j.Logger; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.*; + +/** + * Created by stone + */ +public class DefaultSendPromise extends CountDownLatch implements SendPromise { + + private final Logger logger = ClientLogger.getLog(); + + private Executor executor; + private Object callbacks; + private volatile Object result; + + public DefaultSendPromise(Executor executor) { + super(1); + + if (executor == null) { + throw new NullPointerException("executor"); + } + if (executor instanceof ExecutorService) { + ExecutorService service = (ExecutorService) executor; + if (service.isShutdown() || service.isTerminated()) { + throw new IllegalArgumentException("terminated executor service"); + } + } + + this.executor = executor; + } + + @Override + public SendPromise complete(SendResult result) { + if (result == null) { + throw new NullPointerException("result"); + } + + if (competeOnce(result)) { + invokeCallbacks(); + } + + return this; + } + + public boolean competeOnce(Object result) { + if (isDone()) { + return false; + } + + synchronized (this) { + if (isDone()) { + return false; + } + + this.result = result; + countDown(); + } + + return true; + } + + @Override + public SendPromise report(Throwable cause) { + if (cause == null) { + throw new NullPointerException("cause"); + } + + if (competeOnce(cause)) { + invokeCallbacks(); + } + + return this; + } + + @Override + @SuppressWarnings("unchecked") + public SendFuture addCallback(SendCallback callback) { + if (callback == null) { + throw new NullPointerException("listener"); + } + + if (isDone()) { + invoke(callback); + return this; + } + + synchronized (this) { + if (!isDone()) { + if (callbacks == null) { + callbacks = callback; + return this; + } + + List callbacks; + if (this.callbacks instanceof List) { + callbacks = (List) this.callbacks; + callbacks.add(callback); + } else { + SendCallback previous = (SendCallback) this.callbacks; + callbacks = new ArrayList<>(); + callbacks.add(previous); + + this.callbacks = callbacks; + } + + callbacks.add(callback); + return this; + } + } + + invoke(callback); + return this; + } + + @Override + @SuppressWarnings("unchecked") + public SendFuture removeCallback(SendCallback listener) { + if (listener == null) { + throw new NullPointerException("listener"); + } + + if (isDone()) { + return this; + } + + synchronized (this) { + if (callbacks == null) { + return this; + } + + if (callbacks == listener) { + callbacks = null; + return this; + } + + List listeners = (List) this.callbacks; + listeners.remove(listener); + } + + return this; + } + + @SuppressWarnings("unchecked") + public void invokeCallbacks() { + if (callbacks == null) { + return ; + } + + if (callbacks instanceof List) { + List callbacks = (List) this.callbacks; + for (SendCallback callback : callbacks) { + invoke(callback); + } + } else { + SendCallback callback = (SendCallback) this.callbacks; + invoke(callback); + } + } + + public void invoke(final SendCallback callback) { + try { + executor.execute(new Runnable() { + @Override + public void run() { + Object result = DefaultSendPromise.this.result; + if (result instanceof Throwable) { + callback.onException((Throwable) result); + } else { + callback.onSuccess((SendResult) result); + } + } + }); + } catch (Throwable cause) { + if (logger.isWarnEnabled()) { + logger.warn("invoke listener({}) error", callback.getClass().getName(), cause); + } + } + } + + @Override + public Throwable getCause() { + if (result instanceof Throwable) { + return (Throwable) result; + } + return null; + } + + @Override + public final boolean cancel(boolean mayInterruptIfRunning) { + return false; + } + + @Override + public final boolean isCancelled() { + return false; + } + + @Override + public boolean isDone() { + return result != null; + } + + public SendFuture waitUntil() { + if (isDone()) { + return this; + } + + while (!isDone()) { + try { + await(); + } catch (InterruptedException e) { + // ignore + } + } + + return this; + } + + public boolean waitFor(long timeout, TimeUnit unit) throws InterruptedException { + if (isDone()) { + return true; + } + + if (timeout <= 0) { + return isDone(); + } + + await(timeout, unit); + return isDone(); + } + + @Override + public SendResult get() throws InterruptedException, ExecutionException { + if (Thread.interrupted()) { + throw new InterruptedException(); + } + + waitUntil(); + + Throwable cause = getCause(); + if (cause == null) { + if (!(result instanceof Throwable)) { + return (SendResult) result; + } + + return null; + } + + throw new ExecutionException(cause); + } + + @Override + public SendResult get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException { + if (Thread.interrupted()) { + throw new InterruptedException(); + } + + if (waitFor(timeout, unit)) { + Throwable cause = getCause(); + if (cause == null) { + if (!(result instanceof Throwable)) { + return (SendResult) result; + } + + return null; + } + + throw new ExecutionException(cause); + } + + throw new TimeoutException(); + } +} diff --git a/client/src/main/java/org/apache/rocketmq/client/impl/producer/SendPromise.java b/client/src/main/java/org/apache/rocketmq/client/impl/producer/SendPromise.java new file mode 100644 index 00000000000..d3a3aa2d0a0 --- /dev/null +++ b/client/src/main/java/org/apache/rocketmq/client/impl/producer/SendPromise.java @@ -0,0 +1,11 @@ +package org.apache.rocketmq.client.impl.producer; + +import org.apache.rocketmq.client.producer.SendFuture; +import org.apache.rocketmq.client.producer.SendResult; + +public interface SendPromise extends SendFuture { + + SendPromise complete(SendResult result); + SendPromise report(Throwable cause); + +} diff --git a/client/src/main/java/org/apache/rocketmq/client/producer/DefaultMQProducer.java b/client/src/main/java/org/apache/rocketmq/client/producer/DefaultMQProducer.java index c6773245473..5a7b9cbb345 100644 --- a/client/src/main/java/org/apache/rocketmq/client/producer/DefaultMQProducer.java +++ b/client/src/main/java/org/apache/rocketmq/client/producer/DefaultMQProducer.java @@ -17,6 +17,8 @@ package org.apache.rocketmq.client.producer; import java.util.List; +import java.util.concurrent.Executor; + import org.apache.rocketmq.client.ClientConfig; import org.apache.rocketmq.client.QueryResult; import org.apache.rocketmq.client.exception.MQBrokerException; @@ -100,6 +102,11 @@ public void send(Message msg, SendCallback sendCallback, long timeout) this.defaultMQProducerImpl.send(msg, sendCallback, timeout); } + @Override + public SendFuture send(Message message, Executor executor, long timeout) throws MQClientException, RemotingException { + return defaultMQProducerImpl.send(message, executor, timeout); + } + @Override public void sendOneway(Message msg) throws MQClientException, RemotingException, InterruptedException { this.defaultMQProducerImpl.sendOneway(msg); diff --git a/client/src/main/java/org/apache/rocketmq/client/producer/MQProducer.java b/client/src/main/java/org/apache/rocketmq/client/producer/MQProducer.java index 9fc7586113a..008d7767d0a 100644 --- a/client/src/main/java/org/apache/rocketmq/client/producer/MQProducer.java +++ b/client/src/main/java/org/apache/rocketmq/client/producer/MQProducer.java @@ -17,6 +17,8 @@ package org.apache.rocketmq.client.producer; import java.util.List; +import java.util.concurrent.Executor; + import org.apache.rocketmq.client.MQAdmin; import org.apache.rocketmq.client.exception.MQBrokerException; import org.apache.rocketmq.client.exception.MQClientException; @@ -43,6 +45,8 @@ void send(final Message msg, final SendCallback sendCallback) throws MQClientExc void send(final Message msg, final SendCallback sendCallback, final long timeout) throws MQClientException, RemotingException, InterruptedException; + SendFuture send(Message message, Executor executor, long timeout) throws MQClientException, RemotingException, InterruptedException; + void sendOneway(final Message msg) throws MQClientException, RemotingException, InterruptedException; diff --git a/client/src/main/java/org/apache/rocketmq/client/producer/SendFuture.java b/client/src/main/java/org/apache/rocketmq/client/producer/SendFuture.java new file mode 100644 index 00000000000..549979b4a98 --- /dev/null +++ b/client/src/main/java/org/apache/rocketmq/client/producer/SendFuture.java @@ -0,0 +1,21 @@ +package org.apache.rocketmq.client.producer; + +import java.util.concurrent.Future; + +public interface SendFuture extends Future { + + /** + * Returns the cause resulting in sending error + */ + Throwable getCause(); + + /** + * Returns {@code true} if this task completed. + */ + @Override + boolean isDone(); + + SendFuture addCallback(SendCallback callback); + SendFuture removeCallback(SendCallback callback); + +} diff --git a/client/src/test/java/org/apache/rocketmq/client/TestSendPromise.java b/client/src/test/java/org/apache/rocketmq/client/TestSendPromise.java new file mode 100644 index 00000000000..fddfe38c57f --- /dev/null +++ b/client/src/test/java/org/apache/rocketmq/client/TestSendPromise.java @@ -0,0 +1,65 @@ +package org.apache.rocketmq.client; + +import org.apache.rocketmq.client.impl.producer.DefaultSendPromise; +import org.apache.rocketmq.client.producer.SendCallback; +import org.apache.rocketmq.client.producer.SendResult; +import org.junit.Assert; +import org.junit.Test; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; + +public class TestSendPromise { + + @Test + public void testBasicOperations() throws Exception { + ExecutorService executor = Executors.newSingleThreadExecutor(); + + DefaultSendPromise promise = new DefaultSendPromise(executor); + + final CountDownLatch latch = new CountDownLatch(2); + final CountDownLatch latch1 = new CountDownLatch(2); + SendCallback callback = new SendCallback() { + @Override + public void onSuccess(SendResult sendResult) { + latch.countDown(); + } + + @Override + public void onException(Throwable e) { + latch1.countDown(); + } + }; + SendCallback callback2 = new SendCallback() { + @Override + public void onSuccess(SendResult sendResult) { + latch.countDown(); + } + + @Override + public void onException(Throwable e) { + latch1.countDown(); + } + }; + promise.addCallback(callback).addCallback(callback2); + + SendResult result = new SendResult(); + promise.complete(result); + Assert.assertTrue(latch.await(5000, TimeUnit.MILLISECONDS)); + Assert.assertEquals(promise.get(), result); + Assert.assertEquals(promise.get(1, TimeUnit.SECONDS), result); + + DefaultSendPromise promise1 = new DefaultSendPromise(executor); + Exception cause = new Exception(); + promise1.report(cause); + promise1.addCallback(callback).addCallback(callback2); + + Assert.assertEquals(cause, promise1.getCause()); + + Assert.assertTrue(latch1.await(5000, TimeUnit.MILLISECONDS)); + + executor.shutdown(); + } +} From 0ddbc4471a6ed1305bade5bb5eb58c8182066990 Mon Sep 17 00:00:00 2001 From: thisisbaozi Date: Sun, 1 Jan 2017 11:39:44 +0800 Subject: [PATCH 2/6] [ROCKETMQ-26]add a new way to send message in asynchronous --- .../impl/producer/DefaultSendPromise.java | 27 ++++++++++++++++--- .../client/impl/producer/SendPromise.java | 16 +++++++++++ .../rocketmq/client/producer/SendFuture.java | 16 +++++++++++ .../rocketmq/client/TestSendPromise.java | 16 +++++++++++ 4 files changed, 72 insertions(+), 3 deletions(-) diff --git a/client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultSendPromise.java b/client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultSendPromise.java index 8c55a1370c0..aced037399b 100644 --- a/client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultSendPromise.java +++ b/client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultSendPromise.java @@ -1,5 +1,27 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ package org.apache.rocketmq.client.impl.producer; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Executor; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import org.apache.rocketmq.client.log.ClientLogger; import org.apache.rocketmq.client.producer.SendCallback; import org.apache.rocketmq.client.producer.SendResult; @@ -8,7 +30,6 @@ import java.util.ArrayList; import java.util.List; -import java.util.concurrent.*; /** * Created by stone @@ -17,11 +38,11 @@ public class DefaultSendPromise extends CountDownLatch implements SendPromise { private final Logger logger = ClientLogger.getLog(); - private Executor executor; + private final Executor executor; private Object callbacks; private volatile Object result; - public DefaultSendPromise(Executor executor) { + public DefaultSendPromise(final Executor executor) { super(1); if (executor == null) { diff --git a/client/src/main/java/org/apache/rocketmq/client/impl/producer/SendPromise.java b/client/src/main/java/org/apache/rocketmq/client/impl/producer/SendPromise.java index d3a3aa2d0a0..bbd631e6676 100644 --- a/client/src/main/java/org/apache/rocketmq/client/impl/producer/SendPromise.java +++ b/client/src/main/java/org/apache/rocketmq/client/impl/producer/SendPromise.java @@ -1,3 +1,19 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ package org.apache.rocketmq.client.impl.producer; import org.apache.rocketmq.client.producer.SendFuture; diff --git a/client/src/main/java/org/apache/rocketmq/client/producer/SendFuture.java b/client/src/main/java/org/apache/rocketmq/client/producer/SendFuture.java index 549979b4a98..30487c2ba62 100644 --- a/client/src/main/java/org/apache/rocketmq/client/producer/SendFuture.java +++ b/client/src/main/java/org/apache/rocketmq/client/producer/SendFuture.java @@ -1,3 +1,19 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ package org.apache.rocketmq.client.producer; import java.util.concurrent.Future; diff --git a/client/src/test/java/org/apache/rocketmq/client/TestSendPromise.java b/client/src/test/java/org/apache/rocketmq/client/TestSendPromise.java index fddfe38c57f..d41fb9fe4dc 100644 --- a/client/src/test/java/org/apache/rocketmq/client/TestSendPromise.java +++ b/client/src/test/java/org/apache/rocketmq/client/TestSendPromise.java @@ -1,3 +1,19 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ package org.apache.rocketmq.client; import org.apache.rocketmq.client.impl.producer.DefaultSendPromise; From 85eb28f0500bd2c73fefded7d845736b5d3fc46b Mon Sep 17 00:00:00 2001 From: thisisbaozi Date: Sun, 1 Jan 2017 12:01:36 +0800 Subject: [PATCH 3/6] [ROCKETMQ-26]add new way to send messages in asynchronous way --- .../impl/producer/DefaultMQProducerImpl.java | 22 ++++++++++++++++--- .../impl/producer/DefaultSendPromise.java | 4 ++-- .../rocketmq/client/producer/SendFuture.java | 10 --------- 3 files changed, 21 insertions(+), 15 deletions(-) diff --git a/client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultMQProducerImpl.java b/client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultMQProducerImpl.java index 76ad744b21e..18e968f1a23 100644 --- a/client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultMQProducerImpl.java +++ b/client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultMQProducerImpl.java @@ -24,8 +24,14 @@ import java.util.List; import java.util.Random; import java.util.Set; -import java.util.concurrent.*; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.Executor; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; import org.apache.rocketmq.client.QueryResult; import org.apache.rocketmq.client.Validators; import org.apache.rocketmq.client.common.ClientErrorCode; @@ -40,7 +46,17 @@ import org.apache.rocketmq.client.impl.factory.MQClientInstance; import org.apache.rocketmq.client.latency.MQFaultStrategy; import org.apache.rocketmq.client.log.ClientLogger; -import org.apache.rocketmq.client.producer.*; +import org.apache.rocketmq.client.producer.DefaultMQProducer; +import org.apache.rocketmq.client.producer.LocalTransactionExecuter; +import org.apache.rocketmq.client.producer.LocalTransactionState; +import org.apache.rocketmq.client.producer.MessageQueueSelector; +import org.apache.rocketmq.client.producer.SendCallback; +import org.apache.rocketmq.client.producer.SendFuture; +import org.apache.rocketmq.client.producer.SendResult; +import org.apache.rocketmq.client.producer.SendStatus; +import org.apache.rocketmq.client.producer.TransactionCheckListener; +import org.apache.rocketmq.client.producer.TransactionMQProducer; +import org.apache.rocketmq.client.producer.TransactionSendResult; import org.apache.rocketmq.common.MixAll; import org.apache.rocketmq.common.ServiceState; import org.apache.rocketmq.common.UtilAll; @@ -410,7 +426,7 @@ public SendFuture send(Message message, Executor executor, long timeout) throws try { this.sendDefaultImpl(message, CommunicationMode.ASYNC, null, timeout, promise); } catch (MQBrokerException e) { - throw new MQClientException("unknownn exception", e); + throw new MQClientException("unknown exception", e); } catch (InterruptedException e) { // ignore } diff --git a/client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultSendPromise.java b/client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultSendPromise.java index aced037399b..aaa13f29252 100644 --- a/client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultSendPromise.java +++ b/client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultSendPromise.java @@ -24,8 +24,8 @@ import java.util.concurrent.TimeoutException; import org.apache.rocketmq.client.log.ClientLogger; import org.apache.rocketmq.client.producer.SendCallback; -import org.apache.rocketmq.client.producer.SendResult; import org.apache.rocketmq.client.producer.SendFuture; +import org.apache.rocketmq.client.producer.SendResult; import org.slf4j.Logger; import java.util.ArrayList; @@ -172,7 +172,7 @@ public SendFuture removeCallback(SendCallback listener) { @SuppressWarnings("unchecked") public void invokeCallbacks() { if (callbacks == null) { - return ; + return; } if (callbacks instanceof List) { diff --git a/client/src/main/java/org/apache/rocketmq/client/producer/SendFuture.java b/client/src/main/java/org/apache/rocketmq/client/producer/SendFuture.java index 30487c2ba62..e3ae55d6eda 100644 --- a/client/src/main/java/org/apache/rocketmq/client/producer/SendFuture.java +++ b/client/src/main/java/org/apache/rocketmq/client/producer/SendFuture.java @@ -20,18 +20,8 @@ public interface SendFuture extends Future { - /** - * Returns the cause resulting in sending error - */ Throwable getCause(); - /** - * Returns {@code true} if this task completed. - */ - @Override - boolean isDone(); - SendFuture addCallback(SendCallback callback); SendFuture removeCallback(SendCallback callback); - } From 1a11b655deb8e7c907e70fc367cc1d1dfa72d149 Mon Sep 17 00:00:00 2001 From: thisisbaozi Date: Sun, 1 Jan 2017 18:02:13 +0800 Subject: [PATCH 4/6] [ROCKETMQ-26]remove author info and add new ut --- .../impl/producer/DefaultSendPromise.java | 3 -- .../rocketmq/client/TestSendPromise.java | 45 +++++++++++++++++++ 2 files changed, 45 insertions(+), 3 deletions(-) diff --git a/client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultSendPromise.java b/client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultSendPromise.java index aaa13f29252..32eedca7859 100644 --- a/client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultSendPromise.java +++ b/client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultSendPromise.java @@ -31,9 +31,6 @@ import java.util.ArrayList; import java.util.List; -/** - * Created by stone - */ public class DefaultSendPromise extends CountDownLatch implements SendPromise { private final Logger logger = ClientLogger.getLog(); diff --git a/client/src/test/java/org/apache/rocketmq/client/TestSendPromise.java b/client/src/test/java/org/apache/rocketmq/client/TestSendPromise.java index d41fb9fe4dc..da77c92e6e8 100644 --- a/client/src/test/java/org/apache/rocketmq/client/TestSendPromise.java +++ b/client/src/test/java/org/apache/rocketmq/client/TestSendPromise.java @@ -29,6 +29,51 @@ public class TestSendPromise { + @Test + public void testAddAndRemoveCallbacks() throws Exception { + ExecutorService executor = Executors.newSingleThreadExecutor(); + + DefaultSendPromise promise = new DefaultSendPromise(executor); + final CountDownLatch latch = new CountDownLatch(1); + SendCallback callback = new SendCallback() { + @Override + public void onSuccess(SendResult sendResult) { + latch.countDown(); + } + + @Override + public void onException(Throwable e) { + latch.countDown(); + } + }; + SendCallback callback1 = new SendCallback() { + @Override + public void onSuccess(SendResult sendResult) { + Assert.fail(); + } + + @Override + public void onException(Throwable e) { + Assert.fail(); + } + }; + + promise.addCallback(callback); + promise.removeCallback(callback); + promise.complete(new SendResult()); + Assert.assertEquals(latch.getCount(), 1); + + promise = new DefaultSendPromise(executor); + promise.addCallback(callback); + promise.addCallback(callback1); + promise.removeCallback(callback1); + promise.complete(new SendResult()); + Assert.assertTrue(latch.await(1000, TimeUnit.SECONDS)); + Assert.assertTrue(promise.isDone()); + + executor.shutdown(); + } + @Test public void testBasicOperations() throws Exception { ExecutorService executor = Executors.newSingleThreadExecutor(); From 16ef61a0513a62c0851d4dfa6c2542164c019284 Mon Sep 17 00:00:00 2001 From: thisisbaozi Date: Tue, 3 Jan 2017 16:37:11 +0800 Subject: [PATCH 5/6] [ROCKETMQ-26]add usage for new async API --- .../example/simple/AsyncProducer.java | 49 +++++++++++++++++++ 1 file changed, 49 insertions(+) diff --git a/example/src/main/java/org/apache/rocketmq/example/simple/AsyncProducer.java b/example/src/main/java/org/apache/rocketmq/example/simple/AsyncProducer.java index 038ea65ad09..6b0d02edb45 100644 --- a/example/src/main/java/org/apache/rocketmq/example/simple/AsyncProducer.java +++ b/example/src/main/java/org/apache/rocketmq/example/simple/AsyncProducer.java @@ -17,12 +17,16 @@ package org.apache.rocketmq.example.simple; import java.io.UnsupportedEncodingException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; import org.apache.rocketmq.client.exception.MQClientException; import org.apache.rocketmq.client.producer.DefaultMQProducer; import org.apache.rocketmq.client.producer.SendCallback; +import org.apache.rocketmq.client.producer.SendFuture; import org.apache.rocketmq.client.producer.SendResult; import org.apache.rocketmq.common.message.Message; import org.apache.rocketmq.remoting.common.RemotingHelper; +import org.apache.rocketmq.remoting.exception.RemotingException; public class AsyncProducer { public static void main(String[] args) throws MQClientException, InterruptedException, UnsupportedEncodingException { @@ -54,6 +58,51 @@ public void onException(Throwable e) { e.printStackTrace(); } } + + /** + * Another way to send messages in async, there are some advantages for this way: + * + * 0. Supports multiple callbacks + * 1. Executes the callback(the callback might take too long) in exclusive thread pool + * 2. Converts to blocking mode(by invoking {@link SendFuture#get()}) + * + * It's a more efficient mechanism(higher throughput) if your callback will + * take a short period of time(eg. blocking on I/O) + */ + Message message = new Message(); + message.setTopic("your topic identifier"); + message.setBody("hello, RocketMQ!".getBytes()); + + ExecutorService executor = Executors.newSingleThreadExecutor(); + + try { + SendFuture future = producer.send(message, executor, 1000); + future.addCallback(new SendCallback() { + @Override + public void onSuccess(SendResult sendResult) { + // do sth + } + + @Override + public void onException(Throwable e) { + // do sth + } + }).addCallback(new SendCallback() { + @Override + public void onSuccess(SendResult sendResult) { + // do sth + } + + @Override + public void onException(Throwable e) { + // do sth + } + }); + } catch (RemotingException e) { + // something wrong + } + + executor.shutdown(); producer.shutdown(); } } From d2b3ff5328bd8fd49d925133dba3292293040d89 Mon Sep 17 00:00:00 2001 From: thisisbaozi Date: Tue, 3 Jan 2017 16:50:25 +0800 Subject: [PATCH 6/6] [ROCKETMQ-26]fix code style error! --- .../org/apache/rocketmq/example/simple/AsyncProducer.java | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/example/src/main/java/org/apache/rocketmq/example/simple/AsyncProducer.java b/example/src/main/java/org/apache/rocketmq/example/simple/AsyncProducer.java index 6b0d02edb45..a9348d1e48a 100644 --- a/example/src/main/java/org/apache/rocketmq/example/simple/AsyncProducer.java +++ b/example/src/main/java/org/apache/rocketmq/example/simple/AsyncProducer.java @@ -26,7 +26,6 @@ import org.apache.rocketmq.client.producer.SendResult; import org.apache.rocketmq.common.message.Message; import org.apache.rocketmq.remoting.common.RemotingHelper; -import org.apache.rocketmq.remoting.exception.RemotingException; public class AsyncProducer { public static void main(String[] args) throws MQClientException, InterruptedException, UnsupportedEncodingException { @@ -87,7 +86,8 @@ public void onSuccess(SendResult sendResult) { public void onException(Throwable e) { // do sth } - }).addCallback(new SendCallback() { + }); + future.addCallback(new SendCallback() { @Override public void onSuccess(SendResult sendResult) { // do sth @@ -98,8 +98,8 @@ public void onException(Throwable e) { // do sth } }); - } catch (RemotingException e) { - // something wrong + } catch (Exception e) { + e.printStackTrace(); } executor.shutdown();