【发布时间】：2023-03-31 09:46:01
【问题描述】：
我有一个生产者和许多消费者。
- 生产者速度很快，产生了很多结果
- 具有相同值的令牌需要按顺序处理
- 必须并行处理具有不同值的令牌
- 创建新的 Runnable 会非常昂贵，而且生产代码可以使用 100k 的 Token（为了创建 Runnable，我必须将一些复杂的构建对象传递给构造函数）
我可以使用更简单的算法获得相同的结果吗？用可重入锁嵌套同步块似乎有点不自然。 您可能会注意到任何竞争条件吗？
更新：我发现的第二个解决方案是使用 3 个集合。一是缓存生产者结果，二是阻塞队列，三是使用列表跟踪正在进行的任务。又有点复杂。
我的代码版本
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.locks.ReentrantLock;
public class Main1 {
static class Token {
private int order;
private String value;
Token() {
}
Token(int o, String v) {
order = o;
value = v;
}
int getOrder() {
return order;
}
String getValue() {
return value;
}
}
private final static BlockingQueue<Token> queue = new ArrayBlockingQueue<Token>(10);
private final static ConcurrentMap<String, Object> locks = new ConcurrentHashMap<String, Object>();
private final static ReentrantLock reentrantLock = new ReentrantLock();
private final static Token STOP_TOKEN = new Token();
private final static List<String> lockList = Collections.synchronizedList(new ArrayList<String>());
public static void main(String[] args) {
ExecutorService producerExecutor = Executors.newSingleThreadExecutor();
producerExecutor.submit(new Runnable() {
public void run() {
Random random = new Random();
try {
for (int i = 1; i <= 100; i++) {
Token token = new Token(i, String.valueOf(random.nextInt(1)));
queue.put(token);
}
queue.put(STOP_TOKEN);
}catch(InterruptedException e){
e.printStackTrace();
}
}
});
ExecutorService consumerExecutor = Executors.newFixedThreadPool(10);
for(int i=1; i<=10;i++) {
// creating to many runnable would be inefficient because of this complex not thread safe object
final Object dependecy = new Object(); //new ComplexDependecy()
consumerExecutor.submit(new Runnable() {
public void run() {
while(true) {
try {
//not in order
Token token = queue.take();
if (token == STOP_TOKEN) {
queue.add(STOP_TOKEN);
return;
}
System.out.println("Task start" + Thread.currentThread().getId() + " order " + token.getOrder());
Random random = new Random();
Thread.sleep(random.nextInt(200)); //doLongRunningTask(dependecy)
lockList.remove(token.getValue());
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}});
}
}}
【问题讨论】：
-
...必须并行处理... 很难强制任何两件或多件事情同时发生。 允许不同的线程并行做事，但Java中没有任何东西保证事情会并行发生。
标签： java multithreading synchronization locking