Skip to content

Commit 0dc8841

Browse files
qianye1001qianye1001
committed
fix
Change-Id: I6ab637429dbdbbb54878dde1d0592cc941d7ce33
1 parent 5be1dd7 commit 0dc8841

3 files changed

Lines changed: 45 additions & 22 deletions

File tree

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

Lines changed: 6 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -121,14 +121,12 @@ public Long lock(long timeoutMs) {
121121
} else {
122122
// if the lock is expired, can be acquired again
123123
long expiredTimeMs = ConfigurationManager.getProxyConfig().getLockTimeoutMsInHandleGroup() * 3;
124-
if (currentTimeMs - this.lastLockTimeMs.get() > expiredTimeMs) {
125-
synchronized (this) {
126-
if (currentTimeMs - this.lastLockTimeMs.get() > expiredTimeMs) {
127-
log.warn("HandleData lock expired, acquire lock success and reset lock time. " +
128-
"MessageReceiptHandle={}, lockTime={}", messageReceiptHandle, currentTimeMs);
129-
this.lastLockTimeMs.set(currentTimeMs);
130-
return currentTimeMs;
131-
}
124+
long lockTime = this.lastLockTimeMs.get();
125+
if (currentTimeMs - lockTime > expiredTimeMs) {
126+
if (this.lastLockTimeMs.compareAndSet(lockTime, currentTimeMs)) {
127+
log.warn("HandleData lock expired, acquire lock success and reset lock time. " +
128+
"MessageReceiptHandle={}, lockTime={}", messageReceiptHandle, currentTimeMs);
129+
return currentTimeMs;
132130
}
133131
}
134132
}

‎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: 20 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -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) {
@@ -237,22 +245,31 @@ 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, 0, 0));
250+
}
251+
252+
// When trying for the first time, do not wait for the handle lock, and try to process all handles that can be locked first,
253+
// and then wait for the lock when processing HandleData for the second time.
254+
private void returnHandleGroup(ReceiptHandleGroupKey key, ReceiptHandleGroup handleGroup, long lockTimeout, int attemptTimes) {
255+
if (handleGroup == null || handleGroup.isEmpty() || attemptTimes > 2) {
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+
}, lockTimeout);
252266
} catch (Exception e) {
253267
log.error("error when clear handle for group. key:{}", key, e);
254268
}
255269
});
270+
if (!handleGroup.isEmpty()) {
271+
returnHandleGroupWorkerService.submit(() -> returnHandleGroup(key, handleGroup, proxyConfig.getLockTimeoutMsInHandleGroup(), attemptTimes + 1));
272+
}
256273
}
257274

258275
protected void clearAllHandle() {

0 commit comments

Comments
 (0)