Observer Pattern and Thread

Question

Before we get trigger happy none of the suggested duplicates here on SO have answered my question.

Trying to do this: I have an Observer Pattern. When I want to dispatch Events I want to do it in a Java thread.

Problem

1. I don't want to create a new thread each time. I think that would be more computationally expensive than keeping the reference to the thread.

2. I need to pass an Object to the thread, which is the Event.

   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);
       }
   }
   

So how do I pass my Event object to my run() without using something like this each time?

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

I am aware that the way to pass parameters to the Thread is to put them in the Thread constructor, I don't want that, I learned that here How can I pass a parameter to a Java Thread?.

I need to pass the new Event to the thread each time notify() is called. During the execution of the program notify() will be called many times and it will carry different Event instances.

Answer 1

I put together a simple example that sorta shows how ive been doing async event publications as of late. Check it out, I think its pretty self explanatory:

The main method:

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);
} 

The Observer interface:

public interface IObserver {

    public void update(AEvent event);
}

And Observer implementation:

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());
    }
}

The AEvent class:

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;
    }
}

The event bus:

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();
                }
            }
        }
    }
}

and finally the event implementation:

public class SomeEventImp extends AEvent<String> {

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

And heres the output:

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

Clearly you would want to clean this up a bit... I did just sorta slap this together in a few minutes, and didnt really check it all that well.

Answer 2

Obviously you cannot use a constructor parameter, since you don't want to call a constructor each time you start a thread on the computation you want to do. So the starting of the thread is going to have to be distinct from it getting the work it is supposed to do. It is going to have to get started, then wait until that computation is ready to do, obtain an object that tells it what to work on, and then start work on it.

Since you seem to be worried about creating new threads, I would assume you would want a thread object that did one computation, then got itself ready for another one, waited for it, and began again. I don't see anything wrong with an infinite loop for that.

Other suggestions will likely be made about details of the above implementation -- queue, dispatcher object, etc. -- But this is the answer to your actual question. That thread can have an internal object that is set by a dispatcher, and the dispatcher manages a queue. Whatever. They're all going to be variations of an object obtaining some data after its creation, which of course objects do all the time.