shengminw commented on code in PR #4601:
URL: https://github.com/apache/rocketmq/pull/4601#discussion_r924113814
##########
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:
In tryAquire, if timeout is negative, it will return false. I also added a
conditional judgment to ensure that `timeout-costTime` is positive
--
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]
