Akka WebSocket Client转Kafka Producer：交易数据入Kafka问题求助

嘿，看起来你已经搞定了最基础的部分——测试字符串能正常发Kafka，说明你的Producer配置、ActorSystem这些基础环境都没问题！问题大概率出在交易数据流和Producer的对接逻辑上，咱们一步步来调整：

核心思路：替换控制台Sink为Kafka Producer Sink #

你现在应该是用类似 .runWith(Sink.foreach(println)) 的代码把交易数据打去控制台，咱们要做的就是把这个控制台Sink替换成Reactive Kafka提供的ProducerSink，关键是要把你的交易数据转换成Kafka能识别的ProducerRecord。

步骤1：确认ProducerSettings配置正确 #

先检查你的ProducerSettings是不是包含了正确的序列化器和Kafka地址（你已经用了9092端口，这里重点看序列化器）：

import akka.kafka.ProducerSettings
import org.apache.kafka.clients.producer.ProducerRecord
import org.apache.kafka.common.serialization.StringSerializer
// 假设你的ActorSystem已经定义为system
val producerSettings = ProducerSettings(system, new StringSerializer, new StringSerializer)
  .withBootstrapServers("localhost:9092")

如果你的交易数据不是字符串（比如是自定义的TradeData类），要把值的序列化器换成对应的自定义序列化器（比如JSON序列化器，常用Circe、Play Json来实现）。

步骤2：转换交易数据为ProducerRecord #

把你的交易数据流（假设是tradeStream: Source[TradeData, _]）转换成Kafka需要的ProducerRecord格式：

// 假设TradeData是你的交易数据模型，先把它序列化成JSON字符串
val tradeKafkaStream = tradeStream.map { trade =>
  // 第一个参数是你的Kafka主题名，替换成实际名称
  // 这里用trade.id作为key，trade.toJson作为value，可根据需求调整
  new ProducerRecord[String, String]("stock-trades-topic", trade.id.toString, trade.toJson)
}

这里的trade.toJson需要你自己实现，比如用Circe的话：

import io.circe.generic.auto._
import io.circe.syntax._
// 给TradeData添加toJson方法
def toJson: String = this.asJson.noSpaces

步骤3：对接ProducerSink发送数据 #

最后把转换后的数据流和ProducerSink连接起来，替换原来的控制台输出：

import akka.kafka.scaladsl.Producer

// 用plainSink直接发送，适合不需要处理发送结果的场景
tradeKafkaStream.runWith(Producer.plainSink(producerSettings))

进阶：添加错误处理（方便调试） #

默认的plainSink不会返回发送结果，如果想排查发送失败的问题，可以用flexiFlow来获取发送状态：

import akka.kafka.ProducerMessage

tradeKafkaStream
  .via(Producer.flexiFlow(producerSettings))
  .runWith(Sink.foreach { result =>
    result match {
      case ProducerMessage.Result(record, metadata) =>
        println(s"成功发送交易数据 ${record.key()} 到分区 ${metadata.partition()}，偏移量 ${metadata.offset()}")
      case ProducerMessage.Failure(record, exception) =>
        println(s"发送交易数据 ${record.key()} 失败：${exception.getMessage}")
    }
  })

常见排查点 #

• 序列化错误：如果发送失败，先在map里打印转换后的ProducerRecord，确认内容是合法的字符串/JSON
• Materializer缺失：确保你的代码里有正确的ActorMaterializer隐式实例：implicit val mat: ActorMaterializer = ActorMaterializer()(system)
• 主题不存在：Kafka默认不会自动创建主题的话，要先手动创建你的目标主题

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