Java高并发下请求合并处理方式

package cn.codingxiaxw.combine;
import org.springframework.stereotype.Service;
import javax.annotation.PostConstruct;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;
import java.util.stream.Collectors;
@Service
public class QueryService {
//用来存放请求的队列，我们将请求封装成了一个Request对象
 private LinkedBlockingQueue<Request> queue = new LinkedBlockingQueue<>() ;
       //这个是我们的单个的查询方法，假设每隔请求都根据唯一的code进行查询
  public Map<String,Object> query(String code){
        //这个request是我们自定义的内部类
        Request request = new Request();
        request.code = code;
        CompletableFuture<Map<String,Object>> future = new CompletableFuture<>();
        request.future = future;
        queue.add(request);
        //阻塞 直到返回结果
        try {
            return future.get();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
        return null;
    }
 //这个是个模拟批量查询的方法
    public List<Map<String,Object>> batchQuery(List<String> codes){
        return null;
    }
    //封装的请求
    class Request {
        String code;
        CompletableFuture<Map<String,Object>> future;
    }
    @PostConstruct
    public void init(){
    //在init方法中初始化一个定时任务线程，去定时执行我们的查询任务.具体的任务实现是我们根据唯一code查询出来的结果集，以code为key转成map，然后我们队列中的每个Request对象都有自己的唯一code，我们根据code一一对应，给相应的future返回对应的查询结果。
        ScheduledExecutorService poolExecutor = new ScheduledThreadPoolExecutor(1);
        poolExecutor.scheduleAtFixedRate(()->{
            int size = queue.size();
            //如果没有请求直接返回
            if(size==0)
                return ;
            List<Request> list = new ArrayList<>();
            for (int i = 0; i < size;i++){
                Request request = queue.poll();
                list.add(request);
            }
            System.out.println("批量处理:"+size);
            List<String> codes = list.stream().map(s->s.code).collect(Collectors.toList());
            //合并之后的结果集
            List<Map<String, Object>> batchResult = batchQuery(codes);
            Map<String,Map<String,Object>> responseMap = new HashMap<>();
            for (Map<String,Object> result : batchResult) {
                String code = result.get("code").toString();
                responseMap.put(code,result);
            }
            //返回对应的请求结果
            for (Request request : list) {
                Map<String, Object> response = responseMap.get(request.code);
                request.future.complete(response);
            }
        },0,10,TimeUnit.MILLISECONDS);
    }
}

//单查询接口
@GetMapping("/getUser")
    public String getUser(Long key){
        long currentMillis = System.currentTimeMillis();
        //单查询service，大量线程怼到这个service上去
        String userName = userService.getUser(key);
        System.out.printf("##############################################\n");
        System.out.printf("用户名为:" + userName + "---线程名为:" + Thread.currentThread().getName() +
                "---执行时间为:" + (System.currentTimeMillis() - currentMillis) + "\n");
        return userName;
    }

//用户请求类
public class RequestTest {
    //请求条件
    private Long key;
    //传话人
    private CompletableFuture<String> future;
    public CompletableFuture<String> getFuture() {
        return future;
    }
    public void setFuture(CompletableFuture<String> future) {
        this.future = future;
    }
    public Long getKey() {
        return key;
    }
    public void setKey(Long key) {
        this.key = key;
    }
}

//存放请求的队列
    LinkedBlockingDeque<RequestTest> queue = new LinkedBlockingDeque<>();
    //初始化方法
    @PostConstruct
    public void init() {
        //定时执行的线程池，每隔5毫秒执行一次(间隔时间可以由业务决定)，把所有堆积的请求
        ScheduledExecutorService executorService = Executors.newScheduledThreadPool(1);
        executorService.scheduleAtFixedRate(() -> {
            //在这里具体执行批量查询逻辑
            int size = queue.size();
            if (size == 0) {
                //若没有请求堆积，直接返回，等10毫秒再执行一次
                return;
            }
            //若有请求堆积把所有请求都拿出来
            List<RequestTest> requestTests = new ArrayList<>();
            for (int i = 0; i < size; i++) {
                //把请求拿出来
                RequestTest poll = queue.poll();
                requestTests.add(poll);
            }
            //至此请求已经被合并了
            System.out.printf("##############################################\n");
            System.out.printf("请求合并了" + requestTests.size() + "条！\n");
            //组装批量查询条件
            List<Long> keyList = new ArrayList<>();
            for (RequestTest requestTest : requestTests) {
                keyList.add(requestTest.getKey());
            }
            //进行批量查询
            List<User> nameList = userService.getUserList(keyList);
            //把批查结果放入一个map
            Map<Long,String> map = new HashMap<>();
            for(User user:nameList){
                map.put(user.getId(),user.getName());
            }
            for (RequestTest requestTest : requestTests){
                //把放在map中的结果集放回给对应的线程
                //future是对应每个请求的，因为是每个请求线程都传了自己的future是对应的过来
                requestTest.getFuture().complete(map.get(requestTest.getKey()));
            }
        }, 0, 5, TimeUnit.MILLISECONDS);
    }
    //请求合并
    @GetMapping("/requestMerge/getUser")
    public String getUserRequestMerge(Long key) throws InterruptedException, ExecutionException {
        long currentMillis = System.currentTimeMillis();
        //CompletableFuture可以使一个线程执行操作后，主动返回值给另一个线程
        CompletableFuture<String> future = new CompletableFuture<>();
        RequestTest requestTest = new RequestTest();
        //把future(把future可以认为是线程间的"传话人")放到等待队列中去，让定时调度的线程池执行并返回值
        requestTest.setFuture(future);
        requestTest.setKey(key);
        //把requestTest加入等待队列(LinkedBlockingDeque)
        queue.add(requestTest);
        //future(传话人)阻塞直到有值返回
        String userName = future.get();
        System.out.printf("用户名为:" + userName + "---线程名为:"+Thread.currentThread().getName()+
                "---执行时间为:"+(System.currentTimeMillis() - currentMillis)+"\n");
        return userName;
    }