我有一个场景,我必须处理用户文件夹中的CSV文件,并在处理后将其存储到数据库中。每个用户有5种类型的提要。任何用户都可以在该文件夹中发送任何提要,任何时间进行处理,需要遵循以下规则:
- 同一客户端的同一类型的feed不能在同一时间处理,必须始终阻止并发处理。
- 不允许并发处理超过x个客户端
- 不允许同一客户端并发处理超过"y"个文件
第一个限制可以通过AtomicBoolean的映射来实现。这可能不需要是ConcurrentHashMap,因为初始化后不会更改映射的键。完成后,不要忘记将feed的值重置为false。
checkAndProcessFeed(Feed feed, Map<String, AtomicBoolean> map) {
while(!map.get(feed.type).compareAndSet(false, true)) // assuming the AtomicBooleans were initialized to false
Thread.sleep(500);
}
process(feed); // at this point map.get(feed.type).get() == true
map.get(feed.type).set(false); // reset the AtomicBoolean to false
}
其他两个限制可以通过AtomicInteger实现,它用于维护客户端和每个客户端文件的计数;当处理完成时递减,并使用比较和设置来递增以启动一个新的客户端/文件。
final int maxClient = 5;
AtomicInteger clientCount = new AtomicInteger(0);
ConcurrentLinkedQueue<Client> queue = new ConcurrentLinkedQueue<>(); // hold waiting clients
while(true) {
int temp = clientCount.intValue();
if(!queue.isEmpty() && clientCount.compareAndSet(temp, temp + 1) { // thread-safe increment
process(clients.poll()); // don't forget to decrement the clientCount when the processing finishes
} else Thread.sleep(500);
}