kafka消费者分组消费的再平衡策略

Map<String, Integer> topicCountMap = new HashMap<String, Integer>();topicCountMap.put(topic, new Integer(1));Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = consumer.createMessageStreams(topicCountMap);

val consumerIdString = {  var consumerUuid : String = null  config.consumerId match {    case Some(consumerId) // for testing only    => consumerUuid = consumerId    case None // generate unique consumerId automatically    => val uuid = UUID.randomUUID()    consumerUuid = "%s-%d-%s".format(      InetAddress.getLocalHost.getHostName, System.currentTimeMillis,      uuid.getMostSignificantBits().toHexString.substring(0,8))  }  config.groupId + "_" + consumerUuid}

for ((topic, nConsumers) <- topicCountMap) {  val consumerSet = new mutable.HashSet[ConsumerThreadId]  assert(nConsumers >= 1)  for (i <- 0 until nConsumers)    consumerSet += ConsumerThreadId(consumerIdString, i) //ConusmerId的结尾加上一个常量区别 owners 目录下可以看到  consumerThreadIdsPerTopicMap.put(topic, consumerSet)}

"%s-%d".format(consumer, threadId)

TopicAndPartition(topic: String, partition: Int)

override def toString = "[%s,%d]".format(topic, partition)

class AssignmentContext(group: String, val consumerId: String, excludeInternalTopics: Boolean, zkClient: ZkClient) {  //(topic,ConsumerThreadIdSet) //指定一个消费者，根据每个topic指定的streams数目，构建相同数目的流  val myTopicThreadIds: collection.Map[String, collection.Set[ConsumerThreadId]] = {    val myTopicCount = TopicCount.constructTopicCount(group, consumerId, zkClient, excludeInternalTopics)    myTopicCount.getConsumerThreadIdsPerTopic  }
  //(topic 分区) /brokers/topics/topicname 值  val partitionsForTopic: collection.Map[String, Seq[Int]] =    ZkUtils.getPartitionsForTopics(zkClient, myTopicThreadIds.keySet.toSeq)
  //(topic,ConsumerThreadId(当前消费者id)) ///consumers/Groupid/ids 子节点  val consumersForTopic: collection.Map[String, List[ConsumerThreadId]] =    ZkUtils.getConsumersPerTopic(zkClient, group, excludeInternalTopics)
  ///consumers/Groupid/ids的所有的子节点  val consumers: Seq[String] = ZkUtils.getConsumersInGroup(zkClient, group).sorted} 2，RangeAssignorclass RangeAssignor() extends PartitionAssignor with Logging {
  def assign(ctx: AssignmentContext) = {    val partitionOwnershipDecision = collection.mutable.Map[TopicAndPartition, ConsumerThreadId]()
    for ((topic, consumerThreadIdSet) <- ctx.myTopicThreadIds) {      val curConsumers = ctx.consumersForTopic(topic) //当前topic的所有消费者      val curPartitions: Seq[Int] = ctx.partitionsForTopic(topic) //当前topic的所有分区
      //      val nPartsPerConsumer = curPartitions.size / curConsumers.size      val nConsumersWithExtraPart = curPartitions.size % curConsumers.size
      info("Consumer " + ctx.consumerId + " rebalancing the following partitions: " + curPartitions +        " for topic " + topic + " with consumers: " + curConsumers)
      for (consumerThreadId <- consumerThreadIdSet) {        val myConsumerPosition = curConsumers.indexOf(consumerThreadId) //获取当前消费者线程的在集合中的位置        assert(myConsumerPosition >= 0)        //获取分区的起始位置        val startPart = nPartsPerConsumer * myConsumerPosition + myConsumerPosition.min(nConsumersWithExtraPart)        //获取当前消费者消费的分区数目        val nParts = nPartsPerConsumer + (if (myConsumerPosition + 1 > nConsumersWithExtraPart) 0 else 1)
        /**         *   Range-partition the sorted partitions to consumers for better locality.         *  The first few consumers pick up an extra partition, if any.         */        if (nParts <= 0)          warn("No broker partitions consumed by consumer thread " + consumerThreadId + " for topic " + topic)        else {          //将分区关系描述写入partitionOwnershipDecision          for (i <- startPart until startPart + nParts) {            val partition = curPartitions(i)            info(consumerThreadId + " attempting to claim partition " + partition)            // record the partition ownership decision            partitionOwnershipDecision += (TopicAndPartition(topic, partition) -> consumerThreadId)          }        }      }    }
    partitionOwnershipDecision  }} 

class RoundRobinAssignor() extends PartitionAssignor with Logging {
  def assign(ctx: AssignmentContext) = {    val partitionOwnershipDecision = collection.mutable.Map[TopicAndPartition, ConsumerThreadId]()
    // check conditions (a) and (b)    topic, List[ConsumerThreadId]    val (headTopic, headThreadIdSet) = (ctx.consumersForTopic.head._1, ctx.consumersForTopic.head._2.toSet)
    //测试输出    ctx.consumersForTopic.foreach { case (topic, threadIds) =>      val threadIdSet = threadIds.toSet      require(threadIdSet == headThreadIdSet,              "Round-robin assignment is allowed only if all consumers in the group subscribe to the same topics, " +              "AND if the stream counts across topics are identical for a given consumer instance.\n" +              "Topic %s has the following available consumer streams: %s\n".format(topic, threadIdSet) +              "Topic %s has the following available consumer streams: %s\n".format(headTopic, headThreadIdSet))    }    //为传入的集合创建一个循环迭代器,传入之前排序是按照字典排序    val threadAssignor = Utils.circularIterator(headThreadIdSet.toSeq.sorted)
    info("Starting round-robin assignment with consumers " + ctx.consumers)
    //TopicAndPartition 按照字符串的hash排序    val allTopicPartitions = ctx.partitionsForTopic.flatMap { case(topic, partitions) =>      info("Consumer %s rebalancing the following partitions for topic %s: %s"           .format(ctx.consumerId, topic, partitions))
      partitions.map(partition => {        TopicAndPartition(topic, partition)  //toString = "[%s,%d]".format(topic, partition)      })
    }.toSeq.sortWith((topicPartition1, topicPartition2) => {      /*       * Randomize the order by taking the hashcode to reduce the likelihood of all partitions of a given topic ending       * up on one consumer (if it has a high enough stream count).       */      //按照hash值进行排序      topicPartition1.toString.hashCode < topicPartition2.toString.hashCode    })
    //过滤得到当前消费者的线程id    allTopicPartitions.foreach(topicPartition => {      val threadId = threadAssignor.next()      if (threadId.consumer == ctx.consumerId)        partitionOwnershipDecision += (topicPartition -> threadId)    })
    //返回得到结果    partitionOwnershipDecision  }}