责任链模式（Responsibility Chain Pattern）在实际场景中的应用

背景

• 最近项目中遇到这样的业务场景：用户可以将自己的视频通过系统分发到下游平台需要经历若干个步骤：
  • 初始化分发任务，记录一条分发任务，状态为分发中
  • 视频格式转换，不同的平台所需要的格式不同，此处需要调用第三方格式转换接口，同时还需要提供一个格式转换通知接口，供第三方处理完成后通知系统处理结果。
  • 视频码率转换，生成一个低码率的文件，供用户预览。与上一步一样，此处也需要调用第三方转码接口，同时还需要提供一个转码通知接口，供第三方处理完成后通知系统处理结果。
  • 分发到下游平台
• 经过业务分析后发现，该流程每个步骤分的很清楚，每一步都有自己的职责，是责任链模式的使用场景，与标准责任链场景不同的是，该场景中由第三方回调进行流程的串联，流程图如下：

graph LR
初始化processor-->转格式processor
转格式processor-->第三方格式转换
第三方格式转换-->转码processor
转码processor-->第三方转码
第三方转码-->分发processor
           

• 实现该业务场景需要：
  • 初始化processor（InitProcessor）
  • 转格式processor（FormatProcessor）
  • 转码processor（TranscodeProcessor）
  • 分发processor（DistributeProcessor）
  • 转格式通知接口（Controller）
  • 转码通知接口（Controller）

processor的设计

• 抽象父类：DistributionProcessor，可以看到该数据结构为链表结构

public abstract class DistributionProcessor {
    
    /**
    * 持有下一个处理步骤对象
    */
    protected DistributionProcessor nextProcessor;

    /**
    * 设置下一个处理步骤
    */
    public DistributionProcessor setNextStep(DistributionProcessor nextProcessor){
        this.nextProcessor = nextProcessor;
        return this.nextProcessor;
    }
    
    /**
    * 获取下一个处理步骤
    */
    public DistributionProcessor getNextProcessor(){
        return this.nextProcessor;
    }

    /**
    * 每一步的处理方法，具体实现交给具体的processor
    */
    public abstract void process(DistributeRequest request);

    /**
    * 供接口调用，由于是异步的(需要第三方处理)，需要接口调用该方法来串流程
    */
    public abstract void postProcess(String id, boolean processResult);
}
           

• 子类processor较为类似，唯有转格式和转码的processor有不同，他们需要实现postProcess()方法供接口调用，下面以InitProcessor和TranscodeProcessor为例：
  • InitProcessor

    public class InitProcessor extends DistributionProcessor {
    
        @Override
        public void process(DistributeRequest request) {
            // 处理的相关逻辑
            nextProcessor.process(request); // 交给下一个处理步骤
        }
    
        @Override
        public void postProcess(String id, boolean processResult) {
    
        }
    }
               

  • TranscodeProcessor

    public class TranscodeProcessor extends DistributionProcessor {
    
        @Override
        public void process(DistributeRequest request) {
            // 调用第三方转码
        }
    
        /**
        * 通知接口收到转码结果通知后，调用该方法继续下一步操作
        */
        @Override
        public void postProcess(String id, boolean processResult) {
            // 相关处理
            nextProcessor.process(request);// 交给下一个处理步骤
        }
    }
               

• 项目启动时，将processor都串起来：

@Bean
public CommandLineRunner commandLineRunner(DistributionProcessor initProcessor,
                                           DistributionProcessor formatProcessor,
                                           DistributionProcessor transcodeProcessor,
                                           DistributionProcessor distributeProcessor){
    return clr->{
        initProcessor
            .setNextStep(formatProcessor) 
            .setNextStep(transcodeProcessor)
            .setNextStep(distributeProcessor);
    };
}
           

结果通知接口

• 本场景中涉及到两个结果通知接口，代码较为类似，以转码通知接口为例：

@ApiOperation("转码通知")
@GetMapping(DistributionReqUrls.DISTRIBUTION_NOTIFICATION_TRANSCODE)
public void notifyTranscode(String id, Boolean processResult){
    transcodeProcessor.postProcess(attachmentId, success);
}