This is an automated email from the ASF dual-hosted git repository. lizhanhui pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/rocketmq.git
commit fdc6e87cbe7adb8bc8f5991e06392eee73bc5151 Author: 王松松 <[email protected]> AuthorDate: Wed Mar 28 23:36:05 2018 +0800 Fix Concurrent issue of StoreStatsService --- .../apache/rocketmq/store/StoreStatsService.java | 15 +++-- .../rocketmq/store/StoreStatsServiceTest.java | 77 ++++++++++++++++++++++ 2 files changed, 88 insertions(+), 4 deletions(-) diff --git a/store/src/main/java/org/apache/rocketmq/store/StoreStatsService.java b/store/src/main/java/org/apache/rocketmq/store/StoreStatsService.java index 586947c..8862fd7 100644 --- a/store/src/main/java/org/apache/rocketmq/store/StoreStatsService.java +++ b/store/src/main/java/org/apache/rocketmq/store/StoreStatsService.java @@ -21,6 +21,7 @@ import java.util.HashMap; import java.util.LinkedList; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.locks.ReentrantLock; import org.apache.rocketmq.common.ServiceThread; @@ -42,9 +43,9 @@ public class StoreStatsService extends ServiceThread { private final AtomicLong putMessageFailedTimes = new AtomicLong(0); - private final Map<String, AtomicLong> putMessageTopicTimesTotal = + private final ConcurrentMap<String, AtomicLong> putMessageTopicTimesTotal = new ConcurrentHashMap<String, AtomicLong>(128); - private final Map<String, AtomicLong> putMessageTopicSizeTotal = + private final ConcurrentMap<String, AtomicLong> putMessageTopicSizeTotal = new ConcurrentHashMap<String, AtomicLong>(128); private final AtomicLong getMessageTimesTotalFound = new AtomicLong(0); @@ -545,7 +546,10 @@ public class StoreStatsService extends ServiceThread { AtomicLong rs = putMessageTopicSizeTotal.get(topic); if (null == rs) { rs = new AtomicLong(0); - putMessageTopicSizeTotal.put(topic, rs); + AtomicLong previous = putMessageTopicSizeTotal.putIfAbsent(topic, rs); + if(previous != null){ + rs = previous; + } } return rs; } @@ -554,7 +558,10 @@ public class StoreStatsService extends ServiceThread { AtomicLong rs = putMessageTopicTimesTotal.get(topic); if (null == rs) { rs = new AtomicLong(0); - putMessageTopicTimesTotal.put(topic, rs); + AtomicLong previous = putMessageTopicTimesTotal.putIfAbsent(topic, rs); + if(previous != null){ + rs = previous; + } } return rs; } diff --git a/store/src/test/java/org/apache/rocketmq/store/StoreStatsServiceTest.java b/store/src/test/java/org/apache/rocketmq/store/StoreStatsServiceTest.java new file mode 100644 index 0000000..bb39bf9 --- /dev/null +++ b/store/src/test/java/org/apache/rocketmq/store/StoreStatsServiceTest.java @@ -0,0 +1,77 @@ +package org.apache.rocketmq.store; + +import java.util.concurrent.BrokenBarrierException; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.CyclicBarrier; +import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.atomic.AtomicReference; +import org.junit.Test; + +/** + * @author song + */ +public class StoreStatsServiceTest { + + @Test + public void getSinglePutMessageTopicSizeTotal() throws Exception { + final StoreStatsService storeStatsService = new StoreStatsService(); + int num = Runtime.getRuntime().availableProcessors() * 2; + for (int j = 0; j < 100; j++) { + final AtomicReference<AtomicLong> reference = new AtomicReference<>(null); + final CountDownLatch latch = new CountDownLatch(num); + final CyclicBarrier barrier = new CyclicBarrier(num); + for (int i = 0; i < num; i++) { + new Thread(new Runnable() { + @Override + public void run() { + try { + barrier.await(); + AtomicLong atomicLong = storeStatsService.getSinglePutMessageTopicSizeTotal("test"); + if(reference.compareAndSet(null,atomicLong)){ + }else if (reference.get() != atomicLong){ + throw new RuntimeException("Reference should be same!"); + } + } catch (InterruptedException | BrokenBarrierException e) { + e.printStackTrace(); + }finally { + latch.countDown(); + } + } + }).start(); + } + latch.await(); + } + } + + @Test + public void getSinglePutMessageTopicTimesTotal() throws Exception { + final StoreStatsService storeStatsService = new StoreStatsService(); + int num = Runtime.getRuntime().availableProcessors() * 2; + for (int j = 0; j < 100; j++) { + final AtomicReference<AtomicLong> reference = new AtomicReference<>(null); + final CountDownLatch latch = new CountDownLatch(num); + final CyclicBarrier barrier = new CyclicBarrier(num); + for (int i = 0; i < num; i++) { + new Thread(new Runnable() { + @Override + public void run() { + try { + barrier.await(); + AtomicLong atomicLong = storeStatsService.getSinglePutMessageTopicTimesTotal("test"); + if(reference.compareAndSet(null,atomicLong)){ + }else if (reference.get() != atomicLong){ + throw new RuntimeException("Reference should be same!"); + } + } catch (InterruptedException | BrokenBarrierException e) { + e.printStackTrace(); + }finally { + latch.countDown(); + } + } + }).start(); + } + latch.await(); + } + } + +} \ No newline at end of file -- To stop receiving notification emails like this one, please contact [email protected].
