使用ExecutorCompletionService,LinkedBlockingQueue 管理线程池处理任务并返回两种类型的结果值

使用ExecutorCompletionService,LinkedBlockingQueue 管理线程池处理任务并返回两种类型的结果值

在我们日常使用线程池的时候，经常会有需要获得线程处理结果的时候。下面通过添加具体的商户用户信息并校验的具体实例说明相应的使用场景。

1.一个方法返回的两种结果类型

import java.io.Serializable;
/**
 * 两个元素的元组，用于在一个方法里返回两种类型的值
 */
public class TwoTuple<A, B> implements Serializable {
    public  A first;
    public  B second;

    public TwoTuple(A a, B b) {
        this.first = a;
        this.second = b;
    }

    public TwoTuple() {
    }

    @Override
    public String toString() {
        return "TwoTuple{" +
                "first=" + first +
                ", second=" + second +
                '}';
    }
}

1.Controller层相应的接收前端参数Map集合。

 /**
     * showdoc
     *
     * @param channelCorpTemp 必选 Map 渠道商企业认证信息
     * @return {"code":200,"message":"操作成功","data":{"..":"..."}}
     * @catalog 企业/provider
     * @title 渠道商企业信息入库
     * @description 渠道商企业信息入库
     * @method POST
     * @url http://ip:port/ylw/qhChannelCorp/api/addCorp
     * @number 99
     */
    @ResponseBody
    @RequestMapping(value = "addChannel", method = RequestMethod.POST)
    public ResultMessage addChannel(@RequestBody Map<Object, Object> channelCorpTemp) {
        try {
            return qhChannelCorpService.addChannel(channelCorpTemp);
        } catch (Exception e) {
            logger.error(e.toString());
            return new ResultMessage(SysCode.FAILUE.getCode(), e.toString());
        }
    }

2.Service层具体的接口及实现类代码。

通过创建一个容量为4的阻塞队列LinkedBlockingDeque，通过ExecutorCompletionService 实例new通过其submit方法创建4个线程，每个线程进行校验相关的资料，如果有一个线程校验不通过则返回商户用户添加失败。

   /**
     * 接口
     */
  ResultMessage addChannel(Map<Object, Object> channelCorpTemp);
   /**
     *  创建的线程池 作为类的属性
     */
       /**
  private int corePoolSize = 3;
    private int maximumPoolSize = 15;
    private long keepAliveTime = 10;
    BlockingQueue<Runnable> workQueue = new LinkedBlockingDeque<>(10);
    private ThreadPoolExecutor threadPool = new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime,
            TimeUnit.SECONDS, workQueue, new ThreadPoolExecutor.AbortPolicy());

   /**
     * 实现层方法
     */
       /**
     * showdoc
     *
     * @param entityMap 必选 Map<String,Object> 渠道商客户信息
     * @return {"code":"200","message":"查询成功","data":[{"...":"..."}]}
     * @catalog 企业/渠道商客户
     * @title 快招-新增渠道商客户
     * @description 新增渠道商客户
     * @method POST
     * @url http://b.youlanw.com/api/qhCorp/channel/addCustomer
     * @remark
     * @number 99
     */
    @Override
    @Transactional(rollbackFor = SqlException.class)
    public ResultMessage addCustomer(Map<Object, Object> entityMap) {
        QhChannelCorpCustomerTemp customerTemp = RedisConvertUtil.mapToObject(entityMap, QhChannelCorpCustomerTemp.class);
        //生成企业code
        String companyCode = UUID.randomUUID().toString().replace("-", "");
        customerTemp.setCorpCode(companyCode);
        //map转Object
        CorpPhotoVo supplementVo = RedisConvertUtil.mapToObject(entityMap, CorpPhotoVo.class);
        Date now = DateConvertUtils.getNow();
        //有公司id修改
        if (customerTemp.getId() != null) { // 修改客户临时认证信息
            boolean updateResult = updateCustomerTempInfoWithFailed(customerTemp, supplementVo);
            redisUtil.removeZset(ChannelCorpRedisKey.CORP + customerTemp.getCorpUserId() + ChannelCorpRedisKey.CHECKFAILED, customerTemp.getId());
            if (!updateResult) {
                return new ResultMessage(SysCode.FAILUE.getCode(), "渠道商客户信息更新失败!");
            }
        } else { // 新增客户临时认证信息
            final Long corpUserId = customerTemp.getCorpUserId();
            final String creditCode = customerTemp.getCreditCode();
            BlockingQueue<Future<TwoTuple<Boolean, String>>> bq = new LinkedBlockingQueue<>(4);
            CompletionService<TwoTuple<Boolean, String>> cs = new ExecutorCompletionService<>(threadPool, bq);
            //判断客户是否已提交过客户认证资料
            cs.submit(() -> isChannelHasCustomer(corpUserId, creditCode));
            //判断客户是否已注册为渠道商(待认证,认证通过)
            cs.submit(() -> isChannelExist(creditCode));
            //判断父账号是否已经拥有此客户(合作状态,申请重新合作状态)
            cs.submit(() -> isParentHasCustomer(corpUserId, creditCode));
            //t判断父账号的其他子账号,是否已经拥有此用户(合作状态,申请重新合作状态)
            cs.submit(() -> isOtherSubAccountHasCustomer(corpUserId, creditCode));
            for (int i = 0; i < 4; i++) {
                TwoTuple<Boolean, String> r = null;
                try {
                    r = cs.take().get();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } catch (ExecutionException e) {
                    e.printStackTrace();
                }
                if (r.first)
                    return new ResultMessage(SysCode.FAILUE.getCode(), r.second);
            }
            insertCustomerTemp(customerTemp, supplementVo);
        }
        // redis 客户状态队列新增
        redisUtil.add(ChannelCorpRedisKey.CORP + customerTemp.getCorpUserId() + ChannelCorpRedisKey.CHECKING, customerTemp.getId(), -now.getTime());
        return new ResultMessage(ChannelCorpTempResponseCode.CHANNEL_CUSTOMER_ADD_SUCCESS.getCode(), "渠道商客户认证信息提交成功!");
    }

3.service实现层中调用的校验方法。

//方法1
 private TwoTuple<Boolean, String> isChannelHasCustomer(Long corpUserId, String creditCode) {
        Map<String, Object> params = new HashMap<>();
        params.put("corpUserId", corpUserId);
        params.put("creditCode", creditCode);
        List<QhChannelCorpCustomerTemp> count = customerTempMapper.findPage(params);
        if (count != null && count.size() > 0)
            return new TwoTuple<>(true, "已提交过客户认证资料,请勿重复提交!");
        return new TwoTuple<>(false, "未提交过客户认证资料");
    }
     /**
     * 通过信用代码,判断是否已经存在渠道商企业
     *
     * @return true:存在
     */
    @ServiceImplParamAnnotation("creditCode")
    public TwoTuple<Boolean, String> isChannelExist(String creditCode) {
        Map<String, Object> params = new HashMap<>();
        params.put("creditCode", creditCode);
        params.put("state", ChannelTempCorpStateEnum.CHECK_SUCCESS);
        List<QhChannelCorpTemp> page = channelTempMapper.findPage(params);
        if (page != null && page.size() > 0) {
            return new TwoTuple<>(true, "success");
        }
        return new TwoTuple<>(false, "failed");
    }
 /**
     * 判断主账号是否拥有此客户
     *
     * @param corpUserId 子账号用户id
     * @return true:有
     */
    private TwoTuple<Boolean, String> isParentHasCustomer(Long corpUserId, String creditCode) {
        Long count = customerMapper.isParentHasCustomer(corpUserId, creditCode);
        if (!(count == null || count.intValue() == 0))
            return new TwoTuple<>(true, "主账号已经拥有当前客户!");
        return new TwoTuple<>(false, "主账号未拥有当前客户!");
    }
     /**
     * 判断父账号的其他子账号,是否已经拥有此用户(合作状态,申请重新合作状态)
     *
     * @return true:有
     */
    private TwoTuple<Boolean, String> isOtherSubAccountHasCustomer(Long corpUserId, String creditCode) {
        // 查询父账号的所有子账号id
        List<Long> otherSubAccountIds = channelUserFeign.getAllOtherSubAccountId(corpUserId);
        if (otherSubAccountIds == null || otherSubAccountIds.size() == 0) return new TwoTuple<>(false, "无其他子账号");
        Map<String, Object> params = new HashMap<>();
        params.put("corpUserIds", otherSubAccountIds);
        params.put("creditCode", creditCode);
        Long count = customerMapper.isOtherSubAccountHasCustomer(params);
        if (!(count == null || count.intValue() == 0))
            return new TwoTuple<>(true, "父账号的其他子账号已经有用此客户!");
        return new TwoTuple<>(false, "其他子账号未绑定此客户");
    }