实习踩坑之路：parallelStream并发流+快速失败导致线上CPU300%的血泪史

场景描述

场景我大概描述下，我是要定时10S拉取一次可以执行的任务(任务是抽象概念理解就行)，但是不同公司之间我希望他们可以并发跑，所以我想到的是先查出来需要执行任务的公司ID ---->companyList
但是我一想，10S就就执行一次这个查可执行公司的MySQL是不是太浪费了，而且这张表的数据会越来越大，那么这个SQL说不定会成为运行效率的瓶颈，所以我想到了一种方案：大家来看看，看看大家能不能发现这个方案的bug

初始方案

诶，我每次新增任务的时候，我就把这个公司id记录到redis的List数据结构里面，能否达到我的要求呢？

 @Overridepublic void 新增任务(List<xxxx> xxxxxx) {//插入任务xxxxRepository.insertBatch(thirdAutoSendMsgDOS);//获取公司idLong id = xxxx.getCompanyId();//记录公司ID  todoString companyIdList = RedisKeyConstant.genSendMsgExecutableCompany();List<String> redisList = redisService.getRedisList(companyIdList);if (redisList.isEmpty()){redisService.setRedisList(companyIdList, Arrays.asList(id.toString()));}else {if (!redisList.contains(id.toString())){redisList.add(id.toString());redisService.setRedisList(companyIdList, redisList);}}}

再来看看执行任务的方法

 @Overridepublic void autoSendMsg() {String companyIdList = RedisKeyConstant.genSendMsgExecutableCompany();List<String> redisList = redisService.getRedisList(companyIdList);if (!CollectionUtils.isNotEmpty(redisList)){log.info("并无公司需要执行任务");return;}//这是一个坑redisList.parallelStream().forEach(companyId -> {//拉取该公司最近创建的100条任务List<xxxxxDO> xxxxDOS = xxxxRepository.queryLatelyTask(xxxxxEnum.NOT_SEND.getCode(),Long.valueOf(companyId));if (xxxxxDOS.isEmpty()){//无数据，说明无需发送,将该公司剔除redisList.remove(companyId);redisService.setRedisList(companyIdList,redisList);return;}xxxxxDOS.forEach(thirdAutoSendMsgDO -> {ThreadExecutor.executorTask(() -> {//加锁//。。。。。省略具体业务//这里是坑你能看出来么if (conversationEntity.getStatus().equals(CommonStatusEnum.TRUE.getCode())) {throw new CommonException(ByWechatBotErrorCode.MSG_SEND_FAIL, "该好友/群关系已删除");}String key = RedisKeyConstant.genThirdAutoSendMsg(conversationEntity.getUserId());//这个地方也是一个坑lock = redisService.lockWithExpireTimeAndRetryTimes(key, 90, 3);if (lock){//具体业务逻辑}else {log.info("[三方自动群发消息失败]{}微信正在其他任务中发送",conversationEntity.getWxId());}}, "三方自动发送消息任务");});});}

为什么有坑

1.并发修改 + 快速失败机制
这个机制大家应该都听过，就是在集合里面删除了本集合里面的元素

//这是一个坑redisList.parallelStream().forEach(companyId -> {//拉取该公司最近创建的100条任务List<xxxxxDO> xxxxDOS = xxxxRepository.queryLatelyTask(xxxxxEnum.NOT_SEND.getCode(),Long.valueOf(companyId));if (xxxxxDOS.isEmpty()){//无数据，说明无需发送,将该公司剔除redisList.remove(companyId);redisService.setRedisList(companyIdList,redisList);return;}

可以看到我这个地方是parallelStream来遍历了，导致其实我下面的remove操作是在另一个线程，也就是说方法的主调用线程的状态其实是成功了，所以问题就来了，如果说我的定时时间比较长，那报错我也会发现，修复一下，但是我定时是10S跑一次，问题就大了，10S跑一次，我remove又报错，redisList就一直会有值，一值开线程来遍历，那么我10s就开几个线程，10s开几个线程，所以我CPU爆了，业务线程都被他抢占了，这个是最致命的。

2.逻辑上的错误

 //这里是坑你能看出来么if (conversationEntity.getStatus().equals(CommonStatusEnum.TRUE.getCode())) {throw new CommonException(ByWechatBotErrorCode.MSG_SEND_FAIL, "该好友/群关系已删除");}

这是第二个错误，原因就是我抛出异常了，抛异常有问题么？方便我们捕获嘛，但是我这个地方的任务是有状态的，也就是未执行，执行中，成功，失败
我这个地方之前1是没有把任务状态改成执行中，抛错的时候也没把状态改为失败，所以问题就出现了，效果就是我这条任务一直在跑，每次都抛错，那这堆栈就多了

加锁重试有问题

//这个地方也是一个坑lock = redisService.lockWithExpireTimeAndRetryTimes(key, 90, 3);

这个又是什么问题呢？
你想啊我是Sass端，肯定不止一个人拿到任务，假如我一个人拿到了任务，那另外的人应该等待，诶没错啊，但是你看后面的重试次数是3，也就导致了我一个任务被重复执行了3次，我这个地方是粗心了，不应该重试的

4.隐藏bug

最大的原因还是为了维护这个companyIdList导致的线上问题，在循环里面移除自己的元素了，这是比较明显的错误，还有一个隐藏的：我插入的时候也向redis里面插入了，那会不会出现这种情况呢？

新任务companyId = 1 进来

我另一个线程：发现companyId没有任务了，我要删除了

这两个线程的执行顺序就有问题了：如果先插入在删除，这companyId = 1的任务不就没了？

当时师兄看JMX和Jstack发现CPU爆满没有线程大概就确定可能出现了死锁/死循环，血的教训啊

解决方案

1.使用加锁的机制，对redisService.setRedisList();加锁，谁要写入或者删除都要加速，成功了才能修改，失败不能修改。但是这不能解决快速失败
2.使用迭代器来进行remove操作，并且remove之前需要获取到锁
3.不抛出异常设置标识位，修改数据库该条任务状态直接返回

但是这种加锁的方式还是性能不太高，而且较为繁琐，结合我们项目的情况(分布式的两台机器),我能不能把任务分发到不同机器上呢，这样可以分摊压力，而且我是一增一删不太适合这个场景，因此我就采用了下面的方案：

1.分为两个定时，一个定时负责查询可执行的公司ID，然后根据一个小的分片算法，把companyIdList放到两个redisList里面（两个redisList其实就是 key：xxx_xxx_分片1: value, key : xxx_xxx_分片2: value）

2.拿到机器分片下表，直接取出来这个key对应的reidsList，然后再去开多线程执行我们的业务逻辑

好了，下班，有不同意见的或者哪里有不对的地方感谢指出，人生不易，熊猫叹气