Java中的生产者-消费者多线程FIFO

Question

社区，我试图用10个线程来解决这个生产者/消费者问题，但在实现它时遇到了困难。

问题看起来是这样的：

问题方案

程序本身在插入包含(id，timeout)的消息时应该有一个循环，按id(1,2,3,4…)升序排列，并且应该简单地按输入的顺序打印出消息的id，就像队列一样。

例如，在上面的照片中，3条消息Message(1200)、Message(21000)和Message(3,20)是生产者将产生的前3条消息。虽然应该首先打印被分配了Message(3,20)的线程(因为它的超时值(20)最低)，但我希望它等待第一条超时为200ms的消息打印，然后再次等待1000ms的消息2打印，然后打印自己。所以都是按递增顺序排列的(也许用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}

我的一个朋友告诉我应该用信号量来实现(从来没有用过)，但我真的不知道如何实现信号量，让它们做我想做的事。

感谢任何解决此问题的线索！

Answer 1

据我所知，您需要两样东西：

1. 一起启动所有生产者的工作线程，并让它们并行运行，但是
2. 等待线程按FIFO顺序完成(根据它们的创建id)

因此，您可以一个接一个地启动线程，以便让它们并行运行，但也可以通过升序id维护FIFO队列，并且只为添加到该队列的序列中的每个线程join。

这里有一个关于如何做到这一点的演示代码：

import java.util.LinkedList;
import java.util.Objects;
import java.util.Random;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
public class Main {

private static class Message implements Runnable {
private final TimeUnit sleepUnit;
private final long sleepAmount;
private final int id;

public Message(final int id,
final TimeUnit sleepUnit,
final long sleepAmount) {
this.sleepUnit = Objects.requireNonNull(sleepUnit);
this.sleepAmount = sleepAmount;
this.id = id;
}

@Override
public void run() {
try {
System.out.println(toString() + " started and waiting...");
sleepUnit.sleep(sleepAmount);
}
catch (final InterruptedException ix) {
System.out.println(toString() + " interrupted: " + ix);
}
}

@Override
public String toString() {
return "Message{" + id + ", " + sleepUnit + "(" + sleepAmount + ")}";
}
}

private static class Producer {

private final int parallelism;
private final Consumer<? super Producer> consumer;

public Producer(final int parallelism,
final Consumer<? super Producer> consumer) {
this.parallelism = parallelism;
this.consumer = Objects.requireNonNull(consumer);
}

public void produceWithExecutor() {
System.out.println("Producing with Executor...");
final Random rand = new Random();
final ExecutorService service = Executors.newFixedThreadPool(parallelism);
final LinkedList<Future> q = new LinkedList<>();
for (int i = 0; i < parallelism; ++i) {
final Message msg = new Message(i, TimeUnit.MILLISECONDS, 500 + rand.nextInt(3000));
q.addLast(service.submit(msg, msg));
}
service.shutdown();
while (!q.isEmpty())
try {
System.out.println(q.removeFirst().get().toString() + " joined."); //Will wait for completion of each submitted task (in FIFO sequence).
}
catch (final InterruptedException ix) {
System.out.println("Interrupted: " + ix);
}
catch (final ExecutionException xx) {
System.out.println("Execution failed: " + xx);
}
consumer.accept(this);
}

public void produceWithPlainThreads() throws InterruptedException {
System.out.println("Producing with Threads...");
final Random rand = new Random();
final LinkedList<Thread> q = new LinkedList<>();
for (int i = 0; i < parallelism; ++i) {
final Message msg = new Message(i, TimeUnit.MILLISECONDS, 500 + rand.nextInt(3000));
final Thread t = new Thread(msg, msg.toString());
t.start();
q.add(t);
}
while (!q.isEmpty()) {
final Thread t = q.removeFirst();
t.join(); //Will wait for completion of each submitted task (in FIFO sequence).
System.out.println(t.getName() + " joined.");
}
consumer.accept(this);
}
}

public static void main(final String[] args) throws InterruptedException {
final Consumer<Producer> consumer = producer -> System.out.println("Consuming.");
final int parallelism = 10;
new Producer(parallelism, consumer).produceWithExecutor();
new Producer(parallelism, consumer).produceWithPlainThreads();
}
}

正如您所看到的，这里有两种生产实现：一种是ExecutorService运行所有提交的线程，另一种是(几乎)同时启动的普通线程。

这导致了这样的输出：

使用Executor制作
消息｛1，MILLISECONDS(692)｝已启动并正在等待
消息｛2，MILLISECONDS(1126)｝已启动并正在等待
消息｛0，MILLISECONDS(3403)｝已启动并正在等待
消息｛3，MILLISECONDS(1017)｝已启动并正在等待
消息｛4，MILLISECONDS(2861)｝已启动并正在等待
消息｛5，MILLISECONDS(2735)｝已启动并正在等待
消息｛6，MILLISECONDS(2068)｝已启动并正在等待
消息｛七千万秒(947)｝已启动并正在等待
消息｛8，MILLISECONDS(1091)｝已启动并正在等待
消息｛9，MILLISECONDS(1599)｝已启动并正在等待
消息｛0，MILLISECONDS(3403)｝已加入
消息｛1，MILLISECONDS(692)｝已加入
消息｛2，MILLISECONDS(1126)｝已加入
消息｛3，MILLISECONDS(1017)｝已加入
消息｛4，MILLISECONDS(2861)｝已加入
消息｛5，MILLISECONDS(2735)｝已加入
消息｛6，MILLISECONDS(2068)｝已加入
消息｛七千万秒(947)｝已加入
消息｛89000000秒(1091)｝已加入
消息｛9，MILLISECONDS(1599)｝已加入
消耗
使用线程进行生产
消息｛0，MILLISECONDS(3182)｝已启动并正在等待
消息｛1，MILLISECONDS(2271)｝已启动并正在等待
消息｛2，MILLISECONDS(2861)｝已启动并正在等待
消息｛3，MILLISECONDS(2942)｝已启动并正在等待
消息｛4，MILLISECONDS(2714)｝已启动并正在等待
消息｛5，MILLISECONDS(1228)｝已启动并正在等待
消息｛6，MILLISECONDS(2000)｝已启动并正在等待
消息｛七千万秒(2372)｝已启动并正在等待
消息｛8，MILLISECONDS(764)｝已启动并正在等待
消息｛9，MILLISECONDS(587)｝已启动并正在等待
消息｛0，MILLISECONDS(3182)｝已加入
消息｛1，MILLISECONDS(2271)｝已加入
消息｛2，MILLISECONDS(2861)｝已加入
消息｛3，MILLISECONDS(2942)｝已加入
消息｛4，MILLISECONDS(2714)｝已加入
消息｛5，MILLISECONDS(1228)｝已加入
消息｛600万秒(2000)｝已加入
消息｛七千万秒(2372)｝已加入
消息｛8000万秒(764)｝已加入
消息｛900万秒(587)｝已加入
消耗。

您可以在输出中看到，在这两种情况下，线程都是通过循环(几乎)一起启动的，但是以FIFO顺序连接的。在第一种情况下，您可以看到线程可能以不同的顺序启动，这是启动线程本身的副作用。在普通线程的第二种情况下，所有线程都按照创建和启动它们的顺序调用了它们的run方法，因为这会在很短的时间内发生。但是，根据此代码，每个线程的连接将始终按照id升序进行。如果您多次运行此代码，在这两种情况下，您都可以实现线程eg 2在其运行方法中打印在线程eg 1之前，但我们在Producer的方法中等待线程完成的顺序将始终以升序结束。

所有线程都应该按睡眠升序退出/完成其run方法，而不是按id升序。但是，由于我们对队列进行迭代并以有序的方式等待它们到达join，因此输出将始终按升序排列。

因此，如果您想以升序获得每个Thread的结果，那么相应的代码必须在Producer的product方法中(在每个线程join之后)，并且在每个Message的run方法的末尾而不是(以避免额外的同步和线程间通信)。