異步責任鏈

一、建立抽象記錄器類

public interface IRequestProcessor {
    void process(Request request);
}

二、記錄器實現類1

import java.util.concurrent.LinkedBlockingQueue;

public class PrevProcessor extends Thread implements IRequestProcessor {
    // 阻塞隊列
    private LinkedBlockingQueue<Request> requests = new LinkedBlockingQueue<>();

    private IRequestProcessor nextProcessor;

    private volatile boolean isFinish = false;

    public PrevProcessor(IRequestProcessor nextProcessor) {
        this.nextProcessor = nextProcessor;
    }
    public PrevProcessor() { }

    public void shutdown(){
        isFinish = true;
    }

    @Override
    public void run() {
        while (!isFinish){
            try {
                // 阻塞式獲取
                Request request = requests.take();
                // 處理邏輯
                System.out.println("prevProcessor::" + request);

                // 交給下一個責任鏈
                if(null != nextProcessor){
                    nextProcessor.process(request);
                }

                // 線程退出機制
                shutdown();
            } catch (InterruptedException e) {
                // 全部和阻塞相關的方法都會拋出 InterruptedException
                e.printStackTrace();
            }
        }
    }


    @Override
    public void process(Request request) {
        // 加到隊列
        requests.add(request);
    }


}

三、記錄器實現類2

import java.util.concurrent.LinkedBlockingQueue;

public class SaveProcessor extends Thread implements IRequestProcessor {
    // 阻塞隊列
    private LinkedBlockingQueue<Request> requests = new LinkedBlockingQueue<>();

    private IRequestProcessor nextProcessor;

    private volatile boolean isFinish = false;

    public SaveProcessor() {    }

    public SaveProcessor(IRequestProcessor nextProcessor) {
        this.nextProcessor = nextProcessor;
    }

    public void shutdown(){
        isFinish = true;
    }

    @Override
    public void run() {
        while (!isFinish){
            try {
                // 阻塞式獲取
                Request request = requests.take();
                // 處理邏輯
                System.out.println("saveProcessor::" + request);

                // 交給下一個責任鏈
                if(null != nextProcessor){
                    nextProcessor.process(request);
                }

                // 線程退出機制
                shutdown();
            } catch (InterruptedException e) {
                // 全部和阻塞相關的方法都會拋出 InterruptedException
                e.printStackTrace();
            }
        }
    }


    @Override
    public void process(Request request) {
        // 加到隊列
        requests.add(request);
    }

四、測試類

public class App {
    private static IRequestProcessor processor;
    
    static {
        SaveProcessor saveProcessor = new SaveProcessor();
        saveProcessor.start();
        processor = new PrevProcessor(saveProcessor);
        ((PrevProcessor) processor).start();
    }
    
    public static void main(String[] args) {
        Request request = new Request("aaaa");
        processor.process(request);
    }

}