设计模式之 监听者模式 (Java代码演示)

        所谓监听者模式，我理解的是构建一个容器存放所有被监听的线程或对象，监听每个线程或对象发生的变化，若某个线程或对象触发指定规则，那么则对所有被监听的线程或对象做处理（根据业务需要）。

 代码展示

 Main：测试类

ObServer：每个被监听的对象实现该接口，重写该方法，完成自己的业务

public interface ObServer {

    
    void exceptionHandler();

}

Subject：监听者容器

public interface Subject {


    
    void add(ObServer obServer);

    
    void notifyAllSubject();

}

SubjectHandler：监听者容器的实现类

public class SubjectHandler implements Subject {

    
    private static final List SUBJECTS = Collections.synchronizedList(new linkedList<>());

    @Override
    public void add(ObServer subject) {
        SUBJECTS.add(subject);
    }

    @Override
    public void notifyAllSubject() {
        SUBJECTS.forEach(ObServer::exceptionHandler);
    }
}

Thread1、Thread2 测试对象

模拟两个线程对数据库操作，若Threa1执行时出异常了，那么终止所有线程并对其回滚。

Threa1:

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

Thread2:

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

测试Demo

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

测试结果

在 java.util 包下提供了Observable（监听容器）和 Observer接口（被监听者），用法跟我们完全一致，只需实现Observer的update方法，把每个Observer子类添加到监听容器中。

注意：在调用监听容器的通知所有servers时首先调用一下其setChanged方法

将changed变为true（默认flase）