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 {
//用來(lái)存放請(qǐng)求的隊(duì)列，我們將請(qǐng)求封裝成了一個(gè)Request對(duì)象
 private LinkedBlockingQueue<Request> queue = new LinkedBlockingQueue<>() ;
       //這個(gè)是我們的單個(gè)的查詢(xún)方法，假設(shè)每隔請(qǐng)求都根據(jù)唯一的code進(jìn)行查詢(xún)
  public Map<String,Object> query(String code){
        //這個(gè)request是我們自定義的內(nèi)部類(lèi)
        Request request = new Request();
        request.code = code;
        CompletableFuture<Map<String,Object>> future = new CompletableFuture<>();
        request.future = future;
        queue.add(request);
        //阻塞 直到返回結(jié)果
        try {
            return future.get();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
        return null;
    }
 //這個(gè)是個(gè)模擬批量查詢(xún)的方法
    public List<Map<String,Object>> batchQuery(List<String> codes){
        return null;
    }
    //封裝的請(qǐng)求
    class Request {
        String code;
        CompletableFuture<Map<String,Object>> future;
    }
    @PostConstruct
    public void init(){
    //在init方法中初始化一個(gè)定時(shí)任務(wù)線(xiàn)程，去定時(shí)執(zhí)行我們的查詢(xún)?nèi)蝿?wù).具體的任務(wù)實(shí)現(xiàn)是我們根據(jù)唯一code查詢(xún)出來(lái)的結(jié)果集，以code為key轉(zhuǎn)成map，然后我們隊(duì)列中的每個(gè)Request對(duì)象都有自己的唯一code，我們根據(jù)code一一對(duì)應(yīng)，給相應(yīng)的future返回對(duì)應(yīng)的查詢(xún)結(jié)果。
        ScheduledExecutorService poolExecutor = new ScheduledThreadPoolExecutor(1);
        poolExecutor.scheduleAtFixedRate(()->{
            int size = queue.size();
            //如果沒(méi)有請(qǐng)求直接返回
            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());
            //合并之后的結(jié)果集
            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);
            }
            //返回對(duì)應(yīng)的請(qǐng)求結(jié)果
            for (Request request : list) {
                Map<String, Object> response = responseMap.get(request.code);
                request.future.complete(response);
            }
        },0,10,TimeUnit.MILLISECONDS);
    }
}

//單查詢(xún)接口
@GetMapping("/getUser")
    public String getUser(Long key){
        long currentMillis = System.currentTimeMillis();
        //單查詢(xún)service，大量線(xiàn)程懟到這個(gè)service上去
        String userName = userService.getUser(key);
        System.out.printf("##############################################\n");
        System.out.printf("用戶(hù)名為:" + userName + "---線(xiàn)程名為:" + Thread.currentThread().getName() +
                "---執(zhí)行時(shí)間為:" + (System.currentTimeMillis() - currentMillis) + "\n");
        return userName;
    }

//用戶(hù)請(qǐng)求類(lèi)
public class RequestTest {
    //請(qǐng)求條件
    private Long key;
    //傳話(huà)人
    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;
    }
}

//存放請(qǐng)求的隊(duì)列
    LinkedBlockingDeque<RequestTest> queue = new LinkedBlockingDeque<>();
    //初始化方法
    @PostConstruct
    public void init() {
        //定時(shí)執(zhí)行的線(xiàn)程池，每隔5毫秒執(zhí)行一次(間隔時(shí)間可以由業(yè)務(wù)決定)，把所有堆積的請(qǐng)求
        ScheduledExecutorService executorService = Executors.newScheduledThreadPool(1);
        executorService.scheduleAtFixedRate(() -> {
            //在這里具體執(zhí)行批量查詢(xún)邏輯
            int size = queue.size();
            if (size == 0) {
                //若沒(méi)有請(qǐng)求堆積，直接返回，等10毫秒再執(zhí)行一次
                return;
            }
            //若有請(qǐng)求堆積把所有請(qǐng)求都拿出來(lái)
            List<RequestTest> requestTests = new ArrayList<>();
            for (int i = 0; i < size; i++) {
                //把請(qǐng)求拿出來(lái)
                RequestTest poll = queue.poll();
                requestTests.add(poll);
            }
            //至此請(qǐng)求已經(jīng)被合并了
            System.out.printf("##############################################\n");
            System.out.printf("請(qǐng)求合并了" + requestTests.size() + "條！\n");
            //組裝批量查詢(xún)條件
            List<Long> keyList = new ArrayList<>();
            for (RequestTest requestTest : requestTests) {
                keyList.add(requestTest.getKey());
            }
            //進(jìn)行批量查詢(xún)
            List<User> nameList = userService.getUserList(keyList);
            //把批查結(jié)果放入一個(gè)map
            Map<Long,String> map = new HashMap<>();
            for(User user:nameList){
                map.put(user.getId(),user.getName());
            }
            for (RequestTest requestTest : requestTests){
                //把放在map中的結(jié)果集放回給對(duì)應(yīng)的線(xiàn)程
                //future是對(duì)應(yīng)每個(gè)請(qǐng)求的，因?yàn)槭敲總€(gè)請(qǐng)求線(xiàn)程都傳了自己的future是對(duì)應(yīng)的過(guò)來(lái)
                requestTest.getFuture().complete(map.get(requestTest.getKey()));
            }
        }, 0, 5, TimeUnit.MILLISECONDS);
    }
    //請(qǐng)求合并
    @GetMapping("/requestMerge/getUser")
    public String getUserRequestMerge(Long key) throws InterruptedException, ExecutionException {
        long currentMillis = System.currentTimeMillis();
        //CompletableFuture可以使一個(gè)線(xiàn)程執(zhí)行操作后，主動(dòng)返回值給另一個(gè)線(xiàn)程
        CompletableFuture<String> future = new CompletableFuture<>();
        RequestTest requestTest = new RequestTest();
        //把future(把future可以認(rèn)為是線(xiàn)程間的"傳話(huà)人")放到等待隊(duì)列中去，讓定時(shí)調(diào)度的線(xiàn)程池執(zhí)行并返回值
        requestTest.setFuture(future);
        requestTest.setKey(key);
        //把requestTest加入等待隊(duì)列(LinkedBlockingDeque)
        queue.add(requestTest);
        //future(傳話(huà)人)阻塞直到有值返回
        String userName = future.get();
        System.out.printf("用戶(hù)名為:" + userName + "---線(xiàn)程名為:"+Thread.currentThread().getName()+
                "---執(zhí)行時(shí)間為:"+(System.currentTimeMillis() - currentMillis)+"\n");
        return userName;
    }