opensource/rocketmq
3
0
Fork 0
mirror of https://github.com/apache/rocketmq.git synced 2026-09-24 16:04:00 +08:00
Code Issues Packages Projects Releases Wiki Activity

[ISSUE #3624]Fix producer/consumer re-start may fail introduced in #3454 (#3639)

Browse Source 
Fix producer/consumer re-start may fail introduced in #3454

1. start a producer 
2. shutdown producer
3. start the second producer, exception throws
This commit is contained in:
panzhi
2021-12-15 20:00:21 +08:00
committed by GitHub
parent fe98bed001
commit 64ad7908b8
2 changed files with 35 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files
client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultMQProducerImpl.java
+2 -19
View File
@@ -223,25 +223,10 @@ public class DefaultMQProducerImpl implements MQProducerInner {
this.mQClientFactory.sendHeartbeatToAllBrokerWithLock();
this.startScheduledTask();
RequestFutureHolder.getInstance().startScheduledTask(this);
}
private void startScheduledTask() {
if (RequestFutureHolder.getInstance().getProducerNum().incrementAndGet() == 1) {
RequestFutureHolder.getInstance().getScheduledExecutorService().scheduleAtFixedRate(new Runnable() {
@Override
public void run() {
try {
RequestFutureHolder.getInstance().scanExpiredRequest();
} catch (Throwable e) {
log.error("scan RequestFutureTable exception", e);
}
}
}, 1000 * 3, 1000, TimeUnit.MILLISECONDS);
}
}
private void checkConfig() throws MQClientException {
Validators.checkGroup(this.defaultMQProducer.getProducerGroup());
@@ -269,9 +254,7 @@ public class DefaultMQProducerImpl implements MQProducerInner {
if (shutdownFactory) {
this.mQClientFactory.shutdown();
}
if (RequestFutureHolder.getInstance().getProducerNum().decrementAndGet() == 0) {
RequestFutureHolder.getInstance().getScheduledExecutorService().shutdown();
}
RequestFutureHolder.getInstance().shutdown(this);
log.info("the producer [{}] shutdown OK", this.defaultMQProducer.getProducerGroup());
this.serviceState = ServiceState.SHUTDOWN_ALREADY;
break;
client/src/main/java/org/apache/rocketmq/client/producer/RequestFutureHolder.java
+33 -17
View File
@@ -17,39 +17,36 @@
package org.apache.rocketmq.client.producer;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;
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 java.util.concurrent.TimeUnit;
import org.apache.rocketmq.client.common.ClientErrorCode;
import org.apache.rocketmq.client.exception.RequestTimeoutException;
import org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl;
import org.apache.rocketmq.client.log.ClientLogger;
import org.apache.rocketmq.common.ThreadFactoryImpl;
import org.apache.rocketmq.logging.InternalLogger;
public class RequestFutureHolder {
private static InternalLogger log = ClientLogger.getLog();
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");
}
});
private final Set<DefaultMQProducerImpl> producerSet = new HashSet<>();
private ScheduledExecutorService scheduledExecutorService = null;
public ConcurrentHashMap<String, RequestResponseFuture> getRequestFutureTable() {
return requestFutureTable;
}
public void scanExpiredRequest() {
private void scanExpiredRequest() {
final List<RequestResponseFuture> rfList = new LinkedList<RequestResponseFuture>();
Iterator<Map.Entry<String, RequestResponseFuture>> it = requestFutureTable.entrySet().iterator();
while (it.hasNext()) {
@@ -74,16 +71,35 @@ public class RequestFutureHolder {
}
}
private RequestFutureHolder() {
public synchronized void startScheduledTask(DefaultMQProducerImpl producer) {
this.producerSet.add(producer);
if (null == scheduledExecutorService) {
this.scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(new ThreadFactoryImpl("RequestHouseKeepingService"));
this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
@Override
public void run() {
try {
RequestFutureHolder.getInstance().scanExpiredRequest();
} catch (Throwable e) {
log.error("scan RequestFutureTable exception", e);
}
}
}, 1000 * 3, 1000, TimeUnit.MILLISECONDS);
}
}
public AtomicInteger getProducerNum() {
return producerNum;
public synchronized void shutdown(DefaultMQProducerImpl producer) {
this.producerSet.remove(producer);
if (this.producerSet.size() <= 0 && null != this.scheduledExecutorService) {
ScheduledExecutorService executorService = this.scheduledExecutorService;
this.scheduledExecutorService = null;
executorService.shutdown();
}
}
public ScheduledExecutorService getScheduledExecutorService() {
return scheduledExecutorService;
}
private RequestFutureHolder() {}
public static RequestFutureHolder getInstance() {
return INSTANCE;