java并行的执行两个任务

蔚落 2022-08-31 07:16 190阅读 0赞

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();
            }