CompletableFuture parallelStream并行流使用的一个注意事项

╰+哭是因爲堅強的太久メ 2021-12-24 00:03 403阅读 0赞

并行流 就是多线程 使用的威力大，出问题的概率也大，使用时候要谨慎，一旦出问题，

相对来说不好排查错误

这里就说一个遇到的问题: 不要再每个并行流分别计算，这个流处理需要的公共变量，

可以在外部计算好，传进去。

修改前：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0FsYmVydEZseQ_size_16_color_FFFFFF_t_70][]

上图函数具体实现

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0FsYmVydEZseQ_size_16_color_FFFFFF_t_70 1][]

修改后:

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0FsYmVydEZseQ_size_16_color_FFFFFF_t_70 2][]

类似的一个修改

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0FsYmVydEZseQ_size_16_color_FFFFFF_t_70 3][]

最后给出具体获取

List<String> catServiceNames = catService.getCatServiceNames(); 方法 看看为什么导致并发

代码如下

List<String>  appNameList = null ;

public List<String> getCatServiceNames() {
            List<String> res = null;
            res = appNameList != null ? appNameList : getCatServiceNamesPeriod();
            return  res ;
        }

下图代码大致说明

message集合是取自远程io操作获取

serviceList 来自redis 优先级是redis ，涉及到serviceList 和message合并

finalServiceList是最终需要的数据集合，也是返回值，

所以这个操作首先不论是否耗时多少(其实不多)，但是由于多个并行操作，可能对一些状态的判断会出现问题，

导致redis 存储的数据有变化，一些脏数据

有可能线程1 本来拿到redis 应该是拿到线程2

@Scheduled(cron= "0 0/1 * * * ?") //每5分钟执行一次
        public  List<String> getCatServiceNamesPeriod() {
    //        String url = configService.getCatRealUrl()+"/cat/r/state?domain=cat&ip=All&show=true";
            String url = configService.getCatUrl()+"/cat/r/state?domain=cat&ip=All&show=true";
            Document doc;
            List<String> message = new ArrayList<>();
            //最终结果 返回的
            List<String> finalServiceList = new ArrayList<>();
            try {
                doc = Jsoup.connect(url).timeout(500000).get();
                //根据class取值 table.table table-hover table-striped table-condensed 这样取不到值
                //只能按照下面的方式操作 并且本来应该是get(0) 反而要用get(1) 原因有空再看
                Element table = doc.select("table.table").get(1);
                Elements trList = table.select("tr");
                Elements needDeal = new Elements();
                for (Element tr : trList) {
                    if (tr.select("td").size() == 7) {
                        needDeal.add(tr);
                    }
                }
                //needDeal==1 说明没有项目接入cat当前环境 第一个不是的
                if (needDeal.size() == 1) {
    
                } else {
                    //从第二个元素 遍历，第一个不是需要的数据。
                    for (int i = 1; i < needDeal.size(); i++) {
    //                    System.out.println(b.toJson(needDeal.get(i).select("td").first().text()));
                        if(! needDeal.get(i).select("td").last().text().equals("[]"))
                              message.add(needDeal.get(i).select("td").first().text());
                    }
                }
                //从页面得到的list要和redis里面信息合并 页面可能不是最全的 redis里面数据是递增的
                ListOperations listOperations = shardingStringRedisTemplate.opsForList();
                List<String> serviceList = listOperations.range(RadonConstants.redis_service_list, 0, -1);
    
                // redis 缓存 serviceList 优先级高于 message 本地缓存
                if(message == null){//cat服务器挂了
                    if(serviceList==null || serviceList.size()==0){ //redis也没数据了 极端的极端
                        //还是主动放一个值
                        message.add("radon");
                        finalServiceList = message;
                    }else {
                        finalServiceList = serviceList;
                    }
                }else if(message != null){
                    if(serviceList==null || serviceList.size()==0){
                        listOperations.rightPushAll(RadonConstants.redis_service_list, message);
                        finalServiceList = message;//返回值需要 赋值
                    }else {
                        message.stream().forEach(name->{
                            if (!serviceList.contains(name)) {
                                serviceList.add(name);
                            }
                        });
                        //删除 由redisTemplate来控制
                        shardingStringRedisTemplate.delete(RadonConstants.redis_service_list);
                        //重新赋值 保持最新
                        listOperations.rightPushAll(RadonConstants.redis_service_list, serviceList);
                        finalServiceList = serviceList;
                    }
                }
                return  finalServiceList;
            }catch (Exception e){
                return null;
            }finally {
                //赋值全局变量
                if (finalServiceList != null && finalServiceList.size() > 0) {
                    appNameList = finalServiceList;
                }
            }
        }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0FsYmVydEZseQ_size_16_color_FFFFFF_t_70]: /images/20211223/c4f0992a0e334ae1a9112b22fab85b2a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0FsYmVydEZseQ_size_16_color_FFFFFF_t_70 1]: /images/20211223/743c7a839b5f47f7bd3047e7250ee2a5.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0FsYmVydEZseQ_size_16_color_FFFFFF_t_70 2]: /images/20211223/f35b242cafcf46c79fe12136e9766c22.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0FsYmVydEZseQ_size_16_color_FFFFFF_t_70 3]: /images/20211223/e32c5b520af34cf19b1edf5b2c615081.png