From 07f13fd883e87e40d7de4827e28913e617fb9832 Mon Sep 17 00:00:00 2001 From: Zhanhui Li Date: Mon, 21 Oct 2024 09:25:55 +0800 Subject: [PATCH] fix: remove unnecessary synchronized to improve concurrency (#8840) Signed-off-by: Li Zhanhui --- .../broker/client/ProducerManager.java | 45 ++++++++++--------- 1 file changed, 23 insertions(+), 22 deletions(-) diff --git a/broker/src/main/java/org/apache/rocketmq/broker/client/ProducerManager.java b/broker/src/main/java/org/apache/rocketmq/broker/client/ProducerManager.java index f9fe1193e2..011c9e4be3 100644 --- a/broker/src/main/java/org/apache/rocketmq/broker/client/ProducerManager.java +++ b/broker/src/main/java/org/apache/rocketmq/broker/client/ProducerManager.java @@ -25,6 +25,7 @@ import java.util.List; import java.util.Map; import java.util.Map.Entry; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; import java.util.concurrent.CopyOnWriteArrayList; import org.apache.rocketmq.broker.util.PositiveAtomicCounter; import org.apache.rocketmq.common.constant.LoggerName; @@ -39,11 +40,11 @@ public class ProducerManager { private static final Logger log = LoggerFactory.getLogger(LoggerName.BROKER_LOGGER_NAME); private static final long CHANNEL_EXPIRED_TIMEOUT = 1000 * 120; private static final int GET_AVAILABLE_CHANNEL_RETRY_COUNT = 3; - private final ConcurrentHashMap> groupChannelTable = + private final ConcurrentMap> groupChannelTable = new ConcurrentHashMap<>(); - private final ConcurrentHashMap clientChannelTable = new ConcurrentHashMap<>(); + private final ConcurrentMap clientChannelTable = new ConcurrentHashMap<>(); protected final BrokerStatsManager brokerStatsManager; - private PositiveAtomicCounter positiveAtomicCounter = new PositiveAtomicCounter(); + private final PositiveAtomicCounter positiveAtomicCounter = new PositiveAtomicCounter(); private final List producerChangeListenerList = new CopyOnWriteArrayList<>(); public ProducerManager() { @@ -63,7 +64,7 @@ public class ProducerManager { return channels != null && !channels.isEmpty(); } - public ConcurrentHashMap> getGroupChannelTable() { + public ConcurrentMap> getGroupChannelTable() { return groupChannelTable; } @@ -95,13 +96,13 @@ public class ProducerManager { } public void scanNotActiveChannel() { - Iterator>> iterator = this.groupChannelTable.entrySet().iterator(); + Iterator>> iterator = this.groupChannelTable.entrySet().iterator(); while (iterator.hasNext()) { - Map.Entry> entry = iterator.next(); + Map.Entry> entry = iterator.next(); final String group = entry.getKey(); - final ConcurrentHashMap chlMap = entry.getValue(); + final ConcurrentMap chlMap = entry.getValue(); Iterator> it = chlMap.entrySet().iterator(); while (it.hasNext()) { @@ -132,16 +133,13 @@ public class ProducerManager { } } - public synchronized boolean doChannelCloseEvent(final String remoteAddr, final Channel channel) { + public boolean doChannelCloseEvent(final String remoteAddr, final Channel channel) { boolean removed = false; if (channel != null) { - for (final Map.Entry> entry : this.groupChannelTable - .entrySet()) { + for (final Map.Entry> entry : this.groupChannelTable.entrySet()) { final String group = entry.getKey(); - final ConcurrentHashMap clientChannelInfoTable = - entry.getValue(); - final ClientChannelInfo clientChannelInfo = - clientChannelInfoTable.remove(channel); + final ConcurrentMap clientChannelInfoTable = entry.getValue(); + final ClientChannelInfo clientChannelInfo = clientChannelInfoTable.remove(channel); if (clientChannelInfo != null) { clientChannelTable.remove(clientChannelInfo.getClientId()); removed = true; @@ -150,7 +148,7 @@ public class ProducerManager { clientChannelInfo.toString(), remoteAddr, group); callProducerChangeListener(ProducerGroupEvent.CLIENT_UNREGISTER, group, clientChannelInfo); if (clientChannelInfoTable.isEmpty()) { - ConcurrentHashMap oldGroupTable = this.groupChannelTable.remove(group); + ConcurrentMap oldGroupTable = this.groupChannelTable.remove(group); if (oldGroupTable != null) { log.info("unregister a producer group[{}] from groupChannelTable", group); callProducerChangeListener(ProducerGroupEvent.GROUP_UNREGISTER, group, null); @@ -163,13 +161,16 @@ public class ProducerManager { return removed; } - public synchronized void registerProducer(final String group, final ClientChannelInfo clientChannelInfo) { - ClientChannelInfo clientChannelInfoFound = null; + public void registerProducer(final String group, final ClientChannelInfo clientChannelInfo) { + ClientChannelInfo clientChannelInfoFound; - ConcurrentHashMap channelTable = this.groupChannelTable.get(group); + ConcurrentMap channelTable = this.groupChannelTable.get(group); if (null == channelTable) { channelTable = new ConcurrentHashMap<>(); - this.groupChannelTable.put(group, channelTable); + ConcurrentMap prev = this.groupChannelTable.putIfAbsent(group, channelTable); + if (null != prev) { + channelTable = prev; + } } clientChannelInfoFound = channelTable.get(clientChannelInfo.getChannel()); @@ -186,8 +187,8 @@ public class ProducerManager { } } - public synchronized void unregisterProducer(final String group, final ClientChannelInfo clientChannelInfo) { - ConcurrentHashMap channelTable = this.groupChannelTable.get(group); + public void unregisterProducer(final String group, final ClientChannelInfo clientChannelInfo) { + ConcurrentMap channelTable = this.groupChannelTable.get(group); if (null != channelTable && !channelTable.isEmpty()) { ClientChannelInfo old = channelTable.remove(clientChannelInfo.getChannel()); clientChannelTable.remove(clientChannelInfo.getClientId()); @@ -210,7 +211,7 @@ public class ProducerManager { return null; } List channelList; - ConcurrentHashMap channelClientChannelInfoHashMap = groupChannelTable.get(groupId); + ConcurrentMap channelClientChannelInfoHashMap = groupChannelTable.get(groupId); if (channelClientChannelInfoHashMap != null) { channelList = new ArrayList<>(channelClientChannelInfoHashMap.keySet()); } else {