Skip to content

[ISSUE #2435] Solve the problem that DefaultMQProducer#request() sends messages and waits for timeout synchronously #4313

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

Merged
merged 4 commits into from
Jun 6, 2022
Merged

[ISSUE #2435] Solve the problem that DefaultMQProducer#request() sends messages and waits for timeout synchronously #4313

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
6dfc7cb
Fix message callback timeout bug
li-xiao-shuang May 13, 2022
d663e35
Clean up code format
li-xiao-shuang May 14, 2022
7765d30
Fix message callback timeout bug
li-xiao-shuang May 14, 2022
9859a99
Merge branch 'develop' into feature_#2435
duhenglucky Jun 6, 2022
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
16 changes: 7 additions & 9 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 @@ -480,14 +480,12 @@ public void send(Message msg,
}

/**
* @deprecated
* It will be removed at 4.4.0 cause for exception handling and the wrong Semantics of timeout. A new one will be
* provided in next version
*
* @param msg
* @param sendCallback
* @param timeout the <code>sendCallback</code> will be invoked at most time
* @throws RejectedExecutionException
* @deprecated It will be removed at 4.4.0 cause for exception handling and the wrong Semantics of timeout. A new one will be
* provided in next version
*/
@Deprecated
public void send(final Message msg, final SendCallback sendCallback, final long timeout)
Expand Down Expand Up @@ -1034,17 +1032,15 @@ public void send(Message msg, MessageQueue mq, SendCallback sendCallback)
}

/**
* @deprecated
* It will be removed at 4.4.0 cause for exception handling and the wrong Semantics of timeout. A new one will be
* provided in next version
*
* @param msg
* @param mq
* @param sendCallback
* @param timeout the <code>sendCallback</code> will be invoked at most time
* @throws MQClientException
* @throws RemotingException
* @throws InterruptedException
* @deprecated It will be removed at 4.4.0 cause for exception handling and the wrong Semantics of timeout. A new one will be
* provided in next version
*/
@Deprecated
public void send(final Message msg, final MessageQueue mq, final SendCallback sendCallback, final long timeout)
Expand Down Expand Up @@ -1367,7 +1363,7 @@ public SendResult send(Message msg,
return this.sendDefaultImpl(msg, CommunicationMode.SYNC, null, timeout);
}

public Message request(Message msg,
public Message request(final Message msg,
long timeout) throws RequestTimeoutException, MQClientException, RemotingException, MQBrokerException, InterruptedException {
long beginTimestamp = System.currentTimeMillis();
prepareSendRequest(msg, timeout);
Expand All @@ -1382,6 +1378,7 @@ public Message request(Message msg,
@Override
public void onSuccess(SendResult sendResult) {
requestResponseFuture.setSendRequestOk(true);
requestResponseFuture.putResponseMessage(msg);
}

@Override
Expand Down Expand Up @@ -1412,6 +1409,7 @@ public void request(Message msg, final RequestCallback requestCallback, long tim
@Override
public void onSuccess(SendResult sendResult) {
requestResponseFuture.setSendRequestOk(true);
requestResponseFuture.executeRequestCallback();
}

@Override
Expand Down
2 changes: 2 additions & 0 deletions example/src/main/java/org/apache/rocketmq/example/rpc/RequestProducer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,8 @@ public static void main(String[] args) throws MQClientException, InterruptedExce
long ttl = 3000;

DefaultMQProducer producer = new DefaultMQProducer(producerGroup);

//You need to set namesrvAddr to the address of the local namesrv
producer.setNamesrvAddr("127.0.0.1:9876");

producer.start();
Expand Down