java - 自定义消息传递信号量？

Question

假设我有两个主要运行的线程 A 和 B，以及一个异步调用的线程 T。我希望线程 A 等待直到线程 T 收到消息 aMsg，而线程 B 停止直到线程收到消息 msgB T. 我知道如何使用 2 个信号量来做到这一点:

sempahoreA = new Sempahore(0);
sempahoreB = new Sempahore(0);

//in thread A
//code until where to run 
semaphoreA.acquire()

//in thread B
//code until where to run 
semaphoreB.acquire()

//in thread T
if (msgA.equals(msgRecevied)) {
    semaphoreA.release()
} 
if (msgB.equals(msgReceived)) {
    semaphoreB.release()
}

问题是我有多个 A,B,C,... 线程，我不想使用多个信号量。 java.util.concurrent 中是否有一个类可以仅用一个实例替换所有信号量？

synchronizer = //?

//in thread A
//code until where to run 
synchronizer.acquire(msgA)//only let go if msgA is received from thread calling release

//in thread B
//code until where to run 
synchronizer.acquire(msgB)//only let go if msgA is received from thread calling release

//in thread T
if (msgA.equals(msgRecevied)) {
    synchronizer.release(msgA)
} 
if (msgB.equals(msgReceived)) {
    synchronizer.release(msgB)
}//actually here you can call synchronizer.release(msgReceived)

Answer 1

伟大的直觉 m3th0dman。我想你要找的是 Transfer Queue .

这是“一个 BlockingQueue，生产者可以在其中等待消费者接收 元素”。在你的例子中，线程 A 和 B 是生产者，而线程 T 是唯一的 消费者。

简而言之:

1. 创建共享 TransferQueue<Object> synchronizer = new TransferQueue<>();
2. 线程 A 和 B 调用阻塞方法 synchronizer.transfer(new Object());
3. 与此同时，线程 t 调用 synchronizer.take();有空解锁。

这是一个例子:

import java.util.concurrent.TransferQueue;

public class TransferQueueExample {
    public static void main(String[] args) {
        final TransferQueue<Object> synchronizer = new TransferQueue<>();

        for (int i = 0; i < 10; i++) {
            final Thread t = new Thread(new Runnable() {
                @Override
                public void run() {
                    // Do work until
                    // ...
                    System.out.println("Thread " + i + " is transferring.");

                    synchronizer.transfer(new Object()); // This is blocking
                    System.out.println("Thread " + i + " done transferring.");
                }
            }).start();
        }

        final Thread consumer = new Thread(new Runnable() {
            @Override
            public void run() {
                // TODO: Note thatthis will run indefinitely so in a real
                // scenario, find a way of shutting this thread down.
                while (true) {
                    System.out.println("There are about " +
                                       synchronizer.getWaitingConsumerCount()
                                       + " threads waiting");
                    synchronizer.take();
                    System.sleep(1000);
                }
            }
        }).start();
    }
}

我希望嵌套类和匿名类不会太分散注意力。