多个对象之间的java线程同步答案

【问题标题】：java thread sync between multiple objects多个对象之间的java线程同步
【发布时间】：2017-01-08 09:38:43
【问题描述】：

我有一个多线程多对象系统，其中有一个manager 和多个workers。我需要将workers 与manager 同步，如下所示：

manager做某事，给workers下命令，然后让workers并行运行，相互独立。当他们完成这一轮时，他们必须等待manager 给他们新的任务或命令。只有当所有workers 都完成了他们之前的工作时，manager 才会发布新订单。

我需要使用线程来实现它以避免忙等待。然而，同步令人困惑。

有什么想法吗？

【问题讨论】：

docs.oracle.com/javase/7/docs/api/java/util/concurrent/…
我有很多想法；我可以给你一个每日费率。否则，请编写一些代码并找出您遇到的问题；然后在这里发帖寻求帮助。我们不是代码编写服务。提示：查看CyclicBarrier。
@BoristheSpider 我当然不想/不想要代码编写服务。只是一个提示。 CyclicBarrier 解决了这个问题。
如果您能描述一下为什么经理只有在所有工人都完成工作后才下达新订单会有所帮助。
@Sidias-Korrado 工作人员正在使用共享资源。在分配新订单之前应该释放资源。

标签： java multithreading synchronization

【解决方案1】：

编辑：我错过了一个重要部分，即只有在所有任务都完成后才能到达新任务。因此使用 LinkedBlockingQueue 并不是最好的解决方案。我推荐使用 boris-the-spider 推荐的 CyclicBarrier。

您可以使用LinkedBlockingQueue

设置固定容量。

manager可以put任务，workers可以使用函数take等待。

【讨论】：

排序：我想如果您有 n 任务和 n 工作人员 - 并且任务需要 足够长 才能运行 - 那么当您发布 n 任务时到BlockingQueue，它会让每个工人接受一个任务并处理它。但这并没有解决主人等待工人的问题。使用非常简单的ExecutorService.invokeAll，OP 可能会更好。
master 可以通过查看queue.size() >= n 和 sleep-wait 来等待 worker。或者工人可以表明他们的状态，有很多方法可以做到。
这行不通，除非工作人员将某些东西放回 another 队列中 - 所以你需要两个队列。我不是说你的方法行不通，我只是说你在重新发明轮子；对于并发代码，这通常是一个非常糟糕的主意。 ExecutorService 代码是 JDK 的一部分，因此经过了很好的测试。它完全符合我们的需要......
我很好奇你为什么认为queue.size() >= n 解决方案行不通。队列保存等待的任务准备好接受。
哪个队列？工作项队列在开始时将有一个size 或n，然后在完成所有工作后大小为0。除非将项目放回队列中，否则它将继续具有 0 的 size。此外，您的主线程将不得不忙于确定队列的大小——这很糟糕。

【解决方案2】：

正如@boristhespider 所建议的，我对manager 和workers 都使用了CyclicBarrier。

每个worker 完成其任务后，都会调用barrier.await()。然后对于manager，我检查是否barrier.getNumberWaiting()==NumWorkers。如果是true，则更新每个worker的任务，然后调用barrier.await()。

【讨论】：

如何在不忙于等待经理的情况下查看barrier.getNumberWaiting()？

【解决方案3】：

维护 2 个阻塞队列 1.对于任务 2.免费工人

让工作人员通过回调通知经理，将他们添加到空闲工作队列

在经理线程中，您可以检查可用的工人。

快速实施

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;

public class ManagerWorker {
    public static void main(String[] args) {

        ExecutorService service = Executors.newCachedThreadPool();
        BlockingQueue<String> taskQueue = new LinkedBlockingQueue<>();

        Manager m = new Manager(5, taskQueue);
        service.submit(m);

        for (int i = 0; i < 5; i++) {
            service.submit(new Worker(m, taskQueue));
        }

    }
}

class Manager implements Runnable {

    int workerCount;
    BlockingQueue<Worker> workerqueue = new LinkedBlockingQueue<>();
    BlockingQueue<String> taskQueue;

    public Manager(int workerCount, BlockingQueue<String> taskQueue) {
        this.workerCount = workerCount;
        this.taskQueue = taskQueue;
    }

    public void callBackForFreeNotification(Worker worker) {
        workerqueue.add(worker);
    }

    @Override
    public void run() {
        while (true) {
            try {
                int i = 0;
                while (i < workerCount) {
                    workerqueue.take();
                    i++;
                }

                System.out.println("Manager Worker available");

                // add task to task queue here
                for (int j = 0; j < workerCount; j++) {
                    taskQueue.add("task");
                }
                System.out.println("Manager task added");
            } catch (InterruptedException e) {

                e.printStackTrace();
            }
        }
    }

}

class Worker implements Runnable {

    private Manager manager;
    private BlockingQueue<String> taskQueue;

    public Worker(Manager manager, BlockingQueue<String> taskqueue) {
        this.manager = manager;
        this.taskQueue = taskqueue;
    }

    @Override
    public void run() {
        while(true){


        try {

            System.out.println("Worker - i have no work");
            manager.callBackForFreeNotification(this);
            taskQueue.take();
            System.out.println("Worker working");
            Thread.sleep(2000);
            System.out.println("Worker Done with work");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        }
    }

}

【讨论】：