多线程observer模式

原来java的观察者模式

Snipaste_2020-10-04_14-13-41.png

• subject

public class Subject {

    private List<Observer> observers = new ArrayList<Observer>();

    private int state;

    public int getState() {
        return this.state;
    }

    public void setState(int state) {
        if (state == this.state) {
            return;
        }
        this.state = state;
        //一有改变马上通知观察者
        notifyAllObserver();
    }

    public void attach(Observer observer) {
        //observer在构建时就把自己加入
        observers.add(observer);
    }

    private void notifyAllObserver() {
        for(Observer obsever : observers){
            obsever.update();
        }
    }
}

• obsever接口类

public abstract class Observer {

    protected Subject subject;

    public Observer(Subject subject) {
        this.subject = subject;
        this.subject.attach(this);
    }

    public abstract void update();

}

• 具体的观察者

public class BinaryObserver extends Observer {

    public BinaryObserver(Subject subject) {
        super(subject);
    }

    @Override
    public void update() {
        System.out.println("Binary String:" + Integer.toBinaryString(subject.getState()));
    }
}

public class OctalObserver extends Observer {

    public OctalObserver(Subject subject) {
        super(subject);
    }

    @Override
    public void update() {
        System.out.println("Octal String:" + Integer.toOctalString(subject.getState()));
    }
}

• 运行类

public class ObserverClient {
    public static void main(String[] args) {

        final Subject subject = new Subject();
        new BinaryObserver(subject);
        new OctalObserver(subject);
        System.out.println("==================");
        subject.setState(10);
        System.out.println("==================");
        //这一次不会做任何通知
        subject.setState(10);

        System.out.println("==================");
        subject.setState(15);
    }
}

多线程版本（监控线程的生命周期和一些数据）

• 主题类

public abstract class ObservableRunnable implements Runnable {

    //下面两个步骤相当于原来的观察者模式
    //主题类维护observerslist,只不过这里是主题类的构造函数自动加入观察者
    //原来是有一个attach方法，observer自己去调用加入自己
    final protected LifeCycleListener listener;
    public ObservableRunnable(final LifeCycleListener listener) {
        this.listener = listener;
    }
    
    //相当于notifyAllObserver()
    protected void notifyChange(final RunnableEvent event) {
        listener.onEvent(event);
    }
    
    //枚举状态类
    public enum RunnableState {
        RUNNING, ERROR, DONE
    }
    
    //可以提供给观察者的内容写成一个内部类
    public static class RunnableEvent {
        private final RunnableState state;
        private final Thread thread;
        private final Throwable cause;

        public RunnableEvent(RunnableState state, Thread thread, Throwable cause) {
            this.state = state;
            this.thread = thread;
            this.cause = cause;
        }

        public RunnableState getState() {
            return state;
        }

        public Thread getThread() {
            return thread;
        }

        public Throwable getCause() {
            return cause;
        }
    }
}

• observer接口类（原来是写成抽象类）

public interface LifeCycleListener {
    
    void onEvent(ObservableRunnable.RunnableEvent event);
}

• 观察者具体实现类，运行主题类runnable是在这里批量创建的。

public class ThreadLifeCycleObserver implements LifeCycleListener {

    private final Object LOCK = new Object();

    public void concurrentQuery(List<String> ids) {
        if (ids == null || ids.isEmpty())
            return;

        for(final String id : ids){
            new Thread(new ObservableRunnable(this) {
                public void run() {
                    try {
                        notifyChange(new RunnableEvent(RunnableState.RUNNING, Thread.currentThread(), null));
                        System.out.println("query for the id " + id);
                        Thread.sleep(1000L);
                        notifyChange(new RunnableEvent(RunnableState.DONE, Thread.currentThread(), null));
                    } catch (Exception e) {
                        notifyChange(new RunnableEvent(RunnableState.ERROR, Thread.currentThread(), e));
                    }
                }
            },id).start();
        }
    }
    
    //运行的线程有多个，这个有观察者负责创建。但是observer只有一个，所以要加锁
    public void onEvent(ObservableRunnable.RunnableEvent event) {
        synchronized (LOCK) {
            System.out.println("The runnable [" + event.getThread().getName() + "] data changed and state is [" + event.getState() + "]");
            if (event.getCause() != null) {
                System.out.println("The runnable [" + event.getThread().getName() + "] process failed.");
                event.getCause().printStackTrace();
            }
        }
    }
}

• 运行类

public class ThreadLifeCycleClient {
    public static void main(String[] args) {
        new ThreadLifeCycleObserver().concurrentQuery(Arrays.asList("1", "2"));
    }
}

Snipaste_2020-10-04_15-45-28.png