From 97322eaf8e939e6835b90198bc411572c29a536a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=8E=8B=E6=9D=BE=E6=9D=BE?= Date: Wed, 28 Mar 2018 23:36:05 +0800 Subject: [PATCH] Fix Concurrent issue of StoreStatsService --- .../rocketmq/store/StoreStatsService.java | 15 +++- .../rocketmq/store/StoreStatsServiceTest.java | 90 +++++++++++++++++++ 2 files changed, 101 insertions(+), 4 deletions(-) create mode 100644 store/src/test/java/org/apache/rocketmq/store/StoreStatsServiceTest.java 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 bc6493bf..d43b3434 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 putMessageTopicTimesTotal = + private final ConcurrentMap putMessageTopicTimesTotal = new ConcurrentHashMap(128); - private final Map putMessageTopicSizeTotal = + private final ConcurrentMap putMessageTopicSizeTotal = new ConcurrentHashMap(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 00000000..b8a99701 --- /dev/null +++ b/store/src/test/java/org/apache/rocketmq/store/StoreStatsServiceTest.java @@ -0,0 +1,90 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +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; + +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 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 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 -- GitLab