不同池java中线程之间的通信答案

【问题标题】：communication between threads in the different pools java不同池java中线程之间的通信
【发布时间】：2016-12-26 18:08:14
【问题描述】：

我有两个本地线程池，一个池有 4 个线程，第二个池有 5 个线程。

我希望这两个池相互通信。

例如，第一个池的第二个线程（1.2）与第二个池的第五个线程（2.5）通信，即

1.2 -> 2.5
1.1 -> 2.2
1.3 -> 2.1
1.4 -> 2.3

1.2发送完消息给2.5，想把另一条消息发送到第二个池，但是2.5还在忙，但是2.4如果有空闲就可以了处理来自 1.2 的消息

如何使第一个池中的线程与第二个池中的第一个空闲线程通信？

如何在java中实现它？

也许我应该使用消息代理或类似的东西？（或BlockingQueue、Exchanger/Pipereader）

谢谢

【问题讨论】：

标签： java multithreading concurrency threadpool

【解决方案1】：

（您的示例不清楚，但我认为您要求的方案是一个池中的线程不关心另一个池中的哪个线程获取消息。）

可能有很多方法可以做到这一点，但一个简单的方法是：

为每个池创建有界消息队列
每个池中的每个线程都从其池的队列中读取消息
一个池中的线程通过将消息添加到另一个池的队列来向另一个池发送消息。

消息代理也可以工作，但它可能是过度杀戮。您很可能不想要成熟的消息代理的可靠性/持久性/分发。

【讨论】：

【解决方案2】：

如何使第一个池中的线程与第一个空闲线程通信来自第二个池的线程？

我不确定您是否有任何其他特定需求，但如果两个池都是本地的，并且您只是愿意实现典型的生产者-消费者模式，其中 N 线程（作为池的一部分）充当生产者，另一个M-Threads（作为另一个池的一部分）充当消费者，您不在乎第二个池的哪个线程实例处理消息，我会通过 - BlockingQueue 实现。

您获取BlockingQueue 的实例（如ArrayBlockingQueue 或LinkedBlockingQueue 或PriorityBlockingQueue，包java.util.concurrent 中的实现很少）并在实际池线程之间共享此实例，同时限制它 - @987654327 @ 只能由消费者线程和任何消费者线程完成。

如何在java中实现它？

你可以像下面这样创建你的池，

ExecutorService pool_1 = Executors.newFixedThreadPool(4);

ExecutorService pool_2 = Executors.newFixedThreadPool(4);

然后您将实际线程分配给共享阻塞队列的这些池。可以像下面这样创建线程——它只是一个伪代码。

public class Pool1Runnable implements Runnable {

   private final BlockingQueue queue;

   public Pool1Runnable(BlockingQueue queue){
     this.queue=queue;
   }

    @Override
    public void run() {
        System.out.println("Pool1Runnable");
    }

}

现在您为 pool2 编写线程实现，并确保它们的 run() 实现在队列中使用 take()。

您创建池实例、线程实例 - 为生产者和消费者分开（为所有线程提供单个队列实例，使其充当通信通道），然后使用池执行这些线程实例。

希望对你有帮助！！

【讨论】：

【解决方案3】：

其他人指出的最直接的方法是在池之间有一个BlockingQueue。如果我没记错的话，您的问题与多个生产者和多个消费者分别发送和处理消息相同。

这是您可以构建的一种实现。添加了 cmet 的参数很少，您可以根据您的问题场景调整它们。基本上，您有 2 个池和一个池来并行调用生产者和消费者。

public class MultiProducerConsumer {

private static final int MAX_PRODUCERS = 4;
private static final int MAX_CONSUMERS = 5;

private ExecutorService producerPool = new ThreadPoolExecutor(2, MAX_PRODUCERS, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>());
private ExecutorService consumerPool = new ThreadPoolExecutor(2, MAX_CONSUMERS, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>());

//ThreadPool for holding the main threads for consumer and producer
private ExecutorService mainPool = new ThreadPoolExecutor(2, 2, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>());

/**
 * Indicates the stopping condition for the consumer, without this it has no idea when to stop
 */
private AtomicBoolean readerComplete = new AtomicBoolean(false);

/**
 * This is the queue for passing message from producer to consumer.
 * Keep queue size depending on how slow is your consumer relative to producer, or base it on resource constraints
 */
private BlockingQueue<String> queue = new ArrayBlockingQueue<>(1);

public static void main(String[] args) throws InterruptedException {
    long startTime = System.currentTimeMillis();
    MultiProducerConsumer multiProducerConsumer = new MultiProducerConsumer();
    multiProducerConsumer.process();
    System.out.println("Time taken in seconds - " + (System.currentTimeMillis() - startTime)/1000f);
}

private void process() throws InterruptedException {
    mainPool.execute(this::consume);
    mainPool.execute(this::produce);
    Thread.sleep(10); // allow the pool to get initiated
    mainPool.shutdown();
    mainPool.awaitTermination(5, TimeUnit.SECONDS);
}

private void consume() {
    try {
        while (!readerComplete.get()) { //wait for reader to complete
            consumeAndExecute();
        }
        while (!queue.isEmpty()) { //process any residue tasks
            consumeAndExecute();
        }
    } catch (InterruptedException e) {
        e.printStackTrace();
    } finally {
        try {
            consumerPool.shutdown();
            consumerPool.awaitTermination(5, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

}

private void consumeAndExecute() throws InterruptedException {
    if (!queue.isEmpty()) {
        String msg = queue.take(); //takes or waits if queue is empty
        consumerPool.execute(() -> {
            System.out.println("c-" + Thread.currentThread().getName() + "-" + msg);
        });
    }
}


private void produce() {
    try {
        for (int i = 0; i < MAX_PRODUCERS; i++) {
            producerPool.execute(() -> {
                try {
                    String random = getRandomNumber() + "";
                    queue.put(random);
                    System.out.println("p-" + Thread.currentThread().getName() + "-" + random);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }
    } finally {
        try {
            Thread.sleep(10); //allow pool to get initiated
            producerPool.shutdown();
            producerPool.awaitTermination(5, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        readerComplete.set(true); //mark producer as done, so that consumer can exit
    }
}

private int getRandomNumber() {
    return (int) (Math.random() * 50 + 1);
}

}

这是输出：

p-pool-1-thread-2-43
p-pool-1-thread-2-32
p-pool-1-thread-2-12
c-pool-2-thread-1-43
c-pool-2-thread-1-12
c-pool-2-thread-2-32
p-pool-1-thread-1-3
c-pool-2-thread-1-3
Time taken in seconds - 0.1

【讨论】：