java并行的执行两个任务

Posted 2022-01-26 琅琊山二当家

java并行的执行两个任务 ，每个任务又包含多个小任务，下面是一种代码实践的成果

 ExecutorService executor1 = Executors.newSingleThreadExecutor();
        Map<String, Integer> tenantIdNumMap = new HashMap<>();
        FutureTask<Map<String, Integer>> future1 =
                new FutureTask<>(() -> 
                    long time1 = System.currentTimeMillis();
                    ConcurrentHashMap<String, Integer> finalTenantIdNumMap = new ConcurrentHashMap<>();
                    tenantIdList.parallelStream().forEach(tenantId -> 
                        HashMap<String, Object> reqOtherColumn = new HashMap<>();
                        reqOtherColumn.put("beginTime", beginTime);
                        reqOtherColumn.put("endTime", endTime);
                        reqOtherColumn.put("tenantIdList", Arrays.asList(tenantId));
                        //真正的任务在这里执行，这里的返回值类型为String，可以为任意类型
                        List<HashMap<String, String>> tenantNumByIdDistinctInfo = volumeMapper.getTenantNumById(reqOtherColumn);
                        tenantNumByIdDistinctInfo.parallelStream().forEach(map -> 
                            String tenant_id = map.get("tenant_id");
                            Integer idMun = finalTenantIdNumMap.get(tenant_id);
                            if (idMun == null) 
                                finalTenantIdNumMap.put(tenant_id, 1);
                             else 
                                finalTenantIdNumMap.put(tenant_id, ++idMun);
                            
                        );
                    );
                    long time2 = System.currentTimeMillis();
                    logger.warn("------33333----2222222--: " + (time2 - time1));
                    return finalTenantIdNumMap;
                );
        executor1.execute(future1);

        ExecutorService executor2 = Executors.newSingleThreadExecutor();
        Map<String, Integer> tenantIdSubjectNumMap = new HashMap<>();
        FutureTask<Map<String, Integer>> future2 =
                new FutureTask<>(() -> 
                    ConcurrentHashMap<String, Integer> finalTenantIdSubjectNumMap = new ConcurrentHashMap<>();
                    long time1 = System.currentTimeMillis();
                    tenantIdList.parallelStream().forEach(tenantId -> 
                        HashMap<String, Object> reqOtherColumn = new HashMap<>();
                        reqOtherColumn.put("beginTime", beginTime);
                        reqOtherColumn.put("endTime", endTime);
                        reqOtherColumn.put("tenantIdList", Arrays.asList(tenantId));
                        List<HashMap<String, String>> tenantSubjectNumByIdDistinctInfo = volumeMapper.getTenantSubjectNumById(reqOtherColumn);
                        tenantSubjectNumByIdDistinctInfo.parallelStream().forEach(map -> 
                            String tenant_id = map.get("tenant_id");
                            Integer idMun = finalTenantIdSubjectNumMap.get(tenant_id);
                            if (idMun == null) 
                                finalTenantIdSubjectNumMap.put(tenant_id, 1);
                             else 
                                finalTenantIdSubjectNumMap.put(tenant_id, ++idMun);
                            
                        );
                    );
                    long time2 = System.currentTimeMillis();
                    logger.warn("------4444----2222222--: " + (time2 - time1));
                    return finalTenantIdSubjectNumMap;
                );
        executor2.execute(future2);


        try 
            tenantIdNumMap = future1.get(30000, TimeUnit.MILLISECONDS);//取得结果，同时设置超时执行时间为5秒。同样可以用future.get()，不设置执行超时时间取得结果
         catch (InterruptedException e) 
            future1.cancel(true);
         catch (ExecutionException e) 
            future1.cancel(true);
         catch (TimeoutException e) 
            future1.cancel(true);
         finally 
            executor1.shutdown();
        

        try 
            tenantIdSubjectNumMap = future2.get(30000, TimeUnit.MILLISECONDS);//取得结果，同时设置超时执行时间为5秒。同样可以用future.get()，不设置执行超时时间取得结果
         catch (InterruptedException e) 
            future2.cancel(true);
         catch (ExecutionException e) 
            future2.cancel(true);
         catch (TimeoutException e) 
            future2.cancel(true);
         finally 
            executor2.shutdown();