提交 130c85c3 编写于 作者: L Li Zhanhui

Revert "Fix Concurrent issue of StoreStatsService"

This reverts commit fdc6e87c.
上级 61009b7c
......@@ -21,7 +21,6 @@ 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;
......@@ -43,9 +42,9 @@ public class StoreStatsService extends ServiceThread {
private final AtomicLong putMessageFailedTimes = new AtomicLong(0);
private final ConcurrentMap<String, AtomicLong> putMessageTopicTimesTotal =
private final Map<String, AtomicLong> putMessageTopicTimesTotal =
new ConcurrentHashMap<String, AtomicLong>(128);
private final ConcurrentMap<String, AtomicLong> putMessageTopicSizeTotal =
private final Map<String, AtomicLong> putMessageTopicSizeTotal =
new ConcurrentHashMap<String, AtomicLong>(128);
private final AtomicLong getMessageTimesTotalFound = new AtomicLong(0);
......@@ -546,10 +545,7 @@ public class StoreStatsService extends ServiceThread {
AtomicLong rs = putMessageTopicSizeTotal.get(topic);
if (null == rs) {
rs = new AtomicLong(0);
AtomicLong previous = putMessageTopicSizeTotal.putIfAbsent(topic, rs);
if(previous != null){
rs = previous;
}
putMessageTopicSizeTotal.put(topic, rs);
}
return rs;
}
......@@ -558,10 +554,7 @@ public class StoreStatsService extends ServiceThread {
AtomicLong rs = putMessageTopicTimesTotal.get(topic);
if (null == rs) {
rs = new AtomicLong(0);
AtomicLong previous = putMessageTopicTimesTotal.putIfAbsent(topic, rs);
if(previous != null){
rs = previous;
}
putMessageTopicTimesTotal.put(topic, rs);
}
return rs;
}
......
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
Markdown is supported
0% .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑!
想要评论请 注册