Java：新线程中的观察者模式通知

Question

我有以下问题。给定观察者模式的接口 EventNotifier：

public interface EventNotifier {
    void newEvent(final String value);
}

实现此接口的类可以注册到另一个类，该类经常调用newEvent方法。该接口是由外部库提供的，所以我无法更改它。到目前为止，我使用匿名类实现了它：

Thread t = new Thread(new Runnable() {

    @Override    
    public void run() {

      watcher = new Watcher(new EventNotifier() {

          @Override
          public void newEvent(String value) {
              //do some stuff
              //will be called more than 20 times per second
          }
       });
 });
 t.start();

为了更好的代码可读性，我想将这个匿名类公开到一个新类中，该类扩展了 Thread（因为处理应该与其他事物并行）。

我如何编写一个什么都不做（没有无限循环等）但等待调用 newEvent 方法的线程？问题是，newEvent 每秒会被调用超过 20 次，所以我不能为每个调用启动一个新线程，但整个事情应该在一个线程中。

我希望你能解决问题，有人可以帮助我。

Answer 1

使您的帖子令人困惑的是EventNotifier 实际上是一个观察者/侦听器（它接收事件，它不会触发它们），而Watcher 实际上是通知者（它是观察者创建一个事件并调用newEvent 方法）。

从现在开始，我将使用术语 observable 和 observer。 observable 触发事件，因此调用观察者的 newEvent 方法。

如果您希望在单独的线程中完成事件处理，请使用BlockingQueue。启动一个无限循环的线程，并在每次迭代时尝试从队列中take()。向 observable 注册一个观察者，它只是将接收到的事件和put() 放在阻塞队列中。

Answer 2

您可以使用Executor 来避免编码BlockingQueue 和手动轮询线程。

在您的主课中，您将有类似的内容：

Executor eventExecutor = Executors.newSingleThreadExecutor();
// ...
watcher = new Watcher(new EventNotifier() {
    public void newEvent(final String value) {
        eventExecutor.execute(new ConcurrentEventHandler(value));
    }   
}); 

以及在后台线程中执行处理的并发事件处理程序：

class ConcurrentEventHandler implements EventNotifier, Runnable {
        private final String value;

        public ConcurrentEventHandler(String value) {
            this.value = value;
        }

        public void newEvent(final String value) {
            // do some stuff
        }

        public void run() {
            // executed in background thread
            newEvent(value);
        }
    }

我在这里实现了EventNotifier，但当然没有必要

Answer 3

使用普通的旧wait/notifyAll：

// we need final object to synchronize your code and library code on it
// it's convenient to make this object hold all needed data to be passed from library as well
// in your case AtomicBoolean should suffice (we can't use simple `final Boolean`, since it would be impossible to assign new value to it, as we need in code below).

final AtomicBoolean called = new AtomicBoolean(false);

EventNotifier en = new EventNotifier() {
          @Override
          public void newEvent(String value) {
               // this will be called by your external library
               synchronized(called) {
                    called.set(true); called.notifyAll();
               }
          }
       };

Thread t = new Thread(new Runnable() {
    @Override    
    public void run() {
         synchronized(called) {
            // wait here until library call occurs
            while (!called.get()) {
                 try {
                     called.wait();
                 } catch (InterruptedException e) {
                   // handle exception as desired
                 }
            }
            // reset called flag asap, so we will know when next call occurs
            called.set(false);
            ... // do your stuff
         }
    );
 });
t.start();

有关 Java 多线程编程的一般介绍，请阅读tutorial。然后，如果您对主题感兴趣，请阅读 Goetz 的“Java 并发实践”。

如果您需要处理从库传递到newEvent 的value，则需要某种BlockingQueue 而不是简单的Boolean。