ConcurrentHashMap,LockSupport.parkUntil(),LockSupport.unpark()的使用

ConcurrentHashMap,LockSupport.parkUntil(),LockSupport.unpark(notify)的使用。

业务场景：开通会员服务时，针对用户量较大并发时可以及时的处理业务逻辑，第三方回调后本地依然可以操作当前线程中的业务处理，同时不影响其他线程，只对当前线程做相关的阻塞，关闭操作。

1.定义相关ConcurrentHashMap

@Service
@Transactional(rollbackFor = Exception.class)
public class AccountServiceImpl implements AccountService {

    private static Logger logger = LoggerFactory.getLogger(AccountServiceImpl.class);
    /**
     * 企业信息线程控制器
     */
    private static ConcurrentHashMap<Long, Thread> concurrent_openAccount = new ConcurrentHashMap<>(16);

    @Autowired
    private RedisUtil redisUtil;
    @Autowired
    private SettlementFeign settlementFeign;

2.开通会员接口

将当前线程添加到ConcurrentHashMap中，并通过LockSupport.parkUntil()对当前线程进行阻塞，超过当前时间+1000ms后会自动释放当前线程，通过对ConcurrentHashMap中线程的key的修改来确定第三方有没有产生回调。第三方回调这边的接口时会将ConcurrentHashMap中的对应线程进行修改name ，并通过LockSupport.unpark()对在1000ms内依然加锁的线程进行释放操作。

  @Override
    public ResultMessage openAccount(AccountVo accountVo) {
        logger.info(String.format("==========商户：%s 执行开户操作==========", accountVo.getCorpName()));
        //开户流程状态
        String redisKey = RedisKey_Account_Constants.REDIS_KEY_PREFIX + accountVo.getUserId() + RedisKey_Account_Constants.USER_OPEN_ACCOUNT_STATUS;
        int status = initOpenAccountStatusToRedis(redisKey, accountVo.getUserId());
        if (0 == status) {
            /**1.注册通联会员*/
            ResultMessage createMemberResult = this.createMember(accountVo);
            if (createMemberResult.getCode().intValue() != 200) {
                return createMemberResult;
            }
        }
        if (0 == status || status == OpenAccountStatusEnum.CREATE_MEMBER.getValue()) {
            /**2.设置企业信息*/
            ResultMessage setCompanyInfoResult = this.setCompanyInfo(accountVo);
            if (setCompanyInfoResult.getCode().intValue() != 200 && setCompanyInfoResult.getCode().intValue() != 30023) {
                return setCompanyInfoResult;
            }
            if (setCompanyInfoResult.getCode().intValue() != 30023) {
                Thread thread = Thread.currentThread();
                thread.setName("corpId" + accountVo.getCorpId());
                concurrent_openAccount.put(accountVo.getCorpId(), thread);
                LockSupport.parkUntil("等待结算回调——确认企业状态", System.currentTimeMillis() + TimeUnit.MILLISECONDS.toSeconds(1000));
                logger.info("openAccount——auditCompanyInfo_threadName：" + Thread.currentThread().getName());
                if (Thread.currentThread().getName().equals("corpId" + accountVo.getCorpId())) {
                    //超时则主动发送请求查询企业信息审核状态
                    ResultMessage queryCompanyInfoAuditStatus = this.queryCompanyInfoAuditStatus(accountVo);
                    if (queryCompanyInfoAuditStatus.getCode().intValue() != 200) {
                        return queryCompanyInfoAuditStatus;
                    }
                    Map data = (Map) queryCompanyInfoAuditStatus.getData();
                    Map memberInfo = (Map) data.get("memberInfo");
                    int authType = (int) memberInfo.get("authType");
                    if (1 == authType) {
                        return new ResultMessage(ResponseCode.BUSINESS_DATA_LIMIT.getCode(), "正在认真审核企业信息，请稍后再试！");
                    } else if (3 == authType) {
                        return new ResultMessage(ResponseCode.BUSINESS_DATA_LIMIT.getCode(), "企业信息审核失败，请重新确认！");
                    }
                }
            }
            //更新开户信息
            accountVo.setOpenAccountStatus(OpenAccountStatusEnum.SET_CORP_INFO.getValue());
            saveOpenAccountInfo(accountVo);
            //记录流程状态
            redisUtil.set(redisKey, OpenAccountStatusEnum.SET_CORP_INFO.getValue());
            logger.info("企业信息审核通过");
        }
        logger.info(String.format("==========商户：%s 完成开户操作==========", accountVo.getCorpName()));
        return new ResultMessage(ResponseCode.SUCCESS.getCode(), ResponseCode.SUCCESS.getDesc());
    }

第三方回调接口

这其中主要的是将线程名称修改，释放对应的线程。

 /**
     * 结算回调——确认企业信息
     *
     * @param accountCallback2BizDto
     * @return
     */
    @Override
    public YlwJsonResult callbackAffirmSetCompany(AccountCallback2BizDto accountCallback2BizDto) {
        logger.info("结算中心回调——确认企业信息审核结果.....");
        String jsonString = JSON.toJSONString(accountCallback2BizDto);
        logger.info("结算中心回调——确认企业信息请求数据：" + jsonString);
        if (accountCallback2BizDto.getStatus() == "2") {
            Thread notify = concurrent_openAccount.remove(accountCallback2BizDto.getBizUserId());
            notify.setName("bizUserId" + accountCallback2BizDto.getBizUserId());
            LockSupport.unpark(notify);
        }
        return new YlwJsonResult(10000, ResponseCode.SUCCESS.getDesc());
    }

总结：Lock锁的底层是使用LockSupport.parkUntil(),LockSupport.unpark()对线程进行阻塞，释放等操作。
ConcurrentHashMap 可以在并发情况下线程共享，具有一定的安全性。