【发布时间】：2021-03-25 23:08:12
【问题描述】：
社区，我正在尝试用 10 个线程解决这个生产者/消费者问题，但我在实现它时遇到了困难。
问题如下：
程序本身在插入包含 (id, timeout) 的消息时应该有一个循环，按 id (1,2,3,4...) 升序排列，并且应该简单地打印出来的消息的 id ，按照它进入的顺序，就像一个队列。
例如在上面的照片中，3 条消息 Message(1,200)、Message(2, 1000) 和 Message(3,20) 是生产者将产生的前 3 条消息。 虽然应该首先打印分配有 Message(3,20) 的线程（因为它的超时时间（20）最低），但我希望它等待有 200ms 超时的第一条消息打印，然后再次等待message2 用 1000 毫秒打印，然后打印自己。所以一切都是递增的（也许使用 id 作为订购号？）。
到目前为止，我已经实现了这个：
public class Main {
private static BlockingQueue<Message> queue = new ArrayBlockingQueue<>(5);
public static void main(String[] args) throws InterruptedException {
Thread t1 = new Thread(() -> {
try {
producer();
} catch (InterruptedException exception) {
exception.printStackTrace();
}
});
Thread t2 = new Thread(() -> {
try {
consumer();
} catch (InterruptedException exception) {
exception.printStackTrace();
}
});
t1.start();
t2.start();
t1.join();
t2.join();
}
public static void producer() throws InterruptedException {
while (true) {
queue.put(new Message());
}
}
public static void consumer() throws InterruptedException {
ExecutorService executorService = Executors.newFixedThreadPool(10);
for (int i = 0; i < 1000; i++) {
executorService.submit(queue.take());
}
executorService.shutdown();
}
}
我在这里有我的消息类：
public class Message implements Runnable {
public static int totalIds = 0;
public int id;
public int timeout;
public Random random = new Random();
public Message() {
this.id = totalIds;
totalIds++;
this.timeout = random.nextInt(5000);
}
@Override
public String toString() {
return "Message{" +
"id=" + id +
", timeout=" + timeout +
'}';
}
@Override
public void run() {
System.out.println(Thread.currentThread().getName() + "[RECEIVED] Message = " + toString());
try {
Thread.sleep(timeout);
} catch (InterruptedException exception) {
exception.printStackTrace();
}
System.out.println(Thread.currentThread().getName() + "[DONE] Message = " + toString() + "\n");
}
}
到目前为止，除了线程应该等待具有优先级 id 的部分之外，它一切正常...这里是输出的第一部分：
All tasks submitted
pool-1-thread-9[RECEIVED] Message = Message{id=13, timeout=1361}
pool-1-thread-10[RECEIVED] Message = Message{id=14, timeout=92}
pool-1-thread-3[RECEIVED] Message = Message{id=7, timeout=3155}
pool-1-thread-5[RECEIVED] Message = Message{id=9, timeout=562}
pool-1-thread-2[RECEIVED] Message = Message{id=6, timeout=4249}
pool-1-thread-1[RECEIVED] Message = Message{id=0, timeout=1909}
pool-1-thread-7[RECEIVED] Message = Message{id=11, timeout=2468}
pool-1-thread-4[RECEIVED] Message = Message{id=8, timeout=593}
pool-1-thread-8[RECEIVED] Message = Message{id=12, timeout=3701}
pool-1-thread-6[RECEIVED] Message = Message{id=10, timeout=806}
pool-1-thread-10[DONE] Message = Message{id=14, timeout=92}
pool-1-thread-10[RECEIVED] Message = Message{id=15, timeout=846}
pool-1-thread-5[DONE] Message = Message{id=9, timeout=562}
pool-1-thread-5[RECEIVED] Message = Message{id=16, timeout=81}
pool-1-thread-4[DONE] Message = Message{id=8, timeout=593}
pool-1-thread-4[RECEIVED] Message = Message{id=17, timeout=4481}
pool-1-thread-5[DONE] Message = Message{id=16, timeout=81}
pool-1-thread-5[RECEIVED] Message = Message{id=18, timeout=2434}
pool-1-thread-6[DONE] Message = Message{id=10, timeout=806}
pool-1-thread-6[RECEIVED] Message = Message{id=19, timeout=10}
pool-1-thread-6[DONE] Message = Message{id=19, timeout=10}
pool-1-thread-6[RECEIVED] Message = Message{id=20, timeout=3776}
pool-1-thread-10[DONE] Message = Message{id=15, timeout=846}
pool-1-thread-10[RECEIVED] Message = Message{id=21, timeout=2988}
pool-1-thread-9[DONE] Message = Message{id=13, timeout=1361}
pool-1-thread-9[RECEIVED] Message = Message{id=22, timeout=462}
pool-1-thread-9[DONE] Message = Message{id=22, timeout=462}
pool-1-thread-9[RECEIVED] Message = Message{id=23, timeout=3074}
pool-1-thread-1[DONE] Message = Message{id=0, timeout=1909}
pool-1-thread-1[RECEIVED] Message = Message{id=24, timeout=725}
pool-1-thread-7[DONE] Message = Message{id=11, timeout=2468}
我的一个朋友告诉我应该使用信号量来完成（从未使用过它们），但我真的不知道如何实现信号量以便它们做我想做的事情。
感谢任何解决此问题的线索！
【问题讨论】：
-
在线程中使用 ExecutorService 有什么意义？
-
另外，你的生产者不断地产生一条随机睡眠时间的消息。所以如果你尝试只产生 10 条消息，消费 10 条消息，你会看到线程是按睡眠时间顺序运行的，但是当
queue.take()一个线程不知道睡眠时间所以这个日志是正常的。 -
我在该消费者线程中使用 ExecutorService，因为我希望有 10 个线程处理消息。另外，我不是试图产生 10 条消息并消耗 10 条消息，我试图产生 N 条消息，它与数字无关，或者它甚至可以是无限数量的消息......基本上我能做到的全部要做的就是告诉我的线程互相等待，以便它们按照输入的顺序打印消息。 （即升序：1,2,3,4,5,6...）
-
是的，这就是异步编程的重点。
标签： java multithreading parallel-processing semaphore producer-consumer