java - 观察者模式和线程

Question

在我们开始高兴之前，SO 上的建议重复项都没有回答我的问题。

尝试这样做:我有一个观察者模式。当我想发送 Events 时，我想在 Java thread 中进行.

问题

1. 我不想每次都创建一个新线程。我认为这比保留对线程的引用在计算上更昂贵。

2. 我需要向线程传递一个对象，即事件。

   public class EventBus implements Runnable
   {
       private Thread t1;
   
       public EventBus()
       {
           t1 = new Thread(this);
       } 
   
       public void notify(Event event)
       {
           t1.start();
       }
   
       @Override
       public void run()
       {
           for(Listener l : list)
              l.handle(event);
       }
   }
   

那么我如何将我的事件对象传递给我的 run() 而不是每次都使用这样的东西？

Runnable r = new EventBus(param_value);
new Thread(r).start();

我知道将参数传递给 Thread 的方法是将它们放在 Thread 构造函数中，我不想那样，我在这里了解到 How can I pass a parameter to a Java Thread? .

每次调用 notify() 时，我都需要将新的 Event 传递给线程。在程序执行过程中，notify()会被多次调用，并且会携带不同的Event实例。

Answer 1

我举了一个简单的例子来展示我最近是如何进行异步事件发布的。检查一下，我认为它很容易解释:

主要方法:

public static void main(String[] args) throws InterruptedException {
    Observer obs = new Observer();
    EventBus.subscribe(obs, SomeEventImp.class);

    SomeEventImp evt = new SomeEventImp(new Object(), "This is the value");

    EventBus.publishAsync(evt);

    Thread.sleep(Long.MAX_VALUE);
} 

观察者界面:

public interface IObserver {

    public void update(AEvent event);
}

和观察者实现:

public class Observer implements IObserver {
    @Override
    public void update(AEvent event) {
        System.out.println("I got and event from " + event.getSource() + " with a value of " + event.getValue());
    }
}

AEvent 类:

public abstract class AEvent<T> {

    protected final T value;
    protected final Object source;

    public AEvent(Object source, T value) {
        this.value = value;
        this.source = source;
    }

    public Object getSource() {
        return source;
    }

    public T getValue() {
        return value;
    }
}

事件总线:

public class EventBus {

    // our observers
    private static HashMap<IObserver, Class<?>> m_Observers = new HashMap<IObserver, Class<?>>();
    // our incoming events
    private static BlockingQueue<AEvent<?>> incoming = new LinkedBlockingQueue<AEvent<?>>();

    // start our internal thread
    static {
        new Thread(new DelegationThread()).start();
    }

    // subscribe an observer
    public static void subscribe(IObserver obs, Class<?> evtClass) {
        synchronized (m_Observers) {
            m_Observers.put(obs, evtClass);
        }
    }

    // publish and event
    public static void publishAsync(AEvent<?> event) {
        incoming.add(event);
    }

    private static class DelegationThread implements Runnable {
        @Override
        public void run() {
            while (true) {
                try {
                    AEvent<?> evnt = incoming.take();
                    synchronized (m_Observers) {
                        for (Entry<IObserver, Class<?>> entry : m_Observers.entrySet()) {
                            if (entry.getValue() == evnt.getClass()) {
                                entry.getKey().update(evnt);
                            }
                        }
                    }
                }
                catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

最后是事件实现:

public class SomeEventImp extends AEvent<String> {

    public SomeEventImp(Object source, String value) {
        super(source, value);
    }
}

这是输出:

I got and event from java.lang.Object@5e1387c6 with a value of This is the value

很明显，您可能想要稍微清理一下……我只是在几分钟内把它拼凑在一起，并没有真正检查得那么好。