dongeforever commented on code in PR #4601:
URL: https://github.com/apache/rocketmq/pull/4601#discussion_r924007007
##########
client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultMQProducerImpl.java:
##########
@@ -491,27 +506,66 @@ public void send(Message msg,
public void send(final Message msg, final SendCallback sendCallback, final
long timeout)
throws MQClientException, RemotingException, InterruptedException {
final long beginStartTime = System.currentTimeMillis();
+ Runnable runnable = new Runnable() {
+ @Override
+ public void run() {
+ long costTime = System.currentTimeMillis() - beginStartTime;
+ if (timeout > costTime) {
+ try {
+ sendDefaultImpl(msg, CommunicationMode.ASYNC,
sendCallback, timeout - costTime);
+ } catch (Exception e) {
+ sendCallback.onException(e);
+ }
+ } else {
+ sendCallback.onException(
+ new RemotingTooMuchRequestException("DEFAULT ASYNC
send call timeout"));
+ }
+ }
+ };
+ executeAsyncMessageSend(runnable, msg, sendCallback, timeout,
beginStartTime);
+ }
+
+ public void executeAsyncMessageSend(Runnable runnable, final Message msg,
final SendCallback sendCallback,
+ final long timeout, final long
beginStartTime)
+ throws MQClientException, InterruptedException {
ExecutorService executor = this.getAsyncSenderExecutor();
+ boolean isEnableBackpressureForAsyncMode =
this.getDefaultMQProducer().isEnableBackpressureForAsyncMode();
+ boolean isSemaphoreAsyncNumAquired = false;
+ boolean isSemaphoreAsyncSizeAquired = false;
+ int msgLen = msg.getBody() == null ? 1 : msg.getBody().length;
+
try {
- executor.submit(new Runnable() {
- @Override
- public void run() {
- long costTime = System.currentTimeMillis() -
beginStartTime;
- if (timeout > costTime) {
- try {
- sendDefaultImpl(msg, CommunicationMode.ASYNC,
sendCallback, timeout - costTime);
- } catch (Exception e) {
- sendCallback.onException(e);
- }
- } else {
- sendCallback.onException(
- new RemotingTooMuchRequestException("DEFAULT ASYNC
send call timeout"));
- }
+ if (isEnableBackpressureForAsyncMode) {
+ long costTime = System.currentTimeMillis() - beginStartTime;
+ isSemaphoreAsyncNumAquired =
semaphoreAsyncSendNum.tryAcquire(timeout - costTime, TimeUnit.MILLISECONDS);
+ if (!isSemaphoreAsyncNumAquired) {
+ sendCallback.onException(
+ new RemotingTooMuchRequestException("send message
tryAcquire semaphoreAsyncNum timeout"));
+ return;
+ }
+ costTime = System.currentTimeMillis() - beginStartTime;
+ isSemaphoreAsyncSizeAquired =
semaphoreAsyncSendSize.tryAcquire(msgLen, timeout - costTime,
TimeUnit.MILLISECONDS);
+ if (!isSemaphoreAsyncSizeAquired) {
+ sendCallback.onException(
+ new RemotingTooMuchRequestException("send message
tryAcquire semaphoreAsyncSize timeout"));
+ return;
}
Review Comment:
what will happen if timeout - costTime is negative?
##########
client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultMQProducerImpl.java:
##########
@@ -142,6 +147,8 @@ public Thread newThread(Runnable r) {
return new Thread(r, "AsyncSenderExecutor_" +
this.threadIndex.incrementAndGet());
}
});
+ semaphoreAsyncSendNum = new
Semaphore(defaultMQProducer.getBackPressureForAsyncSendNum(), true);
+ semaphoreAsyncSendSize = new
Semaphore(defaultMQProducer.getBackPressureForAsyncSendSize(), true);
}
Review Comment:
Maybe we need to do some checks for illegal arguments.
for example:
asyncSendNum cannot be smaller than 10;
asyncSize cannot be smaller than 1M;
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
