接口請(qǐng)求合并技巧，用好了效率直接翻倍！

一、什么是請(qǐng)求合并

在WEB項(xiàng)目中，我們一般會(huì)使用HTTP協(xié)議來處理請(qǐng)求

那么我們與服務(wù)器交互方式將會(huì)是這樣的，一次請(qǐng)求，一次處理

我們都知道，調(diào)用批量接口相比調(diào)用非批量接口有更大的性能優(yōu)勢（因?yàn)闇p少了IO交互操作），在高并發(fā)情況下，如果有非常頻繁的接口請(qǐng)求發(fā)生的話，我們則可以考慮請(qǐng)求合并了，將多個(gè)請(qǐng)求進(jìn)行一定的等待延遲，當(dāng)請(qǐng)求累計(jì)達(dá)到一定量級(jí)的時(shí)候，進(jìn)行批量請(qǐng)求處理

二、請(qǐng)求合并的優(yōu)缺點(diǎn)

所謂請(qǐng)求合并，就是講多次請(qǐng)求合并為一次批量請(qǐng)求

優(yōu)點(diǎn):

將多次請(qǐng)求處理進(jìn)行一定時(shí)間或請(qǐng)求數(shù)量的等待，使之合并成為一次請(qǐng)求，減少IO交互

缺點(diǎn):

由于請(qǐng)求需要等待指定時(shí)間或指定請(qǐng)求數(shù)量，所以合并的接口存在延時(shí)，故對(duì)請(qǐng)求合并的接口有所限制，該接口不能對(duì)響應(yīng)及時(shí)性有要求，支持一定時(shí)間的延遲

三、請(qǐng)求合并技術(shù)實(shí)現(xiàn)

采用定時(shí)線程池ScheduledExecutorService，與內(nèi)存隊(duì)列LinkedBlockingDeque進(jìn)行實(shí)現(xiàn)請(qǐng)求合并

?

原理是將用戶的請(qǐng)求進(jìn)行緩存起來，緩存的請(qǐng)求數(shù)量達(dá)到指定數(shù)量或達(dá)到定時(shí)線程池執(zhí)行時(shí)，將已有多個(gè)單請(qǐng)求處理合并為多處理，調(diào)用批量接口進(jìn)行操作

?

依賴

• 只需要JDK,無需任何第三方依賴

批量請(qǐng)求合并工具類定義如下:

核心原理就是 將請(qǐng)求放入隊(duì)列，放入時(shí)檢測內(nèi)存隊(duì)列數(shù)量是否超過設(shè)置閾值，以及時(shí)間閾值到期觸發(fā)定時(shí)線程池執(zhí)行

package com.leilei.support;

import lombok.extern.log4j.Log4j2;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;

/**
 * @author lei
 * @desc 請(qǐng)求合并工具類
 **/
@Log4j2
public class BatchCollapser<T, R> {
    private static final Map<Class, BatchCollapser> BATCH_INSTANCE =new ConcurrentHashMap<>();
    private static final ScheduledExecutorService SCHEDULE_EXECUTOR = Executors.newScheduledThreadPool(1);

    private final LinkedBlockingDeque<T> batchContainer = new LinkedBlockingDeque<>();
    private final BatchHandler<List<T>, R> handler;
    private final int countThreshold;

    /**
     * constructor
     *
     * @param handler        處理器
     * @param countThreshold 數(shù)量閾值，達(dá)到此閾值后觸發(fā)處理器
     * @param timeThreshold  時(shí)間閾值，達(dá)到此時(shí)間后觸發(fā)處理器
     */
    private BatchCollapser(BatchHandler<List<T>, R> handler, int countThreshold, long timeThreshold) {
        this.handler = handler;
        this.countThreshold = countThreshold;
        SCHEDULE_EXECUTOR.scheduleAtFixedRate(() -> {
            try {
                this.popUpAndHandler(BatchHandlerType.BATCH_HANDLER_TYPE_TIME);
            } catch (Exception e) {
                log.error("pop-up container exception", e);
            }
        }, timeThreshold, timeThreshold, TimeUnit.SECONDS);
    }

    /**
     * 添加請(qǐng)求元素入隊(duì)
     * @param event
     */
    public void addRequestParam(T event) {
        batchContainer.add(event);
        if (batchContainer.size() >= countThreshold) {
            popUpAndHandler(BatchHandlerType.BATCH_HANDLER_TYPE_DATA);
        }
    }

    /**
     * 從隊(duì)列獲取請(qǐng)求,并進(jìn)行批量處理
     * @param handlerType
     */
    private void popUpAndHandler(BatchHandlerType handlerType) {
        List<T> tryHandlerList = Collections.synchronizedList(new ArrayList<>(countThreshold));
        batchContainer.drainTo(tryHandlerList, countThreshold);
        if (tryHandlerList.size() < 1) {
            return;
        }

        try {
            R handle = handler.handle(tryHandlerList, handlerType);
            log.info("批處理工具執(zhí)行result:{}", handle);
        } catch (Exception e) {
            log.error("batch execute error, transferList:{}", tryHandlerList, e);
        }
    }

    /**
     * 獲取合并器實(shí)例
     *
     * @param batchHandler   處理執(zhí)行器
     * @param countThreshold 閾值數(shù)量(隊(duì)列數(shù)量)
     * @param timeThreshold  閾值時(shí)間 單位秒（目前設(shè)置是觸發(fā)后獲取閾值數(shù)量請(qǐng)求，可根據(jù)需要修改）
     * @param <E>
     * @param <R>
     * @return
     */
    public static <E, R> BatchCollapser<E, R> getInstance(BatchHandler<List<E>, R> batchHandler, int countThreshold, long timeThreshold) {
        Class jobClass = batchHandler.getClass();
        if (BATCH_INSTANCE.get(jobClass) == null) {
            synchronized (BatchCollapser.class) {
                BATCH_INSTANCE.putIfAbsent(jobClass, new BatchCollapser<>(batchHandler, countThreshold, timeThreshold));
            }
        }
        return BATCH_INSTANCE.get(jobClass);
    }

    /**
     * 請(qǐng)求處理接口
     *
     * @param <T>
     * @param <R>
     */
    public interface BatchHandler<T, R> {
        /**
         * 處理用戶具體請(qǐng)求
         *
         * @param input
         * @param handlerType
         * @return
         */
        R handle(T input, BatchHandlerType handlerType);
    }

    /**
     * 合并執(zhí)行類型枚舉
     */
    public enum BatchHandlerType {
        /**
         * 數(shù)量類型
         */
        BATCH_HANDLER_TYPE_DATA,

        /**
         * 時(shí)間類型
         */
        BATCH_HANDLER_TYPE_TIME,
    }
}

使用方式如下:

package com.leilei.support;

import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.util.List;

/**
 * @author lei
 * @desc
 **/
@Service
public class ProductService implements BatchCollapser.BatchHandler<List<Integer>, Integer> {
    private BatchCollapser<Integer, Integer> batchCollapser;

    @PostConstruct
    private void postConstructorInit() {
        // 當(dāng)請(qǐng)求數(shù)量達(dá)到20個(gè)，或每過5s合并執(zhí)行一次請(qǐng)求
        batchCollapser = BatchCollapser.getInstance(ProductService.this, 20, 5);
    }

    @Override
    public Integer handle(List<Integer> input, BatchCollapser.BatchHandlerType handlerType) {
        System.out.println("處理類型:" + handlerType + ",接受到批量請(qǐng)求參數(shù):" + input);
        return input.stream().mapToInt(x -> x).sum();
    }


    /**
     * 假設(shè)我這里是300ms一次請(qǐng)求
     */
    @Scheduled(fixedDelay = 300)
    public void aaa() {
        Integer requestParam = (int) (Math.random() * 100) + 1;
        batchCollapser.addRequestParam(requestParam);
        System.out.println("當(dāng)前請(qǐng)求參數(shù):" + requestParam);

    }
}

@Data
public class Product {
    private Integer id;
    private String notes;
}

當(dāng)然以上工具類僅僅只是DEMO，各位大佬可自行完善，權(quán)衡請(qǐng)求合并利弊，降低服務(wù)器在高并發(fā)請(qǐng)求時(shí)的壓力