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

一、定义

责任链模式(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());
       }
   }
}

执行结果

请输入：Java异步责任链
asynParamsProcessor>>>>>>>>>>Java异步责任链
asynCheckProcessor>>>>>>>>>>Java异步责任链
asynFinalProcessor>>>>>>>>>>Java异步责任链

来源：https://blog.csdn.net/qq_33375499/article/details/121186261