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

fix thread-safety problem of admin tools (#4843)

Browse Source
This commit is contained in:
HuiTong
2022-08-19 12:26:47 +08:00
committed by GitHub
parent 21f3a0bb41
commit b7f0162ffa
4 changed files with 17 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
common/src/main/java/org/apache/rocketmq/common/admin/ConsumeStats.java
+6 -4
View File
@@ -16,14 +16,16 @@
*/
package org.apache.rocketmq.common.admin;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.protocol.RemotingSerializable;
public class ConsumeStats extends RemotingSerializable {
private HashMap<MessageQueue, OffsetWrapper> offsetTable = new HashMap<MessageQueue, OffsetWrapper>();
private Map<MessageQueue, OffsetWrapper> offsetTable = new ConcurrentHashMap<MessageQueue, OffsetWrapper>();
private double consumeTps = 0;
public long computeTotalDiff() {
@@ -39,11 +41,11 @@ public class ConsumeStats extends RemotingSerializable {
return diffTotal;
}
public HashMap<MessageQueue, OffsetWrapper> getOffsetTable() {
public Map<MessageQueue, OffsetWrapper> getOffsetTable() {
return offsetTable;
}
public void setOffsetTable(HashMap<MessageQueue, OffsetWrapper> offsetTable) {
public void setOffsetTable(Map<MessageQueue, OffsetWrapper> offsetTable) {
this.offsetTable = offsetTable;
}
common/src/main/java/org/apache/rocketmq/common/admin/TopicStatsTable.java
+6 -4
View File
@@ -16,18 +16,20 @@
*/
package org.apache.rocketmq.common.admin;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.protocol.RemotingSerializable;
public class TopicStatsTable extends RemotingSerializable {
private HashMap<MessageQueue, TopicOffset> offsetTable = new HashMap<MessageQueue, TopicOffset>();
private Map<MessageQueue, TopicOffset> offsetTable = new ConcurrentHashMap<MessageQueue, TopicOffset>();
public HashMap<MessageQueue, TopicOffset> getOffsetTable() {
public Map<MessageQueue, TopicOffset> getOffsetTable() {
return offsetTable;
}
public void setOffsetTable(HashMap<MessageQueue, TopicOffset> offsetTable) {
public void setOffsetTable(Map<MessageQueue, TopicOffset> offsetTable) {
this.offsetTable = offsetTable;
}
}
common/src/main/java/org/apache/rocketmq/common/protocol/body/TopicList.java
+3 -2
View File
@@ -16,12 +16,13 @@
*/
package org.apache.rocketmq.common.protocol.body;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArraySet;
import org.apache.rocketmq.remoting.protocol.RemotingSerializable;
public class TopicList extends RemotingSerializable {
private Set<String> topicList = new HashSet<String>();
private Set<String> topicList = new CopyOnWriteArraySet<>();
private String brokerAddr;
public Set<String> getTopicList() {
tools/src/main/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExtImpl.java
+2 -2
View File
@@ -797,7 +797,7 @@ public class DefaultMQAdminExtImpl implements MQAdminExt, MQAdminExtInner {
}
if (!hasConsumed) {
HashMap<MessageQueue, TopicOffset> topicStatus = this.mqClientInstance.getMQClientAPIImpl().getTopicStatsInfo(brokerAddr, topic, timeoutMillis).getOffsetTable();
Map<MessageQueue, TopicOffset> topicStatus = this.mqClientInstance.getMQClientAPIImpl().getTopicStatsInfo(brokerAddr, topic, timeoutMillis).getOffsetTable();
for (int i = 0; i < queueData.getReadQueueNums(); i++) {
MessageQueue queue = new MessageQueue(topic, queueData.getBrokerName(), i);
OffsetWrapper offsetWrapper = new OffsetWrapper();
@@ -1107,7 +1107,7 @@ public class DefaultMQAdminExtImpl implements MQAdminExt, MQAdminExtInner {
return adminToolExecute(new AdminToolHandler() {
@Override
public AdminToolResult doExecute() throws Exception {
final List<QueueTimeSpan> spanSet = new ArrayList<QueueTimeSpan>();
final List<QueueTimeSpan> spanSet = new CopyOnWriteArrayList<>();
TopicRouteData topicRouteData = examineTopicRouteInfo(topic);
if (topicRouteData == null || topicRouteData.getBrokerDatas() == null || topicRouteData.getBrokerDatas().size() == 0) {