Skip to content

[ROCKETMQ-26]add new way to send messages in asynchronous way #26

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
thisisbaozi wants to merge 6 commits into from
Closed

[ROCKETMQ-26]add new way to send messages in asynchronous way #26

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
67545d9
add new way to send message in asynchronous way
thisisbaozi Jan 1, 2017
0ddbc44
[ROCKETMQ-26]add a new way to send message in asynchronous
thisisbaozi Jan 1, 2017
85eb28f
[ROCKETMQ-26]add new way to send messages in asynchronous way
thisisbaozi Jan 1, 2017
1a11b65
[ROCKETMQ-26]remove author info and add new ut
thisisbaozi Jan 1, 2017
16ef61a
[ROCKETMQ-26]add usage for new async API
thisisbaozi Jan 3, 2017
d2b3ff5
[ROCKETMQ-26]fix code style error!
thisisbaozi Jan 3, 2017
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
65 changes: 46 additions & 19 deletions client/src/main/java/org/apache/rocketmq/client/impl/MQClientAPIImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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(//
Expand All @@ -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;
Expand All @@ -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);
Expand All @@ -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, //
Expand All @@ -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() {
Expand Down Expand Up @@ -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);
}
}
}
Expand All @@ -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();
Expand All @@ -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);
}
}

Expand Down
40 changes: 29 additions & 11 deletions client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultMQProducerImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,8 +24,10 @@
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.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
Expand All @@ -49,6 +51,7 @@
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;
Expand Down Expand Up @@ -412,12 +415,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("unknown exception", e);
} catch (InterruptedException e) {
// ignore
}

return promise;
}

public MessageQueue selectOneMessageQueue(final TopicPublishInfo tpInfo, final String lastBrokerName) {
return this.mqFaultStrategy.selectOneMessageQueue(tpInfo, lastBrokerName);
}
Expand All @@ -430,7 +446,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);
Expand All @@ -455,7 +472,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) {
Expand Down Expand Up @@ -582,7 +599,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());
Expand Down Expand Up @@ -680,7 +697,8 @@ private SendResult sendKernelImpl(final Message msg, //
topicPublishInfo, // 8
this.mQClientFactory, // 9
this.defaultMQProducer.getRetryTimesWhenSendAsyncFailed(), // 10
context, //
context,
promise,
this);
break;
case ONEWAY:
Expand Down Expand Up @@ -801,7 +819,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);
}
Expand All @@ -824,7 +842,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);
}

/**
Expand All @@ -845,7 +863,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);
}
Expand All @@ -859,7 +877,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);
}
Expand Down Expand Up @@ -898,7 +916,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);
}
Expand Down Expand Up @@ -1046,7 +1064,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<String, TopicPublishInfo> getTopicPublishInfoTable() {
Expand Down
Loading