{
	"createBy": "aa",
	"receiverList": [
		"bb",
		"cc"
	]
}

{
	"createBy": "aa",
	"receiverList": [
		"bb"
	]
}
或者
{
	"createBy": "aa",
	"receiverList": [
		"cc"
	]
}
或者
{
	"createBy": "bb",
	"receiverList": [
		"cc"
	]
}
或者
{
	"createBy": "aa",
	"receiverList": [
		"bb",
		"cc"
	]
}

package com.demo;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
import org.springframework.stereotype.Component;
import javax.annotation.PreDestroy;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
/**
 * Description: 請(qǐng)求合并管理類
 */
@Slf4j
@Component
public class RequestMerger {
    // 線程池核心線程數(shù)
    private final int corePoolSize = 200;
    // 任務(wù)執(zhí)行超時(shí)時(shí)間，單位：毫秒
    private final int timeout = 5 * 60 * 1000;
    // 隊(duì)列，隊(duì)列長(zhǎng)度為Integer.MAX_VALUE
    private final LinkedBlockingQueue<String> requestQueue = new LinkedBlockingQueue<>();
    // 定時(shí)器，所有任務(wù)共用線程池
    private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(corePoolSize,
            new CustomizableThreadFactory("schedule-executor-"));
    // 是否關(guān)閉標(biāo)志
    private final AtomicBoolean isShutdown = new AtomicBoolean(false);
    /**
     * 構(gòu)造函數(shù)，用于初始化請(qǐng)求合并器。
     *
     * @param batchSize   每次合并的最大請(qǐng)求數(shù)量。
     * @param delayMillis 合并請(qǐng)求的周期間隔，單位為毫秒。
     */
    public RequestMerger(int batchSize, long delayMillis) {
        // 啟動(dòng)定時(shí)器，定期合并請(qǐng)求，延遲delayMillis后開(kāi)始，之后每隔delayMillis執(zhí)行一次
        scheduler.scheduleAtFixedRate(() -> {
            if (!isShutdown.get()) {
                List<String> batch = new ArrayList<>(batchSize);
                int drainedCount = requestQueue.drainTo(batch, batchSize);
                log.info("==>scheduler，drainedCount：{}，nowQueueCount：{}", drainedCount, requestQueue.size());
                if (!batch.isEmpty()) {
                    // 異步執(zhí)行任務(wù)，防止業(yè)務(wù)執(zhí)行時(shí)間過(guò)長(zhǎng)導(dǎo)致業(yè)務(wù)整體延遲過(guò)大
                    scheduler.submit(() -> {
                        sendRequestBatch(batch);
                    });
                }
            }
        }, delayMillis, delayMillis, TimeUnit.MILLISECONDS);
    }
    /**
     * 發(fā)送請(qǐng)求批次的方法。
     *
     * @param batch 請(qǐng)求批次。
     */
    private void sendRequestBatch(List<String> batch) {
        Future<?> future = scheduler.submit(() -> {
            try {
                // 在這里實(shí)現(xiàn)你的請(qǐng)求發(fā)送邏輯
                // 可以使用HTTP客戶端庫(kù)（如Apache HttpClient或OkHttp）來(lái)發(fā)送請(qǐng)求
                // ...
                System.out.println("Sending batch of " + batch.size() + " requests");
            } catch (Exception e) {
                // 異常處理邏輯
                System.err.println("Error sending requests: " + e.getMessage());
            }
        });
        // 嘗試獲取任務(wù)結(jié)果，如果超過(guò)超時(shí)時(shí)間則拋出TimeoutException異常，進(jìn)行取消任務(wù)
        try {
            // 超時(shí)時(shí)間，單位：毫秒
            future.get(timeout, TimeUnit.MILLISECONDS);
        } catch (TimeoutException | ExecutionException e) {
            // 超時(shí)或執(zhí)行異常時(shí)取消任務(wù)
            future.cancel(true);
        } catch (Exception e) {
            log.error("==>任務(wù)執(zhí)行異常", e);
            // 任務(wù)執(zhí)行異常
            future.cancel(true);
        }
    }
    /**
     * 在對(duì)象銷毀前執(zhí)行的關(guān)閉操作。
     * 該方法從請(qǐng)求隊(duì)列中拉取所有未處理的請(qǐng)求，并將它們批量發(fā)送。
     * 無(wú)參數(shù)和返回值。
     */
    @PreDestroy
    public void shutdown() {
        isShutdown.set(true);
        List<String> batch = new ArrayList<>();
        // 獲取請(qǐng)求隊(duì)列中的剩余所有請(qǐng)求
        int drainedCount = requestQueue.drainTo(batch);
        log.info("==>shutdown，drainedCount：{}，nowQueueCount：{}", drainedCount, requestQueue.size());
        // 批量發(fā)送收集到的剩余請(qǐng)求
        sendRequestBatch(batch);
        // 關(guān)閉定時(shí)執(zhí)行器
        scheduler.shutdown();
        try {
            if (!scheduler.awaitTermination(60, TimeUnit.SECONDS)) {
                log.error("Scheduler did not terminate gracefully within 60 seconds, force shutting down.");
                scheduler.shutdownNow();
            }
        } catch (InterruptedException e) {
            log.warn("Interrupted during scheduler termination, force shutting down.");
            scheduler.shutdownNow();
            Thread.currentThread().interrupt();
        }
    }
    /**
     * 向請(qǐng)求隊(duì)列中添加一個(gè)請(qǐng)求。如果服務(wù)未關(guān)閉，則直接添加到請(qǐng)求隊(duì)列中；
     * 如果服務(wù)已關(guān)閉，則將該請(qǐng)求作為一批請(qǐng)求發(fā)送。
     *
     * @param request 要添加的請(qǐng)求字符串。
     */
    public void addRequest(String request) throws InterruptedException {
        // 檢查服務(wù)是否已關(guān)閉
        if (!isShutdown.get()) {
            // 未關(guān)閉，直接添加到請(qǐng)求隊(duì)列
            requestQueue.put(request);
        } else {
            // 已關(guān)閉，將當(dāng)前請(qǐng)求作為一批發(fā)送
            List<String> batch = new ArrayList<>();
            batch.add(request);
            sendRequestBatch(batch);
        }
    }
}