【笔记】高并发编程第二阶段13讲、使用观察者设计模式观察线程的生命周期

爱被打了一巴掌 2022-03-07 00:26 258阅读 0赞

**现在有个需求，我有一个操作，但操作时间多，于是拆分多个子操作（可以并行的子操作），需要监控子操作的状态，甚至数据等。**

为了实现一点，我们可以使用观察者设计模式，首先，子操作可以理解为一个subject，继承runnable，里面包含了observer，及状态变更时，通知observer的方法。

observer，定义一个操作方法，然后运行多个subject。

1.先定义一个subject，线程状态或者里面的数据的变化会主动通知

public abstract class ObservableRunnable implements Runnable {
    
        final protected LifeCycleListener listener; //observer
    
        public ObservableRunnable(final LifeCycleListener listener) {
            this.listener = listener;
        }
    
        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;
            }
        }
    }

2.定义一个observer

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

public class ThreadLifeCycleObserver implements LifeCycleListener {
    
        private final Object LOCK = new Object();
    
        public void concurrentQuery(List<String> ids) {
            if (ids == null || ids.isEmpty())
                return;
    
            ids.stream().forEach(id -> new Thread(new ObservableRunnable(this) {
                @Override
                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());
        }
    
        @Override
        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();
                }
            }
        }
    }

3.定义一个client

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