Akka Stream Kafka WebSocket客户端30秒无消息后停止接收问题求助

嘿，这个问题我之前在做Akka Stream + Kafka + WebSocket的项目时也碰到过！核心就是空闲超时在搞鬼——不管是WebSocket的连接空闲，还是Kafka消费者的会话超时，都会导致30秒没消息后客户端断连。给你几个亲测有效的解决方案：

Akka HTTP的WebSocket默认自带30秒左右的空闲超时，只要这段时间没有数据传输，连接就会被主动关闭。你有两个选择：

• 直接延长超时时间：在WebSocket路由里配置更长的空闲超时，比如5分钟：

import akka.http.scaladsl.model.ws.Message
import akka.http.scaladsl.server.Directives._
import scala.concurrent.duration._

val wsRoute = path("kafka-ws") {
  handleWebSocketMessages(
    kafkaSourceToWebSocketFlow
      .withIdleTimeout(300.seconds) // 按需调整超时时长
  )
}

• 定期发送Ping/Pong帧保活：即使没有Kafka消息，也定时发个Ping帧让连接保持活跃，避免被判定为空闲：

import akka.stream.scaladsl.{Flow, Source}
import scala.concurrent.duration._

// 构建一个添加Ping帧的Flow
val keepAliveFlow: Flow[Message, Message, _] = Flow[Message]
  .merge(Source.tick(25.seconds, 25.seconds, Message.Ping))
  .map {
    case ping: Message.Ping => ping
    case msg => msg
  }

// 在路由里用这个Flow包装你的Kafka消息流
handleWebSocketMessages(kafkaSourceToWebSocketFlow.via(keepAliveFlow))

从你提到的日志来看，Kafka消费者大概率因为长时间没消息触发了会话超时，被集群判定为“死亡”，导致消息流中断。你需要在application.conf里调整消费者的核心参数：

akka.kafka.consumer {
  kafka-clients {
    session.timeout.ms = 300000  // 改成5分钟（默认30秒）
    heartbeat.interval.ms = 100000 // 心跳间隔建议是会话超时的1/3
    max.poll.interval.ms = 600000 // 最大轮询间隔，避免消费者被踢
  }
}

这些参数能让消费者在长时间无消息时，依然和Kafka集群保持会话，不会被踢出消费组，这样消息流就能一直等着新消息过来。

有时候Kafka源因为没消息进入“空闲”状态，Akka Stream内部可能会触发终止逻辑。你可以给Kafka源加个keepAlive操作，确保流不会因为空闲而挂掉：

import akka.stream.scaladsl.Source
import scala.concurrent.duration._

// 自定义一个占位消息（也可以用Option类型过滤掉）
case object PlaceholderMessage

// 给Kafka源添加keepAlive，空闲25秒就发个占位消息
val kafkaSourceWithKeepAlive = kafkaConsumerSource
  .keepAlive(25.seconds, () => Some(PlaceholderMessage))
  .filter(_ != Some(PlaceholderMessage)) // 不需要占位消息就过滤掉
  .map(_.get)

这样哪怕Kafka一直没消息，流也会持续运行，不会因为空闲而断开WebSocket连接。

验证顺序建议 #

先调WebSocket的空闲超时（这是客户端断连的直接原因），再检查Kafka消费者配置，最后加流的keepAlive做双重保障，基本就能解决问题了！

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