Java 设计模式之责任链模式及异步责任链详解

来源：jb51　　时间：2021/11/8 17:19:02　　对本文有异议

一、定义

责任链模式(Chain of Responsibility Pattern)：避免将一个请求的发送者与接受者耦合在一起，让多个对象都有机会处理请求。将接受请求的对象连接成一条链，并且沿着这条链传递请求，直到有一个对象能够处理它为止。

在很多源码都有涉及，如Mybatis拦截器、Filter…

在这里插入图片描述

责任链模式属于行为型模式。

二、普通责任链模式

在这里插入图片描述

抽象处理类：AbstractProcessor

/**
 * 抽象处理类
 */
public abstract class AbstractProcessor {
    // 责任链
    private AbstractProcessor nextProcessor;
 
    public void setNextProcessor(AbstractProcessor nextProcessor){
        this.nextProcessor = nextProcessor;
    }
 
    /**
     * 链路调用
     * @param obj
     */
    public void execute(Object obj) {
        this.processor(obj);
        if (this.nextProcessor != null){
            this.nextProcessor.execute(obj);
        }
    }
 
    public abstract void processor(Object obj);
}

具体类：ParamsProcessor、CheckProcessor、FinalProcessor

public class ParamsProcessor extends AbstractProcessor{
 
    @Override
    public void processor(Object obj) {
        System.out.println("paramsProcessor>>>>>>>>>>" + obj.toString());
    }
}
 
public class CheckProcessor extends AbstractProcessor{
 
    @Override
    public void processor(Object obj) {
        System.out.println("checkProcessor>>>>>>>>>>" + obj.toString());
    }
}
 
public class FinalProcessor extends AbstractProcessor{
 
    @Override
    public void processor(Object obj) {
        System.out.println("finalProcessor>>>>>>>>>>" + obj.toString());
    }
}

测试类：ProcessorMainTest

public class ProcessorMainTest {
    public static void main(String[] args) {
        /**
         * 构建一个链路
         * paramsProcessor -> checkProcessor -> finalProcessor
         */
        AbstractProcessor paramsProcessor = new ParamsProcessor();
        AbstractProcessor checkProcessor = new CheckProcessor();
        AbstractProcessor finalProcessor = new FinalProcessor();
 
        paramsProcessor.setNextProcessor(checkProcessor);
        checkProcessor.setNextProcessor(finalProcessor);
 
        paramsProcessor.execute("Java责任链模式");
    }
}

执行结果

paramsProcessor>>>>>>>>>>Java责任链模式
checkProcessor>>>>>>>>>>Java责任链模式
finalProcessor>>>>>>>>>>Java责任链模式

三、异步责任链模式

异步责任链模式采用责任链模式、生产者消费者模式组合。

在这里插入图片描述

抽象处理类：AbstractAsynProcessor

/**
 * 异步责任链：采用线程、生产者消费者模式实现
 */
public abstract class AbstractAsynProcessor extends Thread {
    // 用于存放当前责任链需要处理的对象
    protected BlockingQueue<Object> queue;
    // 链路对象
    private AbstractAsynProcessor nextProcessor;
 
    /**
     * 执行
     *
     * @param obj
     */
    public void execute(Object obj) throws InterruptedException {
        this.processor(obj);
        if (this.nextProcessor != null) {
            this.nextProcessor.getQueue().put(obj);
        }
    }
 
    @Override
    public void run() {
        // 执行
        try {
            while (true) {
                // 生产者消费者
                Object obj = this.queue.take();
                execute(obj);
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
 
    public abstract void processor(Object obj);
 
    public BlockingQueue<Object> getQueue() {
        return queue;
    }
 
    public void setNextProcessor(AbstractAsynProcessor nextProcessor) {
        this.nextProcessor = nextProcessor;
    }
}

具体类：AsynParamsProcessor、AsynCheckProcessor、AsynFinalProcessor

public class AsynParamsProcessor extends AbstractAsynProcessor {
 
    public AsynParamsProcessor() {
        // 初始化队列
        this.queue = new LinkedBlockingQueue<>();
    }
 
    @Override
    public void processor(Object obj) {
        System.out.println("asynParamsProcessor>>>>>>>>>>" + obj.toString());
    }
}
 
public class AsynCheckProcessor extends AbstractAsynProcessor {
 
    public AsynCheckProcessor(){
        this.queue = new LinkedBlockingQueue<>();
    }
 
    @Override
    public void processor(Object obj) {
        System.out.println("asynCheckProcessor>>>>>>>>>>" + obj.toString());
    }
}
 
public class AsynFinalProcessor extends AbstractAsynProcessor {
 
    public AsynFinalProcessor() {
        this.queue = new LinkedBlockingQueue<>();
    }
 
    @Override
    public void processor(Object obj) {
        System.out.println("asynFinalProcessor>>>>>>>>>>" + obj.toString());
    }
}

测试类：AsynProcessorMainTest

public class AsynProcessorMainTest {
    public static void main(String[] args) throws InterruptedException {
        /**
         * 构建一个链路
         * asynParamsProcessor -> asynCheckProcessor -> asynFinalProcessor
         */
        AbstractAsynProcessor asynParamsProcessor = new AsynParamsProcessor();
        asynParamsProcessor.start();
        AbstractAsynProcessor asynCheckProcessor = new AsynCheckProcessor();
        asynCheckProcessor.start();
        AbstractAsynProcessor asynFinalProcessor = new AsynFinalProcessor();
        asynFinalProcessor.start();
 
        asynParamsProcessor.setNextProcessor(asynCheckProcessor);
        asynCheckProcessor.setNextProcessor(asynFinalProcessor);
 
        Scanner input = new Scanner(System.in);
        while (true) {
            Thread.sleep(1000);
            System.out.print("请输入：");
            asynParamsProcessor.getQueue().put(input.next());
        }
    }
}