java并行的执行两个任务

Posted 2021-08-24 琅琊山二当家

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