问Kafka:如何在Kafka中实现轮询分区

EN

/**
 * The default partitioning strategy:
 * <ul>
 * <li>If a partition is specified in the record, use it
 * <li>If no partition is specified but a key is present choose a partition based on a hash of the key
 * <li>If no partition or key is present choose a partition in a round-robin fashion
 */

class RRPartitioner():
      def __init__():
            # Using topic metadata get total number of partitions
            self.total_partitions = client[topic].get_number_partitions()
            self.part_offset = 0

      def partitioner(self, key, msg):
          if self.part_offset > self.total_partitions:
              self.part_offset = 0
              return self.part_offset
          else:
              self.part_offset += 1
              return self.part_offset