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

[ISSUE #3949] Refector ReceiptHandleGroup

Browse Source
This commit is contained in:
zhouxiang
2022-07-13 11:30:54 +08:00
parent 0a017719c3
commit 042febd9f1
2 changed files with 5 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
proxy/src/main/java/org/apache/rocketmq/proxy/common/ReceiptHandleGroup.java
+3 -3
View File
@@ -28,11 +28,11 @@ public class ReceiptHandleGroup {
receiptHandleMap.put(key, value);
}
public void remove(String key) {
receiptHandleMap.remove(key);
public MessageReceiptHandle remove(String key) {
return receiptHandleMap.remove(key);
}
public Map<String, MessageReceiptHandle> all() {
public Map<String, MessageReceiptHandle> immutableMapView() {
return ImmutableMap.copyOf(receiptHandleMap);
}
}
proxy/src/main/java/org/apache/rocketmq/proxy/processor/ReceiptHandleProcessor.java
+2 -2
View File
@@ -107,7 +107,7 @@ public class ReceiptHandleProcessor extends AbstractStartAndShutdown {
for (Map.Entry<String, ReceiptHandleGroup> entry : receiptHandleGroupMap.entrySet()) {
String key = entry.getKey();
ReceiptHandleGroup group = entry.getValue();
group.all().forEach((k, v) -> {
group.immutableMapView().forEach((k, v) -> {
ReceiptHandle handle = ReceiptHandle.decode(v.getReceiptHandle());
long now = System.currentTimeMillis();
if (handle.getNextVisibleTime() - now > proxyConfig.getRenewAheadTimeMillis()) {
@@ -178,7 +178,7 @@ public class ReceiptHandleProcessor extends AbstractStartAndShutdown {
}
ProxyConfig proxyConfig = ConfigurationManager.getProxyConfig();
receiptHandleGroupMap.computeIfPresent(key, (k, v) -> {
Map<String, MessageReceiptHandle> all = v.all();
Map<String, MessageReceiptHandle> all = v.immutableMapView();
all.forEach((key0, value0) -> {
ReceiptHandle receiptHandle = ReceiptHandle.decode(value0.getReceiptHandle());
messagingProcessor.changeInvisibleTime(