使用CountDownLatch怎么实现一个并发框架

本篇文章给大家分享的是有关使用CountDownLatch怎么实现一个并发框架，小编觉得挺实用的，因此分享给大家学习，希望大家阅读完这篇文章后可以有所收获，话不多说，跟着小编一起来看看吧。

目录结构

package com.**.**.base.support.executor;import lombok.NoArgsConstructor;import org.apache.commons.logging.Log;import org.apache.commons.logging.LogFactory;import org.springframework.util.Assert;import java.util.ArrayList;import java.util.List;import java.util.concurrent.CountDownLatch;import java.util.concurrent.Executor;import java.util.concurrent.Executors;import java.util.stream.Collectors;/** * @Author lbm * @Date 2019/10/9 10:32 上午 * @Description YJ执行器 **/@NoArgsConstructorpublic class YjExecutor {private static final Log log = LogFactory.getLog(YjExecutor.class);private static final int FIRST_RETURN = 1;private CountDownLatch latch;private List executeUnits = new ArrayList<>();private Executor pool = Executors.newFixedThreadPool(10);public static YjExecutor build() {return new YjExecutor();    }public YjExecutor add(ExecuteUnit paramUnit) {        paramCheck(paramUnit);executeUnits.add(paramUnit);return this;    }public YjExecutor addAll(List paramUnits) {        paramCheck(paramUnits);this.executeUnits.addAll(paramUnits);return this;    }/**     * 第一个执行完返回     *     * @return 执行结果     */    public ReturnT firstPriorityReturn() {this.latch = new CountDownLatch(FIRST_RETURN);        List results = this.submit();return results.stream()                .filter(res -> res != null && res.getRes() != null)                .findFirst()                .orElse(null);    }/**     * 所有执行完才返回     *     * @return 执行结果     */    public List allReturn() {this.latch = new CountDownLatch(executeUnits.size());return submit();    }/**     * 参数校验     *     * @param paramUnits 参数列表     */    private void paramCheck(List paramUnits) {        Assert.notNull(paramUnits, "Argument can not be null !!");for (ExecuteUnit paramUnit : paramUnits) {            paramCheck(paramUnit);        }    }/**     * 参数校验     *     * @param paramUnit 单个参数     */    private void paramCheck(ExecuteUnit paramUnit) {        Assert.notNull(paramUnit, "Argument can not be null !!");        Assert.notNull(paramUnit.getFunc(), "Argument function can not be null !!");    }/**     * 执行     */    private void execute() {try {for (ExecuteUnit executeUnit : executeUnits) {pool.execute(executeUnit.getYjRunner());            }latch.await();        } catch (Exception e) {log.error("execute error {}", e);        }    }/**     * 提交     *     * @return 返回执行结果     */    private List submit() {        buildExecuteUnit();        execute();return executeUnits.stream()                .map(ExecuteUnit::getResult)                .collect(Collectors.toList());    }/**     * 创建执行单元     */    private void buildExecuteUnit() {for (ExecuteUnit executeUnit : executeUnits) {            executeUnit.setYjRunner(executeUnit.yjRunnerBuild(latch));        }    }}

package com.**.**.base.support.executor;import lombok.Builder;import lombok.Data;/** * @Author lbm * @Date 2019/10/9 10:10 上午 * @Description 执行结果 **/@Data@Builderpublic class ReturnT {private ExecuteEnum resCode;private R res;}

package com.**.**.base.support.executor;import lombok.AllArgsConstructor;import lombok.Builder;import lombok.Data;import org.slf4j.Logger;import org.slf4j.LoggerFactory;import java.util.concurrent.CountDownLatch;import java.util.function.Function;/** * @Author lbm * @Date 2019/10/9 7:33 下午 * @Description 执行单元 **/@Data@Builderpublic class ExecuteUnit {private static Logger log = LoggerFactory.getLogger(ExecuteUnit.class);private Function func;private T param;private ReturnT result;private YjRunner yjRunner;public YjRunner yjRunnerBuild(CountDownLatch latch) {return new YjRunner(latch, this);    }@AllArgsConstructor    public static class YjRunner implements Runnable {private CountDownLatch latch;private ExecuteUnit executeUnit;@Override        public void run() {            ReturnT returnT = ReturnT.builder()                    .resCode(ExecuteEnum.SUCCESS)                    .build();try {                Object res = executeUnit.getFunc().apply(executeUnit.getParam());                returnT.setRes(res);            } catch (Exception e) {log.error("param: {} execute error: {}", executeUnit.getParam(), e);                returnT.setResCode(ExecuteEnum.FAIL);            } finally {executeUnit.setResult(returnT);this.latch.countDown();            }        }    }}

package com.**.**.base.support.executor;/** * @Author lbm * @Date 2019/10/9 2:54 下午 * @Description 执行结果 **/public enum ExecuteEnum {/**     * 执行结果     */    SUCCESS,FAIL;}

package com.**.**.**.serviceTest;import com.alibaba.fastjson.JSONObject;import com.**.**.base.support.executor.ExecuteUnit;import com.**.**.base.support.executor.ReturnT;import com.**.**.base.support.executor.YjExecutor;import com.**.**.api.domain.SysUser;import com.**.**.server.BaseTest;import org.junit.Test;import org.slf4j.Logger;import org.slf4j.LoggerFactory;import java.util.ArrayList;import java.util.List;import java.util.function.Function;/** * @Author lbm * @Date 2019/10/9 10:45 上午 * @Description yj执行器测试 **/public class ExecutorTest extends BaseTest {private Logger logger = LoggerFactory.getLogger(this.getClass());private Function func1 = (a) -> {        SysUser user = new SysUser();        user.setId((Long) a);return user;    };private Function func2 = (a) -> {try {            Thread.sleep(5000L);        } catch (Exception e) {logger.error("error {}", e.getMessage());        }return (a * a);    };private Function func3 = (a) -> {try {            Thread.sleep(6000L);        } catch (Exception e) {            e.printStackTrace();logger.error("error {}", e.getMessage());        }return a - a;    };@Test    public void allReturnTest() {        List executeUnits = new ArrayList<>();        ExecuteUnit unit1 = ExecuteUnit.builder()                .func(func2)                .param(20L)                .build();        executeUnits.add(unit1);        executeUnits.add(ExecuteUnit.builder()                .func(func1)                .param(1L)                .build());        executeUnits.add(ExecuteUnit.builder()                .func(func3)                .param(22L)                .build());        List results = YjExecutor.build()                .addAll(executeUnits)                .allReturn();for (ReturnT res : results) {logger.info(JSONObject.toJSONString(res));        }    }@Test    public void firstPriorityReturnTest() {        List executeUnits = new ArrayList<>();        executeUnits.add(ExecuteUnit.builder()                .func(func1)                .param(1L)                .build());        executeUnits.add(ExecuteUnit.builder()                .func(func3)                .param(22L)                .build());        ReturnT result = YjExecutor.build()                .add(ExecuteUnit.builder()                        .func(func2)                        .param(20L)                        .build())                .addAll(executeUnits)                .firstPriorityReturn();logger.info(JSONObject.toJSONString(result));    }}

以上就是使用CountDownLatch怎么实现一个并发框架，小编相信有部分知识点可能是我们日常工作会见到或用到的。希望你能通过这篇文章学到更多知识。更多详情敬请关注行业资讯频道。