This is an automated email from the ASF dual-hosted git repository.
yuzhou pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/rocketmq.git
The following commit(s) were added to refs/heads/develop by this push:
new 7f1c14c [ISSUE #3453]Producer cannot shutdown completely (#3454)
7f1c14c is described below
commit 7f1c14c3c15ebd734cc08755b171a0003c9916b9
Author: panzhi <[email protected]>
AuthorDate: Fri Dec 10 10:57:35 2021 +0800
[ISSUE #3453]Producer cannot shutdown completely (#3454)
* [ISSUE #3453]Producer cannot shutdown completely
* [ISSUE #3453]Producer cannot shutdown completely
* [ISSUE #3453]Producer cannot shutdown completely
Co-authored-by: panzhi33 <[email protected]>
---
.../client/impl/ClientRemotingProcessor.java | 6 ++--
.../impl/producer/DefaultMQProducerImpl.java | 40 +++++++++-------------
...stFutureTable.java => RequestFutureHolder.java} | 36 +++++++++++++++----
.../client/producer/DefaultMQProducerTest.java | 6 ++--
4 files changed, 51 insertions(+), 37 deletions(-)
diff --git
a/client/src/main/java/org/apache/rocketmq/client/impl/ClientRemotingProcessor.java
b/client/src/main/java/org/apache/rocketmq/client/impl/ClientRemotingProcessor.java
index a541b2d..b90a5b7 100644
---
a/client/src/main/java/org/apache/rocketmq/client/impl/ClientRemotingProcessor.java
+++
b/client/src/main/java/org/apache/rocketmq/client/impl/ClientRemotingProcessor.java
@@ -25,7 +25,7 @@ import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.impl.factory.MQClientInstance;
import org.apache.rocketmq.client.impl.producer.MQProducerInner;
import org.apache.rocketmq.client.log.ClientLogger;
-import org.apache.rocketmq.client.producer.RequestFutureTable;
+import org.apache.rocketmq.client.producer.RequestFutureHolder;
import org.apache.rocketmq.client.producer.RequestResponseFuture;
import org.apache.rocketmq.common.UtilAll;
import org.apache.rocketmq.common.message.MessageAccessor;
@@ -274,11 +274,11 @@ public class ClientRemotingProcessor extends
AsyncNettyRequestProcessor implemen
private void processReplyMessage(MessageExt replyMsg) {
final String correlationId =
replyMsg.getUserProperty(MessageConst.PROPERTY_CORRELATION_ID);
- final RequestResponseFuture requestResponseFuture =
RequestFutureTable.getRequestFutureTable().get(correlationId);
+ final RequestResponseFuture requestResponseFuture =
RequestFutureHolder.getInstance().getRequestFutureTable().get(correlationId);
if (requestResponseFuture != null) {
requestResponseFuture.putResponseMessage(replyMsg);
- RequestFutureTable.getRequestFutureTable().remove(correlationId);
+
RequestFutureHolder.getInstance().getRequestFutureTable().remove(correlationId);
if (requestResponseFuture.getRequestCallback() != null) {
requestResponseFuture.getRequestCallback().onSuccess(replyMsg);
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 9bbde41..b27117f 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
@@ -28,10 +28,8 @@ import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionException;
-import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
@@ -58,7 +56,7 @@ 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.RequestCallback;
-import org.apache.rocketmq.client.producer.RequestFutureTable;
+import org.apache.rocketmq.client.producer.RequestFutureHolder;
import org.apache.rocketmq.client.producer.RequestResponseFuture;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
@@ -106,12 +104,6 @@ public class DefaultMQProducerImpl implements
MQProducerInner {
private final RPCHook rpcHook;
private final BlockingQueue<Runnable> asyncSenderThreadPoolQueue;
private final ExecutorService defaultAsyncSenderExecutor;
- private final ScheduledExecutorService scheduledExecutorService =
Executors.newSingleThreadScheduledExecutor(new ThreadFactory() {
- @Override
- public Thread newThread(Runnable r) {
- return new Thread(r, "RequestHouseKeepingService");
- }
- });
protected BlockingQueue<Runnable> checkRequestQueue;
protected ExecutorService checkExecutor;
private ServiceState serviceState = ServiceState.CREATE_JUST;
@@ -236,12 +228,12 @@ public class DefaultMQProducerImpl implements
MQProducerInner {
}
private void startScheduledTask() {
- if (RequestFutureTable.getProducerNum().incrementAndGet() == 1) {
- this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
+ if
(RequestFutureHolder.getInstance().getProducerNum().incrementAndGet() == 1) {
+
RequestFutureHolder.getInstance().getScheduledExecutorService().scheduleAtFixedRate(new
Runnable() {
@Override
public void run() {
try {
- RequestFutureTable.scanExpiredRequest();
+ RequestFutureHolder.getInstance().scanExpiredRequest();
} catch (Throwable e) {
log.error("scan RequestFutureTable exception", e);
}
@@ -277,8 +269,8 @@ public class DefaultMQProducerImpl implements
MQProducerInner {
if (shutdownFactory) {
this.mQClientFactory.shutdown();
}
- if (RequestFutureTable.getProducerNum().decrementAndGet() ==
0) {
- scheduledExecutorService.shutdown();
+ if
(RequestFutureHolder.getInstance().getProducerNum().decrementAndGet() == 0) {
+
RequestFutureHolder.getInstance().getScheduledExecutorService().shutdown();
}
log.info("the producer [{}] shutdown OK",
this.defaultMQProducer.getProducerGroup());
this.serviceState = ServiceState.SHUTDOWN_ALREADY;
@@ -1380,7 +1372,7 @@ public class DefaultMQProducerImpl implements
MQProducerInner {
try {
final RequestResponseFuture requestResponseFuture = new
RequestResponseFuture(correlationId, timeout, null);
- RequestFutureTable.getRequestFutureTable().put(correlationId,
requestResponseFuture);
+
RequestFutureHolder.getInstance().getRequestFutureTable().put(correlationId,
requestResponseFuture);
long cost = System.currentTimeMillis() - beginTimestamp;
this.sendDefaultImpl(msg, CommunicationMode.ASYNC, new
SendCallback() {
@@ -1399,7 +1391,7 @@ public class DefaultMQProducerImpl implements
MQProducerInner {
return waitResponse(msg, timeout, requestResponseFuture, cost);
} finally {
- RequestFutureTable.getRequestFutureTable().remove(correlationId);
+
RequestFutureHolder.getInstance().getRequestFutureTable().remove(correlationId);
}
}
@@ -1410,7 +1402,7 @@ public class DefaultMQProducerImpl implements
MQProducerInner {
final String correlationId =
msg.getProperty(MessageConst.PROPERTY_CORRELATION_ID);
final RequestResponseFuture requestResponseFuture = new
RequestResponseFuture(correlationId, timeout, requestCallback);
- RequestFutureTable.getRequestFutureTable().put(correlationId,
requestResponseFuture);
+
RequestFutureHolder.getInstance().getRequestFutureTable().put(correlationId,
requestResponseFuture);
long cost = System.currentTimeMillis() - beginTimestamp;
this.sendDefaultImpl(msg, CommunicationMode.ASYNC, new SendCallback() {
@@ -1436,7 +1428,7 @@ public class DefaultMQProducerImpl implements
MQProducerInner {
try {
final RequestResponseFuture requestResponseFuture = new
RequestResponseFuture(correlationId, timeout, null);
- RequestFutureTable.getRequestFutureTable().put(correlationId,
requestResponseFuture);
+
RequestFutureHolder.getInstance().getRequestFutureTable().put(correlationId,
requestResponseFuture);
long cost = System.currentTimeMillis() - beginTimestamp;
this.sendSelectImpl(msg, selector, arg, CommunicationMode.ASYNC,
new SendCallback() {
@@ -1455,7 +1447,7 @@ public class DefaultMQProducerImpl implements
MQProducerInner {
return waitResponse(msg, timeout, requestResponseFuture, cost);
} finally {
- RequestFutureTable.getRequestFutureTable().remove(correlationId);
+
RequestFutureHolder.getInstance().getRequestFutureTable().remove(correlationId);
}
}
@@ -1467,7 +1459,7 @@ public class DefaultMQProducerImpl implements
MQProducerInner {
final String correlationId =
msg.getProperty(MessageConst.PROPERTY_CORRELATION_ID);
final RequestResponseFuture requestResponseFuture = new
RequestResponseFuture(correlationId, timeout, requestCallback);
- RequestFutureTable.getRequestFutureTable().put(correlationId,
requestResponseFuture);
+
RequestFutureHolder.getInstance().getRequestFutureTable().put(correlationId,
requestResponseFuture);
long cost = System.currentTimeMillis() - beginTimestamp;
this.sendSelectImpl(msg, selector, arg, CommunicationMode.ASYNC, new
SendCallback() {
@@ -1493,7 +1485,7 @@ public class DefaultMQProducerImpl implements
MQProducerInner {
try {
final RequestResponseFuture requestResponseFuture = new
RequestResponseFuture(correlationId, timeout, null);
- RequestFutureTable.getRequestFutureTable().put(correlationId,
requestResponseFuture);
+
RequestFutureHolder.getInstance().getRequestFutureTable().put(correlationId,
requestResponseFuture);
long cost = System.currentTimeMillis() - beginTimestamp;
this.sendKernelImpl(msg, mq, CommunicationMode.ASYNC, new
SendCallback() {
@@ -1512,7 +1504,7 @@ public class DefaultMQProducerImpl implements
MQProducerInner {
return waitResponse(msg, timeout, requestResponseFuture, cost);
} finally {
- RequestFutureTable.getRequestFutureTable().remove(correlationId);
+
RequestFutureHolder.getInstance().getRequestFutureTable().remove(correlationId);
}
}
@@ -1536,7 +1528,7 @@ public class DefaultMQProducerImpl implements
MQProducerInner {
final String correlationId =
msg.getProperty(MessageConst.PROPERTY_CORRELATION_ID);
final RequestResponseFuture requestResponseFuture = new
RequestResponseFuture(correlationId, timeout, requestCallback);
- RequestFutureTable.getRequestFutureTable().put(correlationId,
requestResponseFuture);
+
RequestFutureHolder.getInstance().getRequestFutureTable().put(correlationId,
requestResponseFuture);
long cost = System.currentTimeMillis() - beginTimestamp;
this.sendKernelImpl(msg, mq, CommunicationMode.ASYNC, new
SendCallback() {
@@ -1554,7 +1546,7 @@ public class DefaultMQProducerImpl implements
MQProducerInner {
}
private void requestFail(final String correlationId) {
- RequestResponseFuture responseFuture =
RequestFutureTable.getRequestFutureTable().remove(correlationId);
+ RequestResponseFuture responseFuture =
RequestFutureHolder.getInstance().getRequestFutureTable().remove(correlationId);
if (responseFuture != null) {
responseFuture.setSendRequestOk(false);
responseFuture.putResponseMessage(null);
diff --git
a/client/src/main/java/org/apache/rocketmq/client/producer/RequestFutureTable.java
b/client/src/main/java/org/apache/rocketmq/client/producer/RequestFutureHolder.java
similarity index 66%
rename from
client/src/main/java/org/apache/rocketmq/client/producer/RequestFutureTable.java
rename to
client/src/main/java/org/apache/rocketmq/client/producer/RequestFutureHolder.java
index 9bdc3da..24b3a90 100644
---
a/client/src/main/java/org/apache/rocketmq/client/producer/RequestFutureTable.java
+++
b/client/src/main/java/org/apache/rocketmq/client/producer/RequestFutureHolder.java
@@ -22,6 +22,9 @@ import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.rocketmq.client.common.ClientErrorCode;
@@ -29,16 +32,24 @@ import
org.apache.rocketmq.client.exception.RequestTimeoutException;
import org.apache.rocketmq.client.log.ClientLogger;
import org.apache.rocketmq.logging.InternalLogger;
-public class RequestFutureTable {
+public class RequestFutureHolder {
private static InternalLogger log = ClientLogger.getLog();
- private static ConcurrentHashMap<String, RequestResponseFuture>
requestFutureTable = new ConcurrentHashMap<String, RequestResponseFuture>();
- private static final AtomicInteger PRODUCER_NUM = new AtomicInteger(0);
+ private static final RequestFutureHolder INSTANCE = new
RequestFutureHolder();
+ private ConcurrentHashMap<String, RequestResponseFuture>
requestFutureTable = new ConcurrentHashMap<String, RequestResponseFuture>();
+ private final AtomicInteger producerNum = new AtomicInteger(0);
+ private final ScheduledExecutorService scheduledExecutorService = Executors
+ .newSingleThreadScheduledExecutor(new ThreadFactory() {
+ @Override
+ public Thread newThread(Runnable r) {
+ return new Thread(r, "RequestHouseKeepingService");
+ }
+ });
- public static ConcurrentHashMap<String, RequestResponseFuture>
getRequestFutureTable() {
+ public ConcurrentHashMap<String, RequestResponseFuture>
getRequestFutureTable() {
return requestFutureTable;
}
- public static void scanExpiredRequest() {
+ public void scanExpiredRequest() {
final List<RequestResponseFuture> rfList = new
LinkedList<RequestResponseFuture>();
Iterator<Map.Entry<String, RequestResponseFuture>> it =
requestFutureTable.entrySet().iterator();
while (it.hasNext()) {
@@ -63,7 +74,18 @@ public class RequestFutureTable {
}
}
- public static AtomicInteger getProducerNum() {
- return PRODUCER_NUM;
+ private RequestFutureHolder() {
+ }
+
+ public AtomicInteger getProducerNum() {
+ return producerNum;
+ }
+
+ public ScheduledExecutorService getScheduledExecutorService() {
+ return scheduledExecutorService;
+ }
+
+ public static RequestFutureHolder getInstance() {
+ return INSTANCE;
}
}
diff --git
a/client/src/test/java/org/apache/rocketmq/client/producer/DefaultMQProducerTest.java
b/client/src/test/java/org/apache/rocketmq/client/producer/DefaultMQProducerTest.java
index e161d4a..a890e72 100644
---
a/client/src/test/java/org/apache/rocketmq/client/producer/DefaultMQProducerTest.java
+++
b/client/src/test/java/org/apache/rocketmq/client/producer/DefaultMQProducerTest.java
@@ -369,7 +369,7 @@ public class DefaultMQProducerTest {
final AtomicBoolean finish = new AtomicBoolean(false);
new Thread(new Runnable() {
@Override public void run() {
- ConcurrentHashMap<String, RequestResponseFuture> responseMap =
RequestFutureTable.getRequestFutureTable();
+ ConcurrentHashMap<String, RequestResponseFuture> responseMap =
RequestFutureHolder.getInstance().getRequestFutureTable();
assertThat(responseMap).isNotNull();
while (!finish.get()) {
try {
@@ -411,7 +411,7 @@ public class DefaultMQProducerTest {
}
};
producer.request(message, requestCallback, 3 * 1000L);
- ConcurrentHashMap<String, RequestResponseFuture> responseMap =
RequestFutureTable.getRequestFutureTable();
+ ConcurrentHashMap<String, RequestResponseFuture> responseMap =
RequestFutureHolder.getInstance().getRequestFutureTable();
assertThat(responseMap).isNotNull();
for (Map.Entry<String, RequestResponseFuture> entry :
responseMap.entrySet()) {
RequestResponseFuture future = entry.getValue();
@@ -447,7 +447,7 @@ public class DefaultMQProducerTest {
producer.request(message, requestCallback, 3 * 1000L);
failBecauseExceptionWasNotThrown(Exception.class);
} catch (Exception e) {
- ConcurrentHashMap<String, RequestResponseFuture> responseMap =
RequestFutureTable.getRequestFutureTable();
+ ConcurrentHashMap<String, RequestResponseFuture> responseMap =
RequestFutureHolder.getInstance().getRequestFutureTable();
assertThat(responseMap).isNotNull();
for (Map.Entry<String, RequestResponseFuture> entry :
responseMap.entrySet()) {
RequestResponseFuture future = entry.getValue();
