From 4bf4681d0e7a43e6e3f4ebfe0c31243272d2e8ba Mon Sep 17 00:00:00 2001
From: chengxiangwang <chengxiangwang@cmss.chinamobile.com>
Date: Thu, 27 Dec 2018 12:43:54 +0800
Subject: [PATCH] use putIfAbsent in MomentStatsItemSet.java

---
 .../broker/BrokerStatsManagerTest.java        | 30 ++++++++++++++++++-
 .../common/stats/MomentStatsItemSet.java      |  6 ++--
 2 files changed, 32 insertions(+), 4 deletions(-)

diff --git a/broker/src/test/java/org/apache/rocketmq/broker/BrokerStatsManagerTest.java b/broker/src/test/java/org/apache/rocketmq/broker/BrokerStatsManagerTest.java
index 6268865b..1948b0bf 100644
--- a/broker/src/test/java/org/apache/rocketmq/broker/BrokerStatsManagerTest.java
+++ b/broker/src/test/java/org/apache/rocketmq/broker/BrokerStatsManagerTest.java
@@ -41,7 +41,14 @@ public class BrokerStatsManagerTest {
         }
     }
 
-    public AtomicLong test_unit() throws InterruptedException {
+    @Test
+    public void test_getAndCreateMomentStatsItem_multiThread() throws InterruptedException {
+        for (int i = 0; i < 5; i++) {
+            assertEquals(10, test_unit_moment().longValue());
+        }
+    }
+
+    private AtomicLong test_unit() throws InterruptedException {
         brokerStatsManager = new BrokerStatsManager("DefaultCluster");
         executor = new ThreadPoolExecutor(100, 200, 10, TimeUnit.SECONDS,
                 new ArrayBlockingQueue<Runnable>(10000), new ThreadFactoryImpl("testMultiThread"));
@@ -62,6 +69,27 @@ public class BrokerStatsManagerTest {
         return brokerStatsManager.getStatsItem(TOPIC_PUT_NUMS, "topicTest").getValue();
     }
 
+    private AtomicLong test_unit_moment() throws InterruptedException {
+        brokerStatsManager = new BrokerStatsManager("DefaultCluster");
+        executor = new ThreadPoolExecutor(100, 200, 10, TimeUnit.SECONDS,
+                new ArrayBlockingQueue<Runnable>(10000), new ThreadFactoryImpl("testMultiThread"));
+        for (int i = 0; i < 10000; i++) {
+            executor.submit(new Runnable() {
+                @Override
+                public void run() {
+                    brokerStatsManager.getMomentStatsItemSetFallTime().setValue("test",10);
+                }
+            });
+        }
+        while (true) {
+            if (executor.getCompletedTaskCount() == 10000) {
+                break;
+            }
+            Thread.sleep(1000);
+        }
+        return brokerStatsManager.getMomentStatsItemSetFallTime().getAndCreateStatsItem("test").getValue();
+    }
+
     @After
     public void shutdown() {
         executor.shutdown();
diff --git a/common/src/main/java/org/apache/rocketmq/common/stats/MomentStatsItemSet.java b/common/src/main/java/org/apache/rocketmq/common/stats/MomentStatsItemSet.java
index d3b55966..a38b53ff 100644
--- a/common/src/main/java/org/apache/rocketmq/common/stats/MomentStatsItemSet.java
+++ b/common/src/main/java/org/apache/rocketmq/common/stats/MomentStatsItemSet.java
@@ -79,10 +79,10 @@ public class MomentStatsItemSet {
         if (null == statsItem) {
             statsItem =
                 new MomentStatsItem(this.statsName, statsKey, this.scheduledExecutorService, this.log);
-            MomentStatsItem prev = this.statsItemTable.put(statsKey, statsItem);
-
-            if (null == prev) {
+            MomentStatsItem prev = this.statsItemTable.putIfAbsent(statsKey, statsItem);
 
+            if (null != prev) {
+                statsItem = prev;
                 // statsItem.init();
             }
         }
-- 
GitLab