Java 代码实例解析设计模式之监听者模式

public interface ObServer {
 
    /**
     * 当某一个被监控的对象发生变化时
     * 所有实现该方法处理方法
     */
    void exceptionHandler();
 
}

public interface Subject {
 
 
    /**
     * 添加被观察对象
     */
    void add(ObServer obServer);
 
    /**
     * 通知所有被观察者完成自己的 exceptionHandler 方法
     */
    void notifyAllSubject();
 
}

public class SubjectHandler implements Subject {
 
    /**
     * 存放被监听对象
     */
    private static final List<ObServer> SUBJECTS = Collections.synchronizedList(new LinkedList<>());
 
    @Override
    public void add(ObServer subject) {
        SUBJECTS.add(subject);
    }
 
    @Override
    public void notifyAllSubject() {
        SUBJECTS.forEach(ObServer::exceptionHandler);
    }
}

public class Thread1 implements ObServer, Runnable {
 
    @SneakyThrows
    @Override
    public void run() {
        System.out.println("thread1 run");
        Thread.sleep(1000);
        System.out.println("t1 end");
    }
 
    @SneakyThrows
    @Override
    public void exceptionHandler() {
        System.out.println("thread1 rollback.....");
    }
}

public class Thread2 implements ObServer, Runnable {
 
    private static Thread CURRENT_THREAD;
 
    private static volatile boolean FLAG = false;
 
    @SneakyThrows
    @Override
    public void run() {
        CURRENT_THREAD = Thread.currentThread();
        System.out.println("thread2 running");
        int count = 0;
        while (!FLAG) {
            System.out.println(count);
            count++;
        }
        System.out.println("thread2 end");
    }
 
 
    @SneakyThrows
    @Override
    public void exceptionHandler() {
        FLAG = true;
        CURRENT_THREAD.interrupt();
        System.out.println("thread2 rollback.....");
    }
}

public static void main(String[] args) throws InterruptedException {
 
        // 创建监听容器
        Subject subject = new SubjectHandler();
        Thread1 thread1 = new Thread1();
        Thread2 thread2 = new Thread2();
        subject.add(thread1);
        subject.add(thread2);
        
        CompletableFuture.supplyAsync(() -> {
            new Thread(thread1).start();
            try {
                Thread.sleep(10);
                int a = 1 / 0; // 模拟线程1报错
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return true;
        }).exceptionally((error) -> {
            subject.notifyAllSubject();
            return false;
        });
 
        CompletableFuture.supplyAsync(() -> {
            new Thread(thread2).start();
            return true;
        }).exceptionally((error) -> {
            subject.notifyAllSubject();
            return false;
        });
 
        // main thread await
        Thread.sleep(60 * 1000);
    }