* @Author: xiaokunkun
 * @CreateTime: 2023-04-23  14:37
 * @Description: 合并更新，可以不捕捉異常報(bào)錯(cuò)后外層調(diào)用方直接捕獲異常事務(wù)回滾
 */
@Service
public class CommodityAmountService {
    class Request {
        AcctUpdateDto acctUpdateDto;
        //預(yù)留字段 可不使用
        String atomCode;
        //暫定返回結(jié)果為true或者false
        CompletableFuture<Boolean> future; // 接受結(jié)果
    }
 
    // 積攢請(qǐng)求（每隔N毫秒批量處理一次）
    LinkedBlockingQueue<Request> queue = new LinkedBlockingQueue<>();
 
    // 定時(shí)任務(wù)的實(shí)現(xiàn)，N秒鐘處理一次數(shù)據(jù)
    @PostConstruct
    public void init() {
        ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(5);
        scheduledExecutorService.scheduleAtFixedRate(() -> {
            // 1、取出queue的請(qǐng)求，生成一次合并更新
            int size = queue.size();
            if (size == 0) {
                return;
            }
            ArrayList<Request> requests = new ArrayList<>();
            for (int i = 0; i < size; i++) {
                //隊(duì)列出棧
                Request request = queue.poll();
                requests.add(request);
            }
            System.out.println("批量處理數(shù)據(jù)量:" + size);
            // 2、組裝一個(gè)合并更新 key為賬戶value為sum(amount)
            Map<String, Long> amountMap = new HashMap<>();
            ArrayList<String> commodityCodes = new ArrayList<>();
            for (Request request : requests) {
                //todo 根據(jù)accountNo分組
            }
            for (String key : amountMap.keySet()) {
                Long amount = amountMap.get(key);
                //update mysql
            }
            // 3、將結(jié)果響應(yīng) 分發(fā)給每一個(gè)單獨(dú)的用戶請(qǐng)求。由定時(shí)任務(wù)處理線程 --> n個(gè)用戶的請(qǐng)求線程
            for (Request request : requests) {
                // 將結(jié)果返回到對(duì)應(yīng)的請(qǐng)求線程,只要不報(bào)錯(cuò)此批次全部返回true，否則false
                request.future.complete(true);
            }}, 0, 1000, TimeUnit.MILLISECONDS);
    }
 
    @Autowired
    CommodityRemoteService commodityRemoteService;
 
    // 合并金額并更新，多個(gè)用戶請(qǐng)求
    public Boolean updateMergeAmount(String movieCode)
            throws ExecutionException, InterruptedException {
        // 并非立刻發(fā)起接口調(diào)用，請(qǐng)求收集起來(lái)，再進(jìn)行
        Request request = new Request();
        request.atomCode = movieCode;
        // 異步編程：獲取異步處理的結(jié)果
        CompletableFuture<Boolean> future = new CompletableFuture<>();
        request.future = future;
        queue.add(request);
        return future.get(); // 此處get方法，會(huì)阻塞線程運(yùn)行，直到future有返回
    }
}

//模擬500的并發(fā)量
public void updateMerge() {
    AcctCmdDriver acctCmdDriver = new AcctCmdDriver();
    TradeAccntOrderDetail detail = new TradeAccntOrderDetail();
    AcctNoInfo acctNoInfo = new AcctNoInfo();
    OrderConsist consistForOrder = OrderConsist.newInstance("0200_202304", "trade_accnt_merchant_order");
    detail.setConsistForOrder(consistForOrder);
    acctCmdDriver.setDetail(detail);
    acctCmdDriver.setAcctNoInfo(acctNoInfo);
    detail.setAccountCategory(AccountCategoryEnum.MERCHANT);
    detail.setAccountNo("02020001010000262977202304");
    detail.setAmount(6l);
    System.out.println("start build thread" + acctCmdDriver);
    Random rand = new Random();
    for (int i = 1; i <= 500; i++) {
        final String index = "code_" + i;
        Thread thread = new Thread(() -> {
            try {
                System.out.println("amount is:" + detail.getAmount());
                countDownLatch.await();
                Thread.sleep(rand.nextInt(150));
                Boolean res = updateMergeAmountService.mergeUpdate(acctCmdDriver);
                System.out.println("current i" + index + "res:" + res);
            } catch (InterruptedException e) {
                System.out.println("thread error is:" + e);
            }
        });
        thread.start();
        // 啟動(dòng)后，倒計(jì)時(shí)器倒計(jì)數(shù)減一，代表又有一個(gè)線程就緒了
        countDownLatch.countDown();
    }
}