Kafka Producer机制优化-提高发送消息可靠性

名称解释：

Broker：负责消息的存储和转发，也可以叫消息中介节点 
Topic：每种消息的分类叫做主题(Topic)。 
Partition：每一个Topic被切分为多个Partitions。

背景

Producer构造Message对象时，传入key参数，当Producer发送Message，会根据key确定目标Partition，当Kafka集群中某个Partition所有存活的节点都失效或挂掉。会造成Producer尝试重新发送message.send.max.retries（默认值为3）次后抛出Exception，每次尝试都会休眠一定时间(默认值为100ms)。用户捕捉到异常其结果，停止发送会阻塞，继续发送消息会丢失。

解决思路

Kafka中默认发送消息方式不变，给用户提供一种可选择方式，增加一个消息发送失效转移的开关，当Producer发送到目标Partition的(所有副本本来存活的)节点都失效或挂掉，就转发到其他Partition上。 
M：表示Partition的主分区。 
S：表示Partition的从分区。下面图2所示，消息轮询发送到Partition的0-3上。 

图1 
上图本来一条消息是发送到Partition-0所在Broker的，当Partition-0所在Broker全部失效或挂掉后，如下图所示。 
 
图2 
消息(轮询均衡)发送到其他Partition上，待失效的Brokers恢复，发送消息恢复到图1状态。

Producer代码修改

代码修改如下： 
ProducerConfig.Scala类代码中增加一行： 
val sendSwitchBrokerEnabled = props.getBoolean(“send.switch.broker.enable”, false) 
DefaultEventHandler.scala类增加一行 
val isSendSwitchBrokerEnabled = config.sendSwitchBrokerEnabled

DefaultEventHandler.scala类中增加如下方法：

 

private def getPartition(topic: String, key: Any, topicPartitionList: Seq[PartitionAndLeader], isSendSwitchBrokerEnabled: Boolean = false): Int = {
  val numPartitions = topicPartitionList.size
  if(numPartitions <= 0)
    throw new UnknownTopicOrPartitionException("Topic " + topic + " doesn't exist")
  var partition =
    if (key == null) {
      // If the key is null, we don't really need a partitioner
      // So we look up in the send partition cache for the topic to decide the target partition
      val id = sendPartitionPerTopicCache.get(topic)
      id match {
        case Some(partitionId) =>
          // directly return the partitionId without checking availability of the leader,
          // since we want to postpone the failure until the send operation anyways
          partitionId
        case None =>
          val availablePartitions = topicPartitionList.filter(_.leaderBrokerIdOpt.isDefined)
          if (availablePartitions.isEmpty)
            throw new LeaderNotAvailableException("No leader for any partition in topic " + topic)
          val index = Utils.abs(Random.nextInt) % availablePartitions.size
          val partitionId = availablePartitions(index).partitionId
          sendPartitionPerTopicCache.put(topic, partitionId)
          partitionId
      }
    } else
    partitioner.partition(key, numPartitions)

  if(partition < 0 || partition >= numPartitions)
    throw new UnknownTopicOrPartitionException("Invalid partition id: " + partition + " for topic " + topic +
      "; Valid values are in the inclusive range of [0, " + (numPartitions-1) + "]")
  trace("Assigning message of topic %s and key %s to a selected partition %d".format(topic, if (key == null) "[none]" else key.toString, partition))

  if (isSendSwitchBrokerEnabled) {
        if(partitionsLeaderInTopicsCache.isEmpty()) {
          val availablePartitions = topicPartitionList.map { partitionAndLeader =>
              if(partitionAndLeader.leaderBrokerIdOpt.isDefined) {
                partitionsLeaderInTopicsCache.put(TopicAndPartition(topic, partitionAndLeader.partitionId),
                    partitionAndLeader.leaderBrokerIdOpt.get)
              }
          }
        }

        if(!partitionsLeaderInTopicsCache.containsKey(TopicAndPartition(topic, partition))) {
            val availablePartitions = topicPartitionList.filter(_.leaderBrokerIdOpt.isDefined)
            if (availablePartitions.isEmpty)
              throw new LeaderNotAvailableException("No leader for any partition in topic " + topic)
            val index = Utils.abs(Random.nextInt) % availablePartitions.size
            partition = availablePartitions(index).partitionId
        }
  }
  partition
}