Spring Integration TCP Server自定义反序列化器超时问题排查

问题描述 #

我搭建了一个TCP入站适配器，用于接收某硬件设备发送的自定义负载消息。由于没有适配该设备消息的默认反序列化器，我实现了自定义反序列化逻辑：当输入流有内容时读取数据，否则返回空字节数组；同时通过拦截器过滤空负载，仅允许非空负载进入后续处理环节。

但配置soTimeout=60000（1分钟超时）后，连接并未在指定超时时间后自动关闭，我怀疑问题出在自定义反序列化器将空输入流视为有效消息，导致超时机制未触发。

现有代码 #

服务器配置（Kotlin DSL） #

@Bean
fun tcpServer() = Tcp.netServer(port)
    .leaveOpen(true)
    .deserializer { inputStream ->
        when(inputStream.available()>0){
            true -> inputStream.readNBytes(inputStream.available())
            false -> byteArrayOf()
        }
    }
    .serializer(TcpCodecs.raw())
    .backlog(30)
    .interceptorFactoryChain(interceptor())
    .soTimeout(60000)

空负载过滤拦截器 #

class ValidationInterceptor(
    applicationEventPublisher: ApplicationEventPublisher
) : TcpConnectionInterceptorSupport(applicationEventPublisher) {
    override fun onMessage(message: Message<*>): Boolean {
        if ((message.payload as ByteArray).isEmpty()) return false
        return super.onMessage(message)
    }
}

问题根源 #

你的猜测是对的：soTimeout的作用是控制Socket读取操作的阻塞超时，但当前自定义反序列化器的逻辑导致读取操作根本不会阻塞等待数据。

当输入流没有数据时，inputStream.available()返回0，你直接返回空字节数组，这会让读取操作立刻完成，框架会认为已经处理完一条消息，然后立刻进入下一次读取循环——相当于不断返回空消息，永远不会触发超时等待，连接也就不会关闭。

解决方案 #

1. 修正反序列化器逻辑 #

去掉available()的判断，改为尝试读取数据，让读取操作自然阻塞直到超时（触发SocketTimeoutException），此时框架会自动关闭连接。示例修改如下：

.deserializer { inputStream ->
    val buffer = ByteArray(1024) // 根据设备消息大小调整缓冲区
    try {
        val bytesRead = inputStream.read(buffer)
        if (bytesRead == -1) {
            // 输入流已关闭，抛出异常触发连接关闭
            throw IOException("Client closed connection")
        }
        buffer.copyOf(bytesRead)
    } catch (e: SocketTimeoutException) {
        // 超时异常直接抛出，框架会关闭连接
        throw e
    }
}

2. 调整leaveOpen配置 #

将leaveOpen(true)改为leaveOpen(false)，这样当读取超时、流关闭或出现异常时，框架会自动关闭连接，无需手动处理。

3. 保留拦截器逻辑 #

拦截器的空负载过滤逻辑可以保留，它会在反序列化得到非空数据后，过滤掉可能的空消息（比如设备异常发送的空包），不影响超时机制的触发。

内容的提问来源于stack exchange，提问作者jorge-plaza