Skip to content

Commit 9c3e7fe

Browse files
qianye1001qianye1001
authored
[ISSUE #9847] Reduce lock contention on the HandleData object to prevent threads from hanging (#9848)
1 parent 6c798d6 commit 9c3e7fe

3 files changed

Lines changed: 58 additions & 18 deletions

File tree

‎proxy/src/main/java/org/apache/rocketmq/proxy/common/ReceiptHandleGroup.java‎

Lines changed: 15 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -200,6 +200,14 @@ public boolean isEmpty() {
200200
return this.receiptHandleMap.isEmpty();
201201
}
202202

203+
public long getHandleNum() {
204+
long handleNum = 0L;
205+
for (Map.Entry<String, Map<HandleKey, HandleData>> entry : receiptHandleMap.entrySet()) {
206+
handleNum += entry.getValue().size();
207+
}
208+
return handleNum;
209+
}
210+
203211
public MessageReceiptHandle get(String msgID, String handle) {
204212
Map<HandleKey, HandleData> handleMap = this.receiptHandleMap.get(msgID);
205213
if (handleMap == null) {
@@ -268,13 +276,18 @@ public MessageReceiptHandle removeOne(String msgID) {
268276

269277
public void computeIfPresent(String msgID, String handle,
270278
Function<MessageReceiptHandle, CompletableFuture<MessageReceiptHandle>> function) {
279+
long timeout = ConfigurationManager.getProxyConfig().getLockTimeoutMsInHandleGroup();
280+
computeIfPresent(msgID, handle, function, timeout);
281+
}
282+
283+
public void computeIfPresent(String msgID, String handle,
284+
Function<MessageReceiptHandle, CompletableFuture<MessageReceiptHandle>> function, long lockTimeout) {
271285
Map<HandleKey, HandleData> handleMap = this.receiptHandleMap.get(msgID);
272286
if (handleMap == null) {
273287
return;
274288
}
275-
long timeout = ConfigurationManager.getProxyConfig().getLockTimeoutMsInHandleGroup();
276289
handleMap.computeIfPresent(new HandleKey(handle), (handleKey, handleData) -> {
277-
Long lockTimeMs = handleData.lock(timeout);
290+
Long lockTimeMs = handleData.lock(lockTimeout);
278291
if (lockTimeMs == null) {
279292
throw new ProxyException(ProxyExceptionCode.INTERNAL_SERVER_ERROR, "try to compute failed");
280293
}

‎proxy/src/main/java/org/apache/rocketmq/proxy/config/ProxyConfig.java‎

Lines changed: 19 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -17,17 +17,6 @@
1717

1818
package org.apache.rocketmq.proxy.config;
1919

20-
import org.apache.commons.lang3.StringUtils;
21-
import org.apache.rocketmq.common.MixAll;
22-
import org.apache.rocketmq.common.constant.LoggerName;
23-
import org.apache.rocketmq.common.metrics.MetricsExporterType;
24-
import org.apache.rocketmq.common.utils.NetworkUtil;
25-
import org.apache.rocketmq.logging.org.slf4j.Logger;
26-
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
27-
import org.apache.rocketmq.proxy.ProxyMode;
28-
import org.apache.rocketmq.proxy.common.ProxyException;
29-
import org.apache.rocketmq.proxy.common.ProxyExceptionCode;
30-
3120
import java.net.InetAddress;
3221
import java.net.UnknownHostException;
3322
import java.time.Duration;
@@ -38,6 +27,16 @@
3827
import java.util.concurrent.ConcurrentSkipListMap;
3928
import java.util.concurrent.TimeUnit;
4029
import java.util.stream.Collectors;
30+
import org.apache.commons.lang3.StringUtils;
31+
import org.apache.rocketmq.common.MixAll;
32+
import org.apache.rocketmq.common.constant.LoggerName;
33+
import org.apache.rocketmq.common.metrics.MetricsExporterType;
34+
import org.apache.rocketmq.common.utils.NetworkUtil;
35+
import org.apache.rocketmq.logging.org.slf4j.Logger;
36+
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
37+
import org.apache.rocketmq.proxy.ProxyMode;
38+
import org.apache.rocketmq.proxy.common.ProxyException;
39+
import org.apache.rocketmq.proxy.common.ProxyExceptionCode;
4140

4241
public class ProxyConfig implements ConfigFile {
4342
private final static Logger log = LoggerFactory.getLogger(LoggerName.PROXY_LOGGER_NAME);
@@ -204,6 +203,7 @@ public class ProxyConfig implements ConfigFile {
204203
private long renewAheadTimeMillis = TimeUnit.SECONDS.toMillis(10);
205204
private long renewMaxTimeMillis = TimeUnit.HOURS.toMillis(3);
206205
private long renewSchedulePeriodMillis = TimeUnit.SECONDS.toMillis(5);
206+
private int returnHandleGroupThreadPoolNums = 2;
207207

208208
private boolean enableAclRpcHookForClusterMode = false;
209209

@@ -1537,4 +1537,12 @@ public boolean isEnableMessageBodyEmptyCheck() {
15371537
public void setEnableMessageBodyEmptyCheck(boolean enableMessageBodyEmptyCheck) {
15381538
this.enableMessageBodyEmptyCheck = enableMessageBodyEmptyCheck;
15391539
}
1540+
1541+
public int getReturnHandleGroupThreadPoolNums() {
1542+
return returnHandleGroupThreadPoolNums;
1543+
}
1544+
1545+
public void setReturnHandleGroupThreadPoolNums(int returnHandleGroupThreadPoolNums) {
1546+
this.returnHandleGroupThreadPoolNums = returnHandleGroupThreadPoolNums;
1547+
}
15401548
}

‎proxy/src/main/java/org/apache/rocketmq/proxy/service/receipt/DefaultReceiptHandleManager.java‎

Lines changed: 24 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -40,6 +40,7 @@
4040
import org.apache.rocketmq.common.thread.ThreadPoolMonitor;
4141
import org.apache.rocketmq.common.utils.AbstractStartAndShutdown;
4242
import org.apache.rocketmq.common.utils.ConcurrentHashMapUtils;
43+
import org.apache.rocketmq.common.utils.ExceptionUtils;
4344
import org.apache.rocketmq.common.utils.StartAndShutdown;
4445
import org.apache.rocketmq.common.utils.ThreadUtils;
4546
import org.apache.rocketmq.logging.org.slf4j.Logger;
@@ -53,7 +54,6 @@
5354
import org.apache.rocketmq.proxy.common.RenewEvent;
5455
import org.apache.rocketmq.proxy.common.RenewStrategyPolicy;
5556
import org.apache.rocketmq.proxy.common.channel.ChannelHelper;
56-
import org.apache.rocketmq.common.utils.ExceptionUtils;
5757
import org.apache.rocketmq.proxy.config.ConfigurationManager;
5858
import org.apache.rocketmq.proxy.config.ProxyConfig;
5959
import org.apache.rocketmq.proxy.service.metadata.MetadataService;
@@ -70,6 +70,7 @@ public class DefaultReceiptHandleManager extends AbstractStartAndShutdown implem
7070
protected final ScheduledExecutorService scheduledExecutorService =
7171
ThreadUtils.newSingleThreadScheduledExecutor(new ThreadFactoryImpl("RenewalScheduledThread_"));
7272
protected final ThreadPoolExecutor renewalWorkerService;
73+
protected final ThreadPoolExecutor returnHandleGroupWorkerService;
7374

7475
public DefaultReceiptHandleManager(MetadataService metadataService, ConsumerManager consumerManager, StateEventListener<RenewEvent> eventListener) {
7576
this.metadataService = metadataService;
@@ -83,6 +84,13 @@ public DefaultReceiptHandleManager(MetadataService metadataService, ConsumerMana
8384
"RenewalWorkerThread",
8485
proxyConfig.getRenewThreadPoolQueueCapacity()
8586
);
87+
this.returnHandleGroupWorkerService = ThreadPoolMonitor.createAndMonitor(
88+
proxyConfig.getReturnHandleGroupThreadPoolNums(),
89+
proxyConfig.getReturnHandleGroupThreadPoolNums() * 2,
90+
1, TimeUnit.MINUTES,
91+
"ReturnHandleGroupWorkerThread",
92+
proxyConfig.getRenewThreadPoolQueueCapacity()
93+
);
8694
consumerManager.appendConsumerIdsChangeListener(new ConsumerIdsChangeListener() {
8795
@Override
8896
public void handle(ConsumerGroupEvent event, String group, Object... args) {
@@ -172,7 +180,7 @@ protected void scheduleRenewTask() {
172180

173181
protected void renewMessage(ProxyContext context, ReceiptHandleGroupKey key, ReceiptHandleGroup group, String msgID, String handleStr) {
174182
try {
175-
group.computeIfPresent(msgID, handleStr, messageReceiptHandle -> startRenewMessage(context, key, messageReceiptHandle));
183+
group.computeIfPresent(msgID, handleStr, messageReceiptHandle -> startRenewMessage(context, key, messageReceiptHandle), 0);
176184
} catch (Exception e) {
177185
log.error("error when renew message. msgID:{}, handleStr:{}", msgID, handleStr, e);
178186
}
@@ -237,22 +245,33 @@ protected void clearGroup(ReceiptHandleGroupKey key) {
237245
if (key == null) {
238246
return;
239247
}
240-
ProxyConfig proxyConfig = ConfigurationManager.getProxyConfig();
241248
ReceiptHandleGroup handleGroup = receiptHandleGroupMap.remove(key);
242-
if (handleGroup == null) {
249+
returnHandleGroupWorkerService.submit(() -> returnHandleGroup(key, handleGroup));
250+
}
251+
252+
// There is no longer any waiting for lock, and only the locked handles will be processed immediately,
253+
// while the handles that cannot be acquired will be kept waiting for the next scheduling.
254+
private void returnHandleGroup(ReceiptHandleGroupKey key, ReceiptHandleGroup handleGroup) {
255+
if (handleGroup == null || handleGroup.isEmpty()) {
243256
return;
244257
}
258+
ProxyConfig proxyConfig = ConfigurationManager.getProxyConfig();
245259
handleGroup.scan((msgID, handle, v) -> {
246260
try {
247261
handleGroup.computeIfPresent(msgID, handle, messageReceiptHandle -> {
248262
CompletableFuture<AckResult> future = new CompletableFuture<>();
249263
eventListener.fireEvent(new RenewEvent(key, messageReceiptHandle, proxyConfig.getInvisibleTimeMillisWhenClear(), RenewEvent.EventType.CLEAR_GROUP, future));
250264
return CompletableFuture.completedFuture(null);
251-
});
265+
}, 0);
252266
} catch (Exception e) {
253267
log.error("error when clear handle for group. key:{}", key, e);
254268
}
255269
});
270+
// scheduleRenewTask will trigger cleanup again
271+
if (!handleGroup.isEmpty()) {
272+
log.warn("The handle cannot be completely cleared, the remaining quantity is {}, key:{}", handleGroup.getHandleNum(), key);
273+
receiptHandleGroupMap.putIfAbsent(key, handleGroup);
274+
}
256275
}
257276

258277
protected void clearAllHandle() {

0 commit comments

Comments
 (0)