Apache Flink消费RabbitMQ消息时如何处理毒消息避免作业失败

问题根源 #

你当前的作业崩溃是因为反序列化逻辑未捕获Json解析异常，且Flink原生RMQSource默认在反序列化失败时会直接抛出异常，未被确认的毒消息会反复回队列投递，导致作业持续重启。

解决方案 #

1. 修改反序列化逻辑，捕获解析异常

首先调整反序列化实现，避免解析错误直接抛出异常，将正常消息和异常消息封装为不同的结构返回：

// 定义消息结果封装类，区分合法/非法消息
sealed trait MessageResult
case class ValidMessage(data: Test) extends MessageResult
case class InvalidMessage(rawBytes: Array[Byte], error: Throwable) extends MessageResult

class EventDeserializationSchema extends DeserializationSchema[MessageResult] {
  @throws(classOf[IOException])
  override def deserialize(message: Array[Byte]): MessageResult = {
    try {
      ValidMessage(EventDeserializationSchema.objectMapper.readValue(message, classOf[Test]))
    } catch {
      case e: Exception => InvalidMessage(message, e)
    }
  }

  override def isEndOfStream(nextElement: MessageResult): Boolean = false
  override def getProducedType: TypeInformation[MessageResult] = createTypeInformation[MessageResult]
}

object EventDeserializationSchema{
  val objectMapper: ObjectMapper = new ObjectMapper().registerModule(DefaultScalaModule)
}

2. 自定义RMQSource，开启手动确认对接死信交换机

关闭RMQ的自动ack机制，重写消息处理逻辑，对不同类型的消息做不同的确认处理：

class CustomRMQSource(
  rmqConfig: RMQConnectionConfig,
  queueName: String,
  deserializer: EventDeserializationSchema
) extends RMQSource[MessageResult](rmqConfig, queueName, false, deserializer) {

  override def processMessage(delivery: Delivery): Unit = {
    deserializer.deserialize(delivery.getBody) match {
      case ValidMessage(data) =>
        // 合法消息手动确认后发送到下游
        channel.basicAck(delivery.getEnvelope.getDeliveryTag, false)
        collect(data)
      case InvalidMessage(raw, err) =>
        // 毒消息手动否定确认，设置不重新入队，RabbitMQ会自动路由到预先配置的死信交换机
        channel.basicNack(delivery.getEnvelope.getDeliveryTag, false, false)
        // 可选：打印错误日志方便后续排查
        log.error(s"消息解析失败，原始内容：${new String(raw)}，错误信息：${err.getMessage}")
    }
  }
}

3. 可选无死信交换机方案

如果不需要对接RabbitMQ死信交换机，也可以直接确认毒消息后通过Flink侧输出流将非法消息输出到其他存储（如日志系统、ES）后续排查，无需修改RabbitMQ配置，同样可以保证作业持续运行。

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