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
}

class 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
  }
}