后端性能优化之服务端请求合并与响应合并（批处理）的设计与实现

一、问题描述

在高并发场景中，服务端可能接收到大量类似的小粒度请求（如获取商品基本信息、查询用户余额等）。每个请求都需要经历网络传输、协议解析、数据库查询、结果封装等完整链路，产生大量重复开销，导致：

1. 网络I/O负载过高：每个请求都需要独立的TCP连接/HTTP请求
2. 数据库压力剧增：大量简单查询造成短连接风暴和锁竞争
3. 服务端资源浪费：线程频繁切换，内存重复分配
4. 整体吞吐量下降：单请求处理效率低下

请求合并与响应合并（又称批处理）技术通过将多个独立的客户端请求在服务端合并为单个批量操作，减少重复开销，提升系统吞吐量。本文将详解其设计原理与实现方案。

二、核心原理剖析

2.1 基本思路

原始模式： [请求A] → 处理A → 响应A
           [请求B] → 处理B → 响应B
           [请求C] → 处理C → 响应C
           
批处理模式： [请求A+B+C] → 批量处理 → [响应A+B+C]

关键转变：将N次顺序/并发的离散操作合并为1次批量操作。

2.2 技术收益分析

• 网络层：减少TCP握手、SSL握手、HTTP头部重复传输
• 数据库层：将多个SELECT合并为SELECT ... WHERE id IN (?)，利用索引一次检索
• 服务端：减少线程调度、上下文切换、对象序列化/反序列化次数
• 客户端：降低请求延迟（特别是高网络延迟场景）

三、设计模式详解

3.1 合并触发时机策略

1) 定时触发合并

// 伪代码示例：基于时间窗口的合并器
public class TimedBatchProcessor<T> {
    private List<T> batchBuffer = new ArrayList<>();
    private long windowMillis = 50; // 50ms窗口
    private volatile boolean processing = false;
    
    public void addRequest(T request) {
        synchronized (batchBuffer) {
            batchBuffer.add(request);
            if (!processing) {
                processing = true;
                // 窗口到期后处理累积的请求
                scheduler.schedule(this::processBatch, windowMillis);
            }
        }
    }
    
    private void processBatch() {
        List<T> currentBatch;
        synchronized (batchBuffer) {
            currentBatch = new ArrayList<>(batchBuffer);
            batchBuffer.clear();
            processing = false;
        }
        // 执行批量处理
        executeBatch(currentBatch);
    }
}

适用场景：实时性要求不高，请求量稳定的业务（如日志上报、监控数据采集）

2) 数量触发合并

// 伪代码：达到阈值立即处理
public class CountTriggerBatchProcessor<T> {
    private final int batchSize = 100; // 每100个请求触发一次
    private Queue<T> buffer = new ConcurrentLinkedQueue<>();
    
    public void addRequest(T request) {
        buffer.offer(request);
        if (buffer.size() >= batchSize) {
            List<T> batch = new ArrayList<>(batchSize);
            for (int i = 0; i < batchSize; i++) {
                batch.add(buffer.poll());
            }
            // 异步处理批量请求
            threadPool.submit(() -> processBatch(batch));
        }
    }
}

适用场景：突发流量大，需要快速响应的场景（如秒杀商品查询）

3) 混合触发策略

结合时间和数量双重条件：

• 条件A：缓冲请求数≥N（如50个）
• 条件B：距离上次处理时间≥T（如200ms）
• 满足任一条件即触发批量处理

3.2 请求-响应映射机制

这是批处理的核心难点：需要保证每个响应能正确返回给对应的请求方。

方案一：ID映射表

public class BatchContext {
    // 批量请求结构
    static class BatchRequest {
        String batchId;          // 批次ID
        List<SingleRequest> requests; // 原始请求列表
        Map<String, CompletableFuture<Response>> futures; // 请求ID → Future映射
    }
    
    // 客户端发送单个请求时
    public CompletableFuture<Response> submitRequest(SingleRequest req) {
        CompletableFuture<Response> future = new CompletableFuture<>();
        String requestId = UUID.randomUUID().toString();
        
        // 将请求加入批处理缓冲池
        BatchBuffer buffer = getCurrentBuffer();
        buffer.addRequest(requestId, req, future);
        
        // 触发合并检查
        buffer.tryFlush();
        
        return future;
    }
    
    // 批量处理完成后
    void onBatchComplete(String batchId, List<Response> batchResponses) {
        BatchRequest batch = removeBatch(batchId);
        for (int i = 0; i < batch.requests.size(); i++) {
            Response singleResp = batchResponses.get(i);
            CompletableFuture<Response> future = batch.futures.get(batch.requests.get(i).getId());
            future.complete(singleResp);
        }
    }
}

方案二：顺序保持法

适用于请求和响应顺序严格对应的场景：

// 利用数组索引建立映射关系
public class OrderedBatchProcessor {
    public List<Response> processBatch(List<Request> requests) {
        // 1. 批量执行操作（如数据库IN查询）
        List<Result> batchResults = batchQuery(requests);
        
        // 2. 按原始顺序重组响应
        List<Response> responses = new ArrayList<>(requests.size());
        Map<Key, Result> resultMap = batchResults.stream()
            .collect(Collectors.toMap(Result::getKey, r -> r));
        
        for (Request req : requests) {
            Response resp = convertToResponse(resultMap.get(req.getKey()));
            responses.add(resp);
        }
        
        return responses;
    }
}

四、实战场景与优化策略

4.1 数据库查询合并

原始SQL（N次查询）：

SELECT * FROM products WHERE id = 1001;
SELECT * FROM products WHERE id = 1002;
SELECT * FROM products WHERE id = 1003;

合并后SQL（1次查询）：

SELECT * FROM products WHERE id IN (1001, 1002, 1003);

注意事项：

1. IN列表长度限制：MySQL默认max_allowed_packet限制，需分片处理
2. 索引失效风险：IN列表过大可能导致索引失效，需测试分界点
3. 结果集映射：确保查询结果能正确映射回原始请求

4.2 HTTP请求合并

方案示例：GraphQL或自定义批量端点

# GraphQL批量查询
query {
  product1: product(id: "1001") { name price }
  product2: product(id: "1002") { name price }
  user1: user(id: "2001") { name balance }
}

或RESTful批量接口：

POST /api/batch
Content-Type: application/json

{
  "requests": [
    {"method": "GET", "path": "/products/1001"},
    {"method": "POST", "path": "/orders", "body": {...}},
    {"method": "GET", "path": "/users/2001/balance"}
  ]
}

4.3 异步IO优化

// 使用CompletableFuture实现异步批处理
public class AsyncBatchProcessor {
    private final ExecutorService dbExecutor = Executors.newFixedThreadPool(10);
    private final ExecutorService callbackExecutor = Executors.newCachedThreadPool();
    
    public CompletableFuture<List<Response>> processAsync(List<Request> requests) {
        // 阶段1：参数预处理（CPU密集型）
        return CompletableFuture.supplyAsync(() -> prepareParams(requests))
            // 阶段2：批量数据库查询（IO密集型）
            .thenApplyAsync(params -> executeBatchQuery(params), dbExecutor)
            // 阶段3：结果转换（CPU密集型）
            .thenApplyAsync(results -> convertResults(results), callbackExecutor)
            // 异常处理
            .exceptionally(ex -> handleBatchException(ex, requests));
    }
}

五、高级挑战与解决方案

5.1 超时与部分失败处理

public class ResilientBatchProcessor {
    public BatchResult processWithResilience(List<Request> requests, Duration timeout) {
        // 1. 设置总体超时
        CompletableFuture<BatchResult> future = processBatchAsync(requests);
        try {
            return future.get(timeout.toMillis(), TimeUnit.MILLISECONDS);
        } catch (TimeoutException e) {
            // 2. 超时后降级为逐条处理
            List<Response> responses = new ArrayList<>();
            for (Request req : requests) {
                try {
                    // 设置更短的单个请求超时
                    responses.add(processSingleWithTimeout(req, timeout.dividedBy(requests.size())));
                } catch (Exception ex) {
                    responses.add(Response.failed(req.getId(), ex));
                }
            }
            return new BatchResult(responses, true); // 标记为降级处理
        }
    }
}

5.2 优先级调度

// 多级缓冲队列：高优先级立即处理，低优先级批量处理
public class PriorityBatchDispatcher {
    private Queue<Request> highPriorityQueue = new ConcurrentLinkedQueue<>();
    private Queue<Request> normalPriorityQueue = new ConcurrentLinkedQueue<>();
    
    public void dispatch(Request req) {
        if (req.isHighPriority()) {
            highPriorityQueue.offer(req);
            // 高优先级：立即处理或小批量处理
            if (highPriorityQueue.size() >= 5) {
                processHighPriorityBatch();
            }
        } else {
            normalPriorityQueue.offer(req);
            // 正常优先级：积累到100个或200ms处理
            if (normalPriorityQueue.size() >= 100) {
                processNormalBatch();
            }
        }
    }
}

5.3 流量自适应

// 根据系统负载动态调整批处理参数
public class AdaptiveBatchProcessor {
    private int currentBatchSize = 50;
    private long currentWindowMs = 100;
    private final LoadMonitor loadMonitor;
    
    void adjustParameters() {
        double cpuUsage = loadMonitor.getCpuUsage();
        double memoryUsage = loadMonitor.getMemoryUsage();
        
        if (cpuUsage > 0.8 || memoryUsage > 0.8) {
            // 系统负载高：减小批次，加快响应
            currentBatchSize = Math.max(10, currentBatchSize / 2);
            currentWindowMs = Math.max(20, currentWindowMs / 2);
        } else {
            // 系统负载低：增大批次，提升吞吐
            currentBatchSize = Math.min(500, currentBatchSize * 2);
            currentWindowMs = Math.min(1000, currentWindowMs * 2);
        }
    }
}

六、性能权衡与监控

6.1 需要权衡的维度

6.2 关键监控指标

# 批处理相关监控指标
batch_requests_total{type="incoming"}  # 总接收请求数
batch_executions_total                  # 批量执行次数
batch_size_avg                          # 平均每批大小
batch_processing_duration_seconds       # 批处理耗时
batch_latency_seconds{quantile="0.99"}  # P99延迟
batch_failure_rate                      # 批处理失败率
single_fallback_requests_total          # 降级为单条处理数

七、最佳实践总结

1. 渐进式实施：先对读接口实施批处理，再考虑写接口
2. 配置化控制：所有参数（窗口时间、批次大小）支持动态调整
3. 熔断降级：批处理失败时自动降级为单条处理
4. 压力测试：测试不同批次大小对P99延迟的影响
5. 业务隔离：不同业务使用独立的批处理器，避免相互影响
6. 客户端协作：鼓励客户端主动发送批量请求（如移动端聚合请求）

通过精心设计的请求合并与响应合并机制，可以在吞吐量提升2-10倍的同时，将数据库连接数、网络包数等资源消耗降低一个数量级，是高并发系统的必备优化手段。