java - Java中的生产者-消费者多线程FIFO
问题描述
社区，我正在尝试用 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}
我的一位朋友告诉我应该使用信号量来完成（从未使用过它们），但我真的不知道如何实现信号量以便他们做我想做的事。
感谢解决这个问题的任何线索！
解决方案
据我了解，您需要两件事：
- 一起启动所有生产者的工作线程，让它们并行运行，但是...
- 等待线程以 FIFO 顺序完成（根据它们的创建 ID）。
因此，您可以一个接一个地启动线程，以便让它们并行运行，但也可以维护一个 FIFO 队列，它们按 id 升序排列，并且只是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运行所有提交的线程，另一种使用（几乎）同时启动的普通线程。
这会产生如下输出：
使用执行器进行生产...
消息{1，MILLISECONDS（692）}开始并等待...
消息{2，MILLISECONDS（1126）}开始并等待...
消息{0，MILLISECONDS（3403）}开始并等待...... .
消息{3, MILLISECONDS(1017)} 开始并等待...
消息{4, MILLISECONDS(2861)} 开始并等待...
消息{5, MILLISECONDS(2735)} 开始并等待...
消息{6, MILLISECONDS(2068)} 开始并等待...
消息{7, MILLISECONDS(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)} 已加入。
消息 {7, MILLISECONDS(947)} 已加入。
消息 {8, MILLISECONDS(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)} 开始并等待...
消息{7, MILLISECONDS(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)} 已加入。
消息 {6, MILLISECONDS(2000)} 已加入。
消息 {7, MILLISECONDS(2372)} 已加入。
消息 {8, MILLISECONDS(764)} 已加入。
消息 {9, MILLISECONDS(587)} 已加入。
消耗。
您可以在输出中看到，在这两种情况下，线程都是通过循环（几乎）一起启动的，但以 FIFO 有序方式连接。在第一种情况下，您可以看到线程可能以不同的顺序启动，这是启动线程本身的副作用。在普通线程的第二种情况下，碰巧所有线程都run按照它们创建和启动的顺序调用它们的方法，因为这发生在如此短的时间内。但是根据这段代码，每个线程的加入将始终按 id 升序排列。如果您多次运行此代码，您可以在两种情况下实现线程 eg 2 在线程 eg 1 之前在其 run 方法中打印，但是我们在Producer's 方法中等待线程完成的顺序将始终以升序 id 顺序结束.
run所有线程都应该以升序的睡眠顺序而不是升序的 id 顺序退出/完成它们的方法。但是输出将始终按 id 升序排列，因为我们遍历队列并以join有序的方式等待它们。
因此，如果您想以Thread升序的 id 顺序获取 each 的结果，则相应的代码必须在您Producer的生产方法中（在join每个线程之后），而不是在 each 方法的末尾Message（run为了避免额外的同步和线程间通信）。
推荐阅读
- arrays - 将每个包含文件路径的两个数组区分为第三个数组（用于删除）
- c# - 使用构造函数参数设置字段的冗余/清洁方式更少？
- visual-studio-code - 如何将日语文本输入到 Visual Studio Code？
- printing - 如何使用 Stimulsoft.Report.Web 打印 asp.net MVC
- python - Python AutoPep8 格式不适用于最大行长参数
- python-3.x - 任何人都可以指出这句话有什么问题吗？
- java - 优化 Java 中的 If- else if- else 语句
- ruby - 在类顶部调用的 Ruby 方法
- reactjs - 在同一个远程托管面板上从反应前端与 ASP.NET Core Web API 进行通信
- python - Python 中的矩阵使用 Numpy Python 给出错误的结果