在 Java 线程中等待和通知给定时间间隔

Question

我正在研究如下用例。我是多线程的新手，在使用它时遇到了这个问题。

1. 我在网络上广播了一个事件。
2. 所有听众都收到了，他们用他们的信息单播了我。
3. 这是在回调方法中收到的，如下所示，我将获得 N 个未知数量的回调线程。取决于当时的听众。
4. 我必须收集所有订阅者的列表。

我必须等待至少 10 秒才能让所有订阅者回复我。

        //Sender

            public void sendMulticastEvent() {
                api.sendEvent();
                /* after sending event wait for 15 sec so call back can collect all the subscribers */
                //start waiting now
            }


    //Callback method
            public void receiveEventsCallback(final Event event) {
                //i will receive multiple response threads here..  
                //event object will have the topic and subscribers details, which i will collect here
               list.add(event)

               notify()
               //notify thread here so i have a cumulative list of all received events.
            }

我只关心如何..？

1. 在 sendMulticast 事件处开始等待 X 秒
2. 在收到的所有事件都添加到列表后，在 receiveEventsCallback() 处通知。

我已经阅读了关于等待和通知、倒计时和障碍的理论。但我不确定哪个更好，因为我在多线程方面的经验很差。

Answer 1

1. 在 sendMulticast 事件处开始等待 X 秒

只需使用带有超时参数的wait() 版本。

请注意，您应该在每次成功的wait() 调用（即返回事件）后手动更新超时值。

1. 在收到的所有事件都添加到列表后，在 receiveEventsCallback() 处通知。

您的问题坚持认为您不知道您的网络中有多少听众。你怎么知道他们都收到了（并回复了）事件？

发件人的唯一方法是等待 X 秒并处理所有可用的回复直到那一刻。

Answer 2

如果您知道会收到多少回复 - 假设每个回复都会触发新线程的创建 - 请使用 CyclicBarrier。

https://docs.oracle.com/javase/7/docs/api/java/util/concurrent/CyclicBarrier.html

示例：

CyclicBarrier barrier = new CyclicBarrier(3);

    Runnable thread = new Runnable()
    {
            @Override
            public void run()
            {
                    try
                    {
                            barrier.await();
                            for (int i = 0; i < 10; i++)
                            {
                                    System.out.printf("%d%n", i);
                            }
                    }
                    catch (InterruptedException | BrokenBarrierException ex)
                    {
                            ex.printStackTrace();
                            // handle the exception properly in real code.
                    }

            }
    };

直到第三个barrier.await()，每个线程都会等待。