一、parseq简介

ParSeq 是一个框架，它使用 Java 编写异步代码变得更加容易。它的特点有异步操作的并行化、非阻塞计算的序列化执行、通过任务组合重用代码、常见的错误传播和恢复、执行跟踪和可视化、异步批量的批处理、具有重试策略的任务。

可用版本：https://mvnrepository.com/artifact/com.linkedin.parseq/parseq。

change log：https://github.com/linkedin/parseq/blob/master/CHANGELOG.md

二、关键概念

三、API简介

四、使用方法

4.1 快速开始

第1-9行，声明了一个Engine，第10-21行，将一个任务进行异步执行。当然，这样做毫无意义，只是展示了API的基本用法。

final int numCores = Runtime.getRuntime().availableProcessors();
final ExecutorService taskScheduler = Executors.newFixedThreadPool(numCores + 1);
final ScheduledExecutorService timerScheduler = Executors.newSingleThreadScheduledExecutor();

final Engine engine = new EngineBuilder()
    .setTaskExecutor(taskScheduler)
    .setTimerScheduler(timerScheduler)
    .build();

public List<Blog> findBlogList(String userId) {
	Task<List<Blog>> blogListFuture = Task.callable("findLogList", 
                                () -> blogMapper.findBlogList(userId));
    engine.run(blogListFuture);
    
    try {
        blogListFuture.await();
    } catch (InterruptedException e) {
        // 处理异常
    }
    
    return blogListFuture.get();
}

4.2 任务并联

下面代码是一个查询博客列表的例子，查询博客列表和查询博客数目可以并发执行。

第3行查询博客列表，第7行查询共有多少条博客，第11行将两个任务进行并联执行，并将结果进行封装。

public PageResult queryBlogPage(PageQueryUtil pageUtil) {

    Task<List<Blog>> blogListFuture = Task.callable("findLogList", () ->
        blogMapper.findBlogList(pageUtil)
    );

    Task<Integer> blogTotalFuture = Task.callable("getTotalBlogs", () ->
            blogMapper.getTotalBlogs(pageUtil)
    );

    Task<PageResult> result = Task.par(blogListFuture,blogTotalFuture).map("mergeResult", (blogList,blogTotal) ->
        new PageResult(blogList, blogTotal, pageUtil.getLimit(), pageUtil.getPage())
    );

    engine.run(result);

    try {
        result.await();
    } catch (InterruptedException e) {
        // 处理异常
    }

    return result.get();
}

4.3 任务串联

假如后一个任务对前一个任务的返回结果有依赖，代码又该如何写呢。

下面代码中，步骤[2]查询博客的分类需要依赖步骤[1]查出来的博客列表，步骤[2]利用flatMap方法取出步骤[1]的返回结果，再创建一个新的Task来执行步骤[2]，达到串行的目的。

步骤[2]和步骤[3]又可以并发执行，使用Task.par可以实现任务的并发编排。

public PageResult getBlogsForIndexPage(int page) {
    Map params = new HashMap<String, Object>();
    params.put("page", page);
    params.put("limit", 10);
    params.put("blogStatus", 1);

    PageQueryUtil pageUtil = new PageQueryUtil(params);

    // [1] 查询博客列表
    Task<List<Blog>> blogListFuture = Task.callable("findLogList", () ->
            blogMapper.findBlogList(pageUtil)
    );

    // [2] 根据博客列表查询博客的分类，依赖步骤[1]
    Task<List<BlogListVO>> blogListVOSListFuture =
            blogListFuture.flatMap("flatMap", blogList ->
                    Task.callable("queryBlogCategory", () -> queryBlogCategory(blogList))
            );

    // [3] 查询博客总条数
    Task<Integer> blogTotalFuture = Task.callable("getTotalBlogs", () ->
            blogMapper.getTotalBlogs(pageUtil)
    );

    // [2][3]并发执行
    Task<PageResult> result = Task.par(blogListVOSListFuture, blogTotalFuture)
            .map("mergeResult", (blogListVOSList, blogTotal) -> new PageResult(blogListVOSList, blogTotal, pageUtil.getLimit(), pageUtil.getPage()));

    engine.run(result);

    try {
        result.await();
    } catch (Exception e) {
        e.printStackTrace();
    }

    return result.get();
}

4.4 重试机制

带重试策略的异步任务，RetryPolicy.attempts(3, 500) 表示重试3次，每次间隔500ms。

 Task<List<Blog>> blogListFuture =
                Task.withRetryPolicy("retry",RetryPolicy.attempts(3, 500),
                () -> Task.callable("findBlogList",() -> blogMapper.findBlogList(pageUtil))
        );

4.5 异常恢复

使用默认值：使用recover方法，可以在getTotalBlogs的方法发生异常时，使用默认值兜底。

Task<Integer> blogTotalFuture = Task.callable("getTotalBlogs", () ->
                blogMapper.getTotalBlogs(pageUtil)
        ).recover("dealException",e -> 0);

降级到具体的降级方法：使用recoverWith方法，可以在getTotalBlogs的方法发生异常时，使用降级方法兜底。

 Task<Integer> blogTotalFuture = Task.callable("getTotalBlogs", () ->
                blogMapper.getTotalBlogs(pageUtil)
        ).recoverWith("dealException",
      e -> Task.callable("degradationMethod",() -> degradationMethod(pageUtil)));