JAVA多线程执行，等待返回结果，再执行

1.实现callable接口

1）配置线程池

package com.neusoft.demo.server.config;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.task.TaskExecutor;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.concurrent.ThreadPoolExecutor;

/**
 * 线程池配置、启用异步
 * 
 * @author
 *
 */
@EnableAsync(proxyTargetClass = true)
@Configuration
public class AsycTaskExecutorConfig {

	@Bean
	public TaskExecutor taskExecutor() {
		ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
        // 核心线程数
		taskExecutor.setCorePoolSize(50);
		//最大线程数
		taskExecutor.setMaxPoolSize(100);
		//最大队列数
        taskExecutor.setQueueCapacity(1000);
        // 线程的空闲时间
        taskExecutor.setKeepAliveSeconds(100);
        //线程前缀
        //taskExecutor.setThreadNamePrefix("asyncTaskExecutor-");
        //拒绝策略交给主线程执行
        taskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
		return taskExecutor;
	}
}

2）编写线程实现callable接口

package com.neusoft.demo.server.thread;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.neusoft.demo.server.model.TrafficFlow;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.query.RangeQueryBuilder;
import org.elasticsearch.index.query.TermQueryBuilder;
import org.elasticsearch.index.query.TermsQueryBuilder;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.BucketOrder;
import org.elasticsearch.search.aggregations.bucket.histogram.DateHistogramAggregationBuilder;
import org.elasticsearch.search.aggregations.bucket.histogram.DateHistogramInterval;
import org.elasticsearch.search.aggregations.bucket.histogram.Histogram;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Scope;

import java.io.IOException;
import java.util.List;
import java.util.concurrent.Callable;

/**
 * @author dume
 * @create 2021-10-26 9:44
 **/
@Scope("prototype")
@Configuration
public class ThreadTaskProcess implements Callable {
    private static final Logger log = LoggerFactory.getLogger("adminLogger");
    @Qualifier("MyHighLevelClient")
    @Autowired
    private RestHighLevelClient rhlClient;
    @Value("${elasticsearch.motor-vehicle-node-name}")
    private String motorVehicleNode;

    private int searchnum;
    private String EndTime;
    private String BeginTime;
    private String Precision;
    private String DeviceCityCode;
    private String DeviceID;
    private String PlateColor;
    private String Direction;


    public ThreadTaskProcess getInstance(String EndTime, String BeginTime, String Precision, String DeviceCityCode, String DeviceID, String PlateColor, String Direction,int searchnum) {
        this.EndTime = EndTime;
        this.BeginTime = BeginTime;
        this.Precision = Precision;
        this.DeviceCityCode = DeviceCityCode;
        this.DeviceID = DeviceID;
        this.PlateColor = PlateColor;
        this.Direction = Direction;
        this.searchnum = searchnum;
        return this;
    }
    @Override
    public Object call() throws Exception {
        //es请求
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
        BoolQueryBuilder bqb = QueryBuilders.boolQuery();
        if (BeginTime != null && EndTime != null) {
            RangeQueryBuilder rqb = QueryBuilders.rangeQuery("PassTime");
            rqb.gte(BeginTime);
            rqb.lt(EndTime);
            rqb.format("yyyyMMddHHmmss");
            bqb.must(rqb);
        }
        if(StringUtils.isNotBlank(DeviceCityCode)){
            if(DeviceCityCode.contains(",")){
                String[] strings = DeviceCityCode.split(",");
                TermsQueryBuilder tqb = QueryBuilders.termsQuery("DeviceCityCode",strings);
                bqb.must(tqb);
            }else{
                TermQueryBuilder tqb = QueryBuilders.termQuery("DeviceCityCode",DeviceCityCode);
                bqb.must(tqb);
            }

        }
        if(StringUtils.isNotBlank(DeviceID)){
            TermQueryBuilder tqb = QueryBuilders.termQuery("DeviceID",DeviceID);
            bqb.must(tqb);
        }
        if(StringUtils.isNotBlank(PlateColor)){

            if(PlateColor.length()==1){
                TermQueryBuilder tqb = QueryBuilders.termQuery("PlateColor",PlateColor);
                bqb.must(tqb);
            }else{
                TermQueryBuilder tqb = QueryBuilders.termQuery("PlateColor","6");
                bqb.mustNot(tqb);
            }
        }
        if(StringUtils.isNotBlank(Direction)){
            TermQueryBuilder tqb = QueryBuilders.termQuery("Direction",Direction);
            bqb.must(tqb);
        }

        sourceBuilder.query(bqb);
        sourceBuilder.size(0);
        sourceBuilder.timeout(new TimeValue(600000));
        DateHistogramAggregationBuilder fieldBuilder = AggregationBuilders
                .dateHistogram("articles_over_time")
                .field("PassTime")
                .dateHistogramInterval(new DateHistogramInterval(Precision+"m"))
                .format("yyyy/MM/dd HH:mm")
                .order(BucketOrder.key(true));
        SearchRequest searchRequest = new SearchRequest(motorVehicleNode).source(sourceBuilder.aggregation(fieldBuilder));
        SearchResponse searchResponse = null;
        JSONArray array = new JSONArray();
        JSONArray arrayback = new JSONArray();
        try {
            searchResponse = rhlClient.search(searchRequest, RequestOptions.DEFAULT);
            Histogram histo = searchResponse.getAggregations().get("articles_over_time");
            List<Histogram.Bucket> buckets = (List<Histogram.Bucket>)histo.getBuckets();
            array = JSONArray.parseArray(JSONArray.toJSONString(buckets));
            if(CollectionUtils.isNotEmpty(array)){
                for(int i=0;i<array.size();i++){
                    JSONObject object1 = new JSONObject();
                    object1.put("num",String.valueOf(i+1));
                    object1.put("datetime",array.getJSONObject(i).getString("keyAsString"));
                    object1.put("flownumber",array.getJSONObject(i).getString("docCount"));
                    arrayback.add(object1);
                }
            }
        } catch (IOException e1) {
            e1.printStackTrace();
        }

        TrafficFlow trafficFlow = new TrafficFlow();
        trafficFlow.setSearchnum(searchnum);
        if(CollectionUtils.isNotEmpty(arrayback)){
            List<TrafficFlow> list = JSONArray.parseArray(JSONArray.toJSONString(arrayback),TrafficFlow.class);
            trafficFlow.setTrafficFlows(list);
        }
        return trafficFlow;

    }




}

3）编写逻辑，线程执行完成后获取返回值

package com.neusoft.demo.server.service.impl;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.neusoft.demo.server.model.BeginAndEndDate;
import com.neusoft.demo.server.model.TrafficFlow;
import com.neusoft.demo.server.service.VbdTrafficFlowService;
import com.neusoft.demo.server.thread.ThreadTaskProcess;
import com.neusoft.demo.server.utils.DateSplitUtils;
import com.neusoft.demo.server.utils.RedisUtil;
import org.apache.commons.collections4.CollectionUtils;
import org.elasticsearch.client.RestHighLevelClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.ObjectFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.Date;
import java.util.List;
import java.util.concurrent.FutureTask;
import java.util.stream.Collectors;

/**
 * @author dume
 * @create 2021-10-20 17:54
 **/
@Service
public class VbdTrafficFlowServiceImpl implements VbdTrafficFlowService {
    private static final Logger log = LoggerFactory.getLogger("adminLogger");

    @Qualifier("MyHighLevelClient")
    @Autowired
    private RestHighLevelClient rhlClient;
    @Value("${elasticsearch.motor-vehicle-node-name}")
    private String motorVehicleNode;

    @Autowired
    protected ThreadPoolTaskExecutor executorService;

    @Autowired
    private ObjectFactory<ThreadTaskProcess> processFactory;

    private static SimpleDateFormat simpleDateFormatOne = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
    private static SimpleDateFormat simpleDateFormatTwo = new SimpleDateFormat("yyyyMMddHHmmss");
    private static int  oneDayTimes = 24*60*60*1000;
    private static int fiveMinutes = 5*60*1000;
    private static int fifthyMinutes = 15*60*1000;
    private static int thirtyMinutes = 30*60*1000;
    private static int sixtyMinutes = 60*60*1000;

    @Override
    public JSONObject getTrafficFlow(String EndTime, String BeginTime, String Precision, String DeviceCityCode, String DeviceID, String PlateColor, String Direction) {
        JSONObject object = new JSONObject();
        List<TrafficFlow> trafficFlows = new ArrayList();
        /**
         * 时间段处理
         */
        int precisionMinutes =  getMinutes(Precision);
        Date start = new Date();
        Date end = new Date();
        List<BeginAndEndDate> beginAndEndDateList = new ArrayList<>();
        //时间分片数

        try{
            start = simpleDateFormatTwo.parse(BeginTime);
            end = simpleDateFormatTwo.parse(EndTime);
        }catch (Exception e){
            e.printStackTrace();
            log.error("时间转化失败",e.getMessage());
        }

        if(end.getTime()-start.getTime()<=oneDayTimes){

            BeginAndEndDate beginAndEndDate =  new BeginAndEndDate();
            beginAndEndDate.setBeginTime(BeginTime);
            beginAndEndDate.setEndTime(EndTime);
            beginAndEndDateList.add(beginAndEndDate);
        }else{
          //将时间转化为可以被精度整除的时间
            long startlong = start.getTime();
            long endlong = end.getTime();
            int precision = Integer.valueOf(Precision);
            precision = precision*60*1000;
            startlong = startlong%precision==0?startlong:startlong-(startlong%precision);
            endlong = endlong%precision==0?endlong:endlong+(precision-(endlong%precision));
            List<DateSplitUtils.DateSplit> dateSplits =
                    DateSplitUtils.splitDate(
                            new Date(startlong),
                            new Date(endlong),
                            DateSplitUtils.IntervalType.DAY,
                            1);
            for (DateSplitUtils.DateSplit dateSplit : dateSplits) {
                BeginAndEndDate beginAndEndDate =  new BeginAndEndDate();
                beginAndEndDate.setBeginTime(dateSplit.getStartDateTimeStr());
                beginAndEndDate.setEndTime(dateSplit.getEndDateTimeStr());
                beginAndEndDateList.add(beginAndEndDate);
            }
        }

        List<FutureTask<Object>> futureTasks = new ArrayList<FutureTask<Object>>();
        JSONObject data = new JSONObject();

        for(int i=0;i<beginAndEndDateList.size();i++){
            futureTasks.add(
                    new FutureTask<>(
                            processFactory.getObject().getInstance(
                                    beginAndEndDateList.get(i).getEndTime(),
                                    beginAndEndDateList.get(i).getBeginTime(),
                                    Precision,
                                    DeviceCityCode,
                                    DeviceID,
                                    PlateColor,
                                    Direction,
                                    i+1
                            )
                    )
            );
        }
        // 加入 线程池
        for (FutureTask<Object> futureTask : futureTasks) {
            executorService.submit(futureTask);
        }
        // 获取线程返回结果
        for (int i = 0; i < futureTasks.size(); i++) {
            try {
                TrafficFlow trafficFlow = (TrafficFlow) futureTasks.get(i).get();
                if(null!=trafficFlow){
                    trafficFlows.add(trafficFlow);
                }

            } catch (Exception e) {
                e.printStackTrace();
                log.error("多线程获取返回结果失败",e.getMessage());
            }
        }
        /**
         * 排序并去空
         */
        List<TrafficFlow> backTrafficFlows = new ArrayList();
        if(CollectionUtils.isNotEmpty(trafficFlows)){
            trafficFlows = trafficFlows
                    .stream()
                    .sorted(Comparator.comparing(TrafficFlow::getSearchnum))
                    .collect(Collectors.toList());
            for(TrafficFlow trafficFlow : trafficFlows){
                if(CollectionUtils.isNotEmpty(trafficFlow.getTrafficFlows())){
                    backTrafficFlows.addAll(trafficFlow.getTrafficFlows());
                }

            }
            int num =1;
            for(TrafficFlow trafficFlow : backTrafficFlows){
                trafficFlow.setNum(num);
                num++;
            }
        }
        data.put("dataList",backTrafficFlows);
        object.put("data",data);
        return object;
    }

    public int getMinutes( String Precision){
        int back;
        switch (Precision){
            case "5":
                back = fiveMinutes;
                break;
            case "15":
                back = fifthyMinutes;
                break;
            case "30":
                back = thirtyMinutes;
                break;
            case "60":
                back = sixtyMinutes;
                break;
            default:
                back = sixtyMinutes;
                break;

        }
        return back;
    }
}

2.实现Runnable， 使用计数器

// 创建线程
public class MyThread implements Runnable {
    private List<MyVo> myVos;
    private CountDownLatch latch;

    public MyThread(List<MyVo> myVos, CountDownLatch latch){
        this.myVos = myVos;
        this.latch = latch;
    }

    @Override
    public void run() {
        // 执行代码逻辑
		 /**
		 *写逻辑
		 */
	
	
		// 计数器计数
        latch.countDown();
    }
}



// 创建线程池
    ExecutorService executorService = new ThreadPoolExecutor(12, 24,
            60L, TimeUnit.SECONDS,
            new LinkedBlockingQueue<>());


public void executorMothod(){

try {
	List<List<MyVo>> list = new ArrayList<>();
	// 创建计数器
    CountDownLatch latch = new CountDownLatch(list.size());
    for (List<MyVo> myVos : list) {
        executorService.submit(new MyThread(myVos, latch));
    }
    // 等待计数器执行完成再获取数据
    latch.await();
} catch (InterruptedException e) {
    log.error(e.getMessage());
}finally {
    if(executorService != null && !executorService.isShutdown()){
        executorService.shutdown();
    }
}
	
log.info("线程执行完毕!");


// 执行后续程序

}