如何用Apache Kafka实现多客户数据流并行生产与消费（Tomcat集群）

核心设计思路 #

要实现每个客户的10条记录被两台消费者Tomcat并行消费，关键是利用Kafka的分区策略与消费者组特性，通过自定义分区逻辑让同一客户的记录分散到不同分区，从而实现并行处理。

具体实现步骤 #

1. Kafka主题配置 #

• 创建主题时，将分区数设为2（与消费者节点数匹配，最大化并行度）
• 副本数根据集群稳定性需求设置，建议至少2个
• 命令示例：

  kafka-topics.sh --create --topic customer_records --bootstrap-server <kafka_broker>:9092 --partitions 2 --replication-factor 2
  

2. 生产者端（Tomcat）配置 #

• 自定义分区器：结合客户ID和**记录序号（1-10）**计算分区，确保同一客户的10条记录均匀分配到2个分区：

  public class CustomerRecordPartitioner implements Partitioner {
      @Override
      public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
          List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
          int numPartitions = partitions.size();
          // key为客户ID，value包含该客户的记录序号
          String customerId = (String) key;
          CustomerRecord record = (CustomerRecord) value;
          // 通过异或运算混合客户ID哈希与记录序号，避免同客户记录集中在单一分区
          int hash = customerId.hashCode() ^ record.getSeq();
          return Math.abs(hash) % numPartitions;
      }
  
      // 省略configure、close等接口实现
  }
  

• 在生产者配置中指定该分区器：

  props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, "com.yourpackage.CustomerRecordPartitioner");
  props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "<kafka_broker>:9092");
  props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
  props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, CustomerRecordSerializer.class.getName());
  

3. 消费者端（两台Tomcat）配置 #

• 两台Tomcat的消费者加入同一个消费者组（如customer-record-consumers），Kafka会自动将2个分区分配给两台消费者（每台1个分区）
• 配置max.poll.records控制单次拉取的记录数，根据业务处理能力调整
• 配置示例：

  props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "<kafka_broker>:9092");
  props.put(ConsumerConfig.GROUP_ID_CONFIG, "customer-record-consumers");
  props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
  props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, CustomerRecordDeserializer.class.getName());
  props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "5"); // 适配10条记录分两次拉取处理
  

关键注意事项 #

• 验证分区逻辑：测试同一客户的10条记录是否均匀分布在2个分区，避免出现集中分配导致单消费者处理的情况
• 消费者数量不要超过主题分区数，多余的消费者会处于空闲状态
• 若后续需要扩展消费者节点，只需增加主题分区数（分区数仅支持增加，不支持减少）

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