[实践总结] 典型的串行任务局部并行化处理案例:多任务并发获取航班信息

假设你有一个APP，主要用于查询航班信息，你的APP是没有这些实时数据的，当用户发起查询请求时，你需要到各大航空公司的接口获取信息，最后统一整理加工返回到APP客户端。当然JDK自带了很多高级工具，比如CountDownLatch和CyclicBarrier等都可以完成类似的功能，但是仅就我们目前所学的知识，使用join方法即可完成下面的功能。

该例子是典型的串行任务局部并行化处理，用户在APP客户端输入出发地“北京”和目的地“上海”，服务器接收到这个请求之后，先来验证用户的信息，然后到各大航空公司的接口查询信息，最后经过整理加工返回给客户端，每一个航空公司的接口不会都一样，获取的数据格式也不一样，查询的速度也存在着差异，如果再跟航空公司进行串行化交互（逐个地查询），很明显客户端需要等待很长的时间，这样的话，用户体验就会非常差。如果我们将每一个航空公司的查询都交给一个线程去工作，然后在它们结束工作之后统一对数据进行整理，这样就可以极大地节约时间，从而提高用户体验效果。

// 面相接口编程,定义一个查询接口FightQuery
public interface FightQuery {
    List<String> getRes();
}

----------------------------------------------------------------------------------------
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;

// 查询Fight的task，其实就是一个线程的子类，主要用于到各大航空公司获取数据
public class FightQueryTask extends Thread implements FightQuery {
    private final String origin;
    private final String destination;
    private final List<String> flightMsgs = new ArrayList<>();

    /**
     *
     * @param airline 航空公司
     * @param origin 始发站
     * @param destination 目的地
     */
    public FightQueryTask(String airline, String origin, String destination) {
        super("[" + airline + "]"); // 学到了这里居然是给线程起名字
        this.origin = origin;
        this.destination = destination;
    }

    @Override
    public void run() {
        System.out.println(Thread.currentThread().getName() + "Query from " + origin + " to " + destination);

        // 模拟业务逻辑处理
        int randomVal = ThreadLocalRandom.current().nextInt(10);
        shortSleep(randomVal);
        flightMsgs.add(getName() + "-" + randomVal);

        System.out.println(getName() + "query done");
    }

    @Override
    public List<String> getRes() {
        return this.flightMsgs;
    }

    private static void shortSleep(long time) {
        try {
            TimeUnit.SECONDS.sleep(time);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
----------------------------------------------------------------------------------------
public class FightQueryMain {
    public static void main(String[] args) {
        // 实现一下从SH（上海）到北京（BJ）的航班查询
        List<String> results = flightSearch("SH", "BJ");

        System.out.println("===========result===========");
        results.forEach(System.out::println);
    }

    private static List<String> flightSearch(String original, String dest) {
        //①合作的各大航空公司
        List<String> fightCompanies = Arrays.asList("CSA", "CEA", "HNA");
        //②创建查询航班信息的线程列表
        List<FightQueryTask> queryTasks = fightCompanies.stream().map(fightCompany -> new FightQueryTask(fightCompany, original, dest)).toList();
        //③分别启动这几个线程
        queryTasks.forEach(Thread::start);

        //④分别调用每一个线程的join方法，阻塞当前线程
        queryTasks.forEach(t -> {
            try {
                t.join();
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        });

        //⑤当前线程会阻塞住,直到获取每一个查询线程的结果
        return queryTasks.stream()
                .map(FightQuery::getRes)
                .flatMap(Collection::stream)
                .toList();
    }
}


[CSA]Query from SH to BJ
[HNA]Query from SH to BJ
[CEA]Query from SH to BJ
[HNA]query done
[CEA]query done
[CSA]query done
===========result===========
[CSA]-5
[CEA]-3
[HNA]-0

-----------------------------------------------------------------------------读书笔记摘自书名：Java高并发编程详解：多线程与架构设计 作者：汪文君