面向多用户的csv并发处理

Question

我有一个场景，在这个场景中，我必须为用户处理一个文件夹中的CSV文件，并在处理后将它们存储到数据库中。我们有每个用户5种类型的提要。任何用户都可以在该文件夹中发送任何提要，任何时间进行处理，都需要遵循以下规则：

• 同一客户端的相同类型的提要不能以相同的方式处理，因此必须始终阻止并发处理。
• 不允许跨"x“客户端进行并发处理。
• 不允许对同一客户端的"y“文件进行并发处理。

实现这一目标的好方法是什么？

Answer 1

第一个限制可以用AtomicBoolean映射来实现。这可能不需要是ConcurrentHashMap，因为初始化后不会更改映射的键。完成后不要忘记将提要的值重置为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);
}