1. 響應式程式設計定義

響應式程式設計是一種非同步非阻塞程式設計。

非同步程式設計的目的是為了尋求更高的程式執行效率，通過編寫非同步的非阻塞程式碼可以將當先執行的任務切換到另一個任務，並在非同步處理完成後返回到當前執行緒。

2.Java提供兩種非同步程式設計模型：

2.1 Callback(回撥)：

當一個方法呼叫另一個方法，等待另一個方法執行完成後，將結果通知自己，這就叫callback

public interface CallBack {
    public void stepBCallstepA(String result);
}

@Service
@Slf4j
public class
 
 StepA implements CallBack{
    @Autowired
    private StepB stepB;
    //StepA call stepB to do something
    public void stepACallStepB(){
        long start=System.currentTimeMillis();
        new Thread(new Runnable() {
            @Override
            public void run() {
                stepB.stepBprocess(StepA.
 
this,"stepA call stepB"); //stepA呼叫stepB邏輯
            }
        }).start();
        stepASelfBusiness();
        log.info("time consume:{}",String.valueOf(System.currentTimeMillis()-start));
    }

　　//暴漏給stepB回撥stepA的回撥函式
    @Override
    public void stepBCallstepA(String result) {
        try {
            log.info(
 
"stepB return stepA result:{}"+result);
        }catch (Exception e){

        }
    }
//stepA自己的業務邏輯
    public void stepASelfBusiness(){
        try {
            Thread.sleep(10000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

@Service
@Slf4j
public class StepB {
    public void stepBprocess(CallBack callBack,String input){
        String result=businessProcess(input);
callBack.stepBCallstepA(result);//通過回撥函式呼叫stepA中方法
    }
    public String stepBPurelyprocess(String input){
        return businessProcess(input);
    }
    private String businessProcess(String input){
        //business process
        for(int i=0;i<10000;i++){
            try {
                Thread.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        return "B businessProcess completed";
    }
}

執行結果如下，可見stepA已經執行完，stepB才將計算結果通過回撥函式將結果返回給stepA

2022-12-11 20:05:32.042 c.c.a.m.s.q.r.c.StepA http-nio-8080-exec-2 [INFO] time consume:10004
2022-12-11 20:05:37.650 c.c.a.m.s.q.r.c.StepA Thread-2 [INFO] stepB return stepA result:{}B businessProcess completed

如果不用回撥函式，執行結果如下：

@Service
@Slf4j
public class StepC {
    @Autowired
    private StepB stepB;
    public void stepCCallStepB(){
        long start=System.currentTimeMillis();
        String result=stepB.stepBPurelyprocess("stepC call stepB"); //直接呼叫stepB方法，並將StepB計算結果返回給StepC
        log.info("stepB return stepC result:{}"+result);
        stepCSelfBusiness();
        log.info("time consume:{}",String.valueOf(System.currentTimeMillis()-start));
    }

    public void stepCSelfBusiness(){
        try {
            Thread.sleep(10000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

執行結果：

2022-12-11 20:04:59.487 c.c.a.m.s.q.r.c.StepC http-nio-8080-exec-1 [INFO] stepB return stepC result:{}B businessProcess completed
2022-12-11 20:05:09.499 c.c.a.m.s.q.r.c.StepC http-nio-8080-exec-1 [INFO] time consume:25666

2.2 Future：

非同步方法立即返回Future<T>。非同步過程計算一個T值，Future物件封裝裝了對T值的訪問邏輯。T值不是立即可用的，執行緒可以輪詢該物件，直到該值可用為止。例如，ExecutorService正在執行的Callable<T>任務使用Future物件。

2.2.1 callable & runable

 建立執行緒方式： thread runable callable。 callable較runable有返回結果。

 多執行緒使用場景：當任務不是很耗時，多執行緒作用不大，反而效能上比不上單執行緒，而在多執行緒中，推薦CompleteableFuture去建立任務開啟執行緒操作，效能比Callable與FutureTask組合好很多

單執行緒vsFutreTask Callable

@Service
@Slf4j
public class CallableTest {
    @Autowired
    private MongoTemplate mongoTemplate;
    public void implementCallable(){
        try {
            long start=System.currentTimeMillis();
            FutureTask<List<Map>> table1Task = new FutureTask<>(new MongoCallble1());
            FutureTask<List<Map>> table3Task = new FutureTask<>(new MongoCallble2());
            FutureTask<List<Map>> table2Task = new FutureTask<>(new MongoCallble3());
            FutureTask<List<Map>> ranktable2Task = new FutureTask<>(new MongoCallble4());
            new Thread(table1Task).start();
            new Thread(table3Task).start();
            new Thread(table2Task).start();
            new Thread(ranktable2Task).start();
            List<Integer> date2 = Arrays.asList(20221101);
            Query query2 = new Query();
            query2.addCriteria(Criteria.where("businessDate").in(date2));
            List<Map> result2=mongoTemplate.find(query2, Map.class, "table1-v2");
            List<Map> table1List = table1Task.get();
            List<Map> table3List = table3Task.get();
            log.info("time consume:{}",String.valueOf(System.currentTimeMillis()-start));
        }catch (Exception e){
            log.error(e.toString());
        }
    }
    public void noCallable(){
        long start=System.currentTimeMillis();
        List<Integer> date = Arrays.asList(20221102);
        Query query = new Query();
        query.addCriteria(Criteria.where("businessDate").in(date));
        List<Map> result=mongoTemplate.find(query, Map.class, "table1-v2");
        List<Integer> businessTag1 = Arrays.asList(20221102);
        Query query1 = new Query();
        query1.addCriteria(Criteria.where("businessDate").in(businessTag1));
        List<Map> result1=mongoTemplate.find(query1, Map.class, "table3-v2");
        List<Integer> date2 = Arrays.asList(20221102);
        Query query2 = new Query();
        query2.addCriteria(Criteria.where("businessDate").in(date2));
        List<Map> result2=mongoTemplate.find(query2, Map.class, "table1-v2");

        List<Integer> date3 = Arrays.asList(20221102);
        Query query3 = new Query();
        query3.addCriteria(Criteria.where("businessDate").in(date));
        List<Map> result3=mongoTemplate.find(query3, Map.class, "table2-v2");

        List<Integer> date4 = Arrays.asList(20221102);
        Query query4 = new Query();
        query4.addCriteria(Criteria.where("businessDate").in(date));
        List<Map> result4=mongoTemplate.find(query4, Map.class, "ranktable2-v2");
        log.info("time consume:{}",String.valueOf(System.currentTimeMillis()-start));
    }


    class MongoCallble1 implements Callable<List<Map>> {

        @Override
        public List<Map> call() throws Exception {
            List<Integer> date = Arrays.asList(20221102);
            Query query = new Query();
            query.addCriteria(Criteria.where("businessDate").in(date));
            List<Map> result=mongoTemplate.find(query, Map.class, "table1-v2");
            return result;
        }
    }

    class MongoCallble2 implements Callable<List<Map>> {

        @Override
        public List<Map> call() throws Exception {
            List<Integer> date = Arrays.asList(20221102);
            Query query = new Query();
            query.addCriteria(Criteria.where("businessDate").in(date));
            List<Map> result=mongoTemplate.find(query, Map.class, "table3-v2");
            return result;
        }
    }
    class MongoCallble3 implements Callable<List<Map>> {

        @Override
        public List<Map> call() throws Exception {
            List<Integer> date = Arrays.asList(20221102);
            Query query = new Query();
            query.addCriteria(Criteria.where("businessDate").in(date));
            List<Map> result=mongoTemplate.find(query, Map.class, "table2-v2");
            return result;
        }
    }
    class MongoCallble4 implements Callable<List<Map>> {

        @Override
        public List<Map> call() throws Exception {
            List<Integer> date = Arrays.asList(20221102);
            Query query = new Query();
            query.addCriteria(Criteria.where("businessDate").in(date));
            List<Map> result=mongoTemplate.find(query, Map.class, "ranktable2-v2");
            return result;
        }
    }
}

通過執行程式碼log可以發現，在不執行query3 query4時，由於任務不怎麼耗時，單執行緒效率比callable多執行緒效率還高；但如果加入query3 query4甚至更多耗時任務時，多執行緒效率是更高的 

 2.2.2 Futuretask Callable vs CompletableFuture

   public void completableService(){
        try {
            long start = System.currentTimeMillis();
            CompletableFuture completableFuture1 = CompletableFuture.supplyAsync(() -> {
                List<Integer> date = Arrays.asList(20221102);
                Query query = new Query();
                query.addCriteria(Criteria.where("businessDate").in(date));
                List<Map> result = mongoTemplate.find(query, Map.class, "table1-v2");
                return result;
            });
            CompletableFuture completableFuture2 = CompletableFuture.supplyAsync(() -> {
                List<Integer> date = Arrays.asList(20221102);
                Query query = new Query();
                query.addCriteria(Criteria.where("businessDate").in(date));
                List<Map> result = mongoTemplate.find(query, Map.class, "table2-v2");
                return result;
            });
            CompletableFuture completableFuture3 = CompletableFuture.supplyAsync(() -> {
                List<Integer> date = Arrays.asList(20221102);
                Query query = new Query();
                query.addCriteria(Criteria.where("businessDate").in(date));
                List<Map> result = mongoTemplate.find(query, Map.class, "table3-v2");
                return result;
            });
            CompletableFuture completableFuture4 = CompletableFuture.supplyAsync(() -> {
                List<Integer> date = Arrays.asList(20221102);
                Query query = new Query();
                query.addCriteria(Criteria.where("businessDate").in(date));
                List<Map> result = mongoTemplate.find(query, Map.class, "table4-v2");
                return result;
            });
            List<Map> list1= (List<Map>) completableFuture1.get();
            completableFuture2.get();
            completableFuture3.get();
            completableFuture4.get();
            log.info("time consume:{}",String.valueOf(System.currentTimeMillis()-start));
        }catch (Exception e){

        }

    }

分析CompletableFuture比Callable與FutureTask組合好的原因：
檢視CompletableFuture.supplyAsync（）的原始碼，看到有個asyncPool->Executor ，用到執行緒池。

public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier) {
    return asyncSupplyStage(asyncPool, supplier);
}

static <U> CompletableFuture<U> asyncSupplyStage(Executor e,
                                                 Supplier<U> f) {
    if (f == null) throw new NullPointerException();
    CompletableFuture<U> d = new CompletableFuture<U>();
    e.execute(new AsyncSupply<U>(d, f));
    return d;
}

 2.3 publish subscrbe方式

參考文獻：

callback參考：https://www.jianshu.com/p/7ee7edac1a13  https://blog.csdn.net/u012393791/article/details/52777029

callable參考：https://blog.csdn.net/qq_22744093/article/details/111468896