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<>() ;
       //這個是我們的單個的查詢方法，假設(shè)每隔請求都根據(jù)唯一的code進行查詢
  public Map<String,Object> query(String code){
        //這個request是我們自定義的內(nè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;
    }
 //這個是個模擬批量查詢的方法
    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方法中初始化一個定時任務(wù)線程，去定時執(zhí)行我們的查詢?nèi)蝿?wù).具體的任務(wù)實現(xiàn)是我們根據(jù)唯一code查詢出來的結(jié)果集，以code為key轉(zhuǎn)成map，然后我們隊列中的每個Request對象都有自己的唯一code，我們根據(jù)code一一對應(yīng)，給相應(yīng)的future返回對應(yīng)的查詢結(jié)果。
        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());
            //合并之后的結(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);
            }
            //返回對應(yīng)的請求結(jié)果
            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() +
                "---執(zhí)行時間為:" + (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() {
        //定時執(zhí)行的線程池，每隔5毫秒執(zhí)行一次(間隔時間可以由業(yè)務(wù)決定)，把所有堆積的請求
        ScheduledExecutorService executorService = Executors.newScheduledThreadPool(1);
        executorService.scheduleAtFixedRate(() -> {
            //在這里具體執(zhí)行批量查詢邏輯
            int size = queue.size();
            if (size == 0) {
                //若沒有請求堆積，直接返回，等10毫秒再執(zhí)行一次
                return;
            }
            //若有請求堆積把所有請求都拿出來
            List<RequestTest> requestTests = new ArrayList<>();
            for (int i = 0; i < size; i++) {
                //把請求拿出來
                RequestTest poll = queue.poll();
                requestTests.add(poll);
            }
            //至此請求已經(jīng)被合并了
            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);
            //把批查結(jié)果放入一個map
            Map<Long,String> map = new HashMap<>();
            for(User user:nameList){
                map.put(user.getId(),user.getName());
            }
            for (RequestTest requestTest : requestTests){
                //把放在map中的結(jié)果集放回給對應(yīng)的線程
                //future是對應(yīng)每個請求的，因為是每個請求線程都傳了自己的future是對應(yīng)的過來
                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可以使一個線程執(zhí)行操作后，主動返回值給另一個線程
        CompletableFuture<String> future = new CompletableFuture<>();
        RequestTest requestTest = new RequestTest();
        //把future(把future可以認為是線程間的"傳話人")放到等待隊列中去，讓定時調(diào)度的線程池執(zhí)行并返回值
        requestTest.setFuture(future);
        requestTest.setKey(key);
        //把requestTest加入等待隊列(LinkedBlockingDeque)
        queue.add(requestTest);
        //future(傳話人)阻塞直到有值返回
        String userName = future.get();
        System.out.printf("用戶名為:" + userName + "---線程名為:"+Thread.currentThread().getName()+
                "---執(zhí)行時間為:"+(System.currentTimeMillis() - currentMillis)+"\n");
        return userName;
    }