Spring Integration压测：请求互扰、性能不达预期问题求助

问题根源分析 #

1. 请求互扰（响应内容混入） #

核心问题出在自定义反序列化器CustomDeserializer的逻辑缺陷：

• 依赖']'作为消息结束标记，但未处理网络粘包/拆包场景，若某次读取未完整获取到结束标记，剩余字节会被下一次反序列化读取，导致消息串扰。
• 循环终止条件loop < 300 && read != ']'存在漏洞：若读取到300字节仍未遇到']'，会直接截断返回，后续请求的响应内容会被混入当前结果。
• 未处理inputStream.read()返回-1（连接关闭）的情况，会将无效字符char(-1)加入结果，进一步破坏消息边界。

2. 性能无法达到1000TPS #

• 线程模型不匹配：当前使用DirectChannel，所有请求串行通过唯一的TcpOutboundGateway处理线程，无法并发利用连接池中的50个持久化连接。
• NIO处理能力不足：TcpNioClientConnectionFactory默认IO线程池大小有限，未显式配置专用线程池处理NIO读写，形成IO瓶颈。
• 同步调用阻塞：网关send方法为同步调用，若调用线程数不足，会限制并发处理能力，导致连接池资源闲置。

解决方案 #

一、修复请求互扰问题 #

重写反序列化器，确保严格处理消息边界：

@Component
class CustomDeserializer extends DefaultDeserializer {
    private static final int MAX_LENGTH = 300;
    private static final char END_MARKER = ']';

    @Override
    public Object deserialize(final InputStream inputStream) throws IOException {
        StringBuilder stringBuffer = new StringBuilder(MAX_LENGTH);
        int read;
        boolean foundEndMarker = false;

        while ((read = inputStream.read()) != -1) {
            char c = (char) read;
            stringBuffer.append(c);
            if (c == END_MARKER) {
                foundEndMarker = true;
                break;
            }
            if (stringBuffer.length() >= MAX_LENGTH) {
                throw new IOException("消息超过最大长度未找到结束标记");
            }
        }

        if (!foundEndMarker && read == -1) {
            throw new IOException("连接已关闭，未收到消息结束标记");
        }

        return stringBuffer.toString();
    }
}

关键改进：

• 循环直到找到结束标记或连接关闭，避免提前截断消息。
• 超过最大长度未找到标记时抛出异常，防止错误数据流入后续流程。
• 处理连接关闭场景，避免无效字符干扰。

二、优化性能达到1000TPS #

1. 替换为线程池消息通道
   将outboundChannel改为ExecutorChannel，用线程池并发处理请求，充分利用连接池资源：

@Bean
public MessageChannel outboundChannel(TaskExecutor taskExecutor) {
    return new ExecutorChannel(taskExecutor);
}

@Bean
public TaskExecutor taskExecutor() {
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    executor.setCorePoolSize(100); // 建议大于连接池大小，根据压测调整
    executor.setMaxPoolSize(200);
    executor.setQueueCapacity(500);
    executor.setThreadNamePrefix("tcp-outbound-");
    executor.initialize();
    return executor;
}

2. 配置NIO专用IO线程池
   为TcpNioClientConnectionFactory添加IO线程池，提升NIO处理效率：

private TcpNioClientConnectionFactory getTcpNioClientConnectionFactoryOf(
        final String ipAddress, final int port) {
    TcpNioClientConnectionFactory tcpNioClientConnectionFactory =
            new TcpNioClientConnectionFactory(ipAddress, port);
    // 保留原有配置...
    
    // 添加IO线程池
    ThreadPoolTaskExecutor ioExecutor = new ThreadPoolTaskExecutor();
    ioExecutor.setCorePoolSize(10);
    ioExecutor.setMaxPoolSize(20);
    ioExecutor.setThreadNamePrefix("tcp-nio-io-");
    ioExecutor.initialize();
    tcpNioClientConnectionFactory.setTaskExecutor(ioExecutor);
    
    return tcpNioClientConnectionFactory;
}

3. 优化网关超时与连接池配置

• 确认connectionProperty.getConnectionPoolSize()设置为50，保证连接池容量符合预期。
• 为TcpOutboundGateway设置合理超时，避免线程长时间阻塞：

@Bean
@ServiceActivator(inputChannel = "outboundChannel")
public MessageHandler outboundGateway(AbstractClientConnectionFactory clientConnectionFactory) {
    TcpOutboundGateway tcpOutboundGateway = new TcpOutboundGateway();
    tcpOutboundGateway.setConnectionFactory(clientConnectionFactory);
    tcpOutboundGateway.setReplyTimeout(5000); // 根据实际响应时间调整
    return tcpOutboundGateway;
}

验证建议 #

1. 先通过单连接发送多条消息，验证反序列化器修复后无消息串扰。
2. 逐步提升并发压测，监控TPS、连接池使用率、线程池负载，调整线程参数至最优。
3. 监控GKE Pod的CPU、内存使用率，排除Pod侧资源瓶颈。

内容的提问来源于stack exchange，提问作者Zahid Khan