【发布时间】：2014-03-04 13:18:21
【问题描述】：
任务定义：我需要测试自定义并发集合或一些在并发环境中操作集合的容器。 更准确地说 - 我有读 API 和写 API。我应该测试是否有任何情况下我会得到不一致的数据。
问题：所有并发测试框架（如MultiThreadedTC，请查看我的问题的 MultiThreadedTc 部分）只是为您提供了控制异步代码执行顺序的能力。我的意思是你应该自己假设一个关键场景。
宽泛的问题：是否有框架可以采用 @SharedResource、@readAPI、@writeAPI 等注释并检查您的数据是否始终一致？这是不可能的，还是我只是泄露了一个创业想法？
注解：如果没有这样的框架，但你觉得这个想法很有吸引力，欢迎联系我或提出你的想法。
狭义问题：我是并发新手。那么你能建议我应该在下面的代码中测试哪些场景吗？ （看PeerContainer类）
PeerContainer：
public class PeersContainer {
public class DaemonThreadFactory implements ThreadFactory {
private int counter = 1;
private final String prefix = "Daemon";
@Override
public Thread newThread(Runnable r) {
Thread thread = new Thread(r, prefix + "-" + counter);
thread.setDaemon(true);
counter++;
return thread;
}
}
private static class CacheCleaner implements Runnable {
private final Cache<Long, BlockingDeque<Peer>> cache;
public CacheCleaner(Cache<Long, BlockingDeque<Peer>> cache) {
this.cache = cache;
Thread.currentThread().setDaemon(true);
}
@Override
public void run() {
cache.cleanUp();
}
}
private final static int MAX_CACHE_SIZE = 100;
private final static int STRIPES_AMOUNT = 10;
private final static int PEER_ACCESS_TIMEOUT_MIN = 30;
private final static int CACHE_CLEAN_FREQUENCY_MIN = 1;
private final static PeersContainer INSTANCE;
private final Cache<Long, BlockingDeque<Peer>> peers = CacheBuilder.newBuilder()
.maximumSize(MAX_CACHE_SIZE)
.expireAfterWrite(PEER_ACCESS_TIMEOUT_MIN, TimeUnit.MINUTES)
.removalListener(new RemovalListener<Long, BlockingDeque<Peer>>() {
public void onRemoval(RemovalNotification<Long, BlockingDeque<Peer>> removal) {
if (removal.getCause() == RemovalCause.EXPIRED) {
for (Peer peer : removal.getValue()) {
peer.sendLogoutResponse(peer);
}
}
}
})
.build();
private final Striped<Lock> stripes = Striped.lock(STRIPES_AMOUNT);
private final ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(1, new DaemonThreadFactory());
private PeersContainer() {
scheduledExecutorService.schedule(new CacheCleaner(peers), CACHE_CLEAN_FREQUENCY_MIN, TimeUnit.MINUTES);
}
static {
INSTANCE = new PeersContainer();
}
public static PeersContainer getInstance() {
return INSTANCE;
}
private final Cache<Long, UserAuthorities> authToRestore = CacheBuilder.newBuilder()
.maximumSize(MAX_CACHE_SIZE)
.expireAfterWrite(PEER_ACCESS_TIMEOUT_MIN, TimeUnit.MINUTES)
.build();
public Collection<Peer> getPeers(long sessionId) {
return Collections.unmodifiableCollection(peers.getIfPresent(sessionId));
}
public Collection<Peer> getAllPeers() {
BlockingDeque<Peer> result = new LinkedBlockingDeque<Peer>();
for (BlockingDeque<Peer> deque : peers.asMap().values()) {
result.addAll(deque);
}
return Collections.unmodifiableCollection(result);
}
public boolean addPeer(Peer peer) {
long key = peer.getSessionId();
Lock lock = stripes.get(key);
lock.lock();
try {
BlockingDeque<Peer> userPeers = peers.getIfPresent(key);
if (userPeers == null) {
userPeers = new LinkedBlockingDeque<Peer>();
peers.put(key, userPeers);
}
UserAuthorities authorities = restoreSession(key);
if (authorities != null) {
peer.setAuthorities(authorities);
}
return userPeers.offer(peer);
} finally {
lock.unlock();
}
}
public void removePeer(Peer peer) {
long sessionId = peer.getSessionId();
Lock lock = stripes.get(sessionId);
lock.lock();
try {
BlockingDeque<Peer> userPeers = peers.getIfPresent(sessionId);
if (userPeers != null && !userPeers.isEmpty()) {
UserAuthorities authorities = userPeers.getFirst().getAuthorities();
authToRestore.put(sessionId, authorities);
userPeers.remove(peer);
}
} finally {
lock.unlock();
}
}
void removePeers(long sessionId) {
Lock lock = stripes.get(sessionId);
lock.lock();
try {
peers.invalidate(sessionId);
authToRestore.invalidate(sessionId);
} finally {
lock.unlock();
}
}
private UserAuthorities restoreSession(long sessionId) {
BlockingDeque<Peer> activePeers = peers.getIfPresent(sessionId);
return (activePeers != null && !activePeers.isEmpty()) ? activePeers.getFirst().getAuthorities() : authToRestore.getIfPresent(sessionId);
}
public void resetAccessedTimeout(long sessionId) {
Lock lock = stripes.get(sessionId);
lock.lock();
try {
BlockingDeque<Peer> deque = peers.getIfPresent(sessionId);
peers.invalidate(sessionId);
peers.put(sessionId, deque);
} finally {
lock.unlock();
}
}
}
MultiThreadedTC 测试用例示例：[问题的可选部分]
public class ProducerConsumerTest extends MultithreadedTestCase {
private LinkedTransferQueue<String> queue;
@Override
public void initialize() {
super.initialize();
queue = new LinkedTransferQueue<String>();
}
public void thread1() throws InterruptedException {
String ret = queue.take();
}
public void thread2() throws InterruptedException {
waitForTick(1);
String ret = queue.take();
}
public void thread3() {
waitForTick(1);
waitForTick(2);
queue.put("Event 1");
queue.put("Event 2");
}
@Override
public void finish() {
super.finish();
assertEquals(true, queue.size() == 0);
}
}
【问题讨论】：
标签： java multithreading unit-testing testing concurrency