如何通过Bash脚本为Kafka创建多个消费者

前提准备 #

确保你的Java Kafka消费者已经打包成可执行Jar包，且支持通过命令行参数或系统属性加载配置（比如覆盖消费组ID、客户端ID等）。

方案1：启动同组消费者（分区负载均衡） #

如果需要多个消费者同属一个消费组，共同分担Topic的分区消费，用以下脚本：

#!/bin/bash

# 核心配置
KAFKA_TOPIC="your-target-topic"
CONSUMER_GROUP="your-group-id"
JAR_PATH="/path/to/your/consumer-app.jar"
NUM_CONSUMERS=3  # 要启动的消费者数量

# 循环启动实例
for i in $(seq 1 $NUM_CONSUMERS); do
    # 每个实例指定唯一客户端ID，避免集群内标识冲突
    java -Dkafka.consumer.client.id="consumer-instance-$i" \
         -Dkafka.consumer.group.id="$CONSUMER_GROUP" \
         -Dkafka.consumer.topic="$KAFKA_TOPIC" \
         -jar "$JAR_PATH" &
    echo "已启动消费者实例 $i，进程ID: $!"
done

# 可选：保持脚本运行，直到所有消费者进程结束
wait

方案2：启动不同组消费者（全量消费） #

如果每个消费者需要独立消费Topic的全部消息，给每个实例分配唯一的消费组ID即可：

#!/bin/bash

KAFKA_TOPIC="your-target-topic"
JAR_PATH="/path/to/your/consumer-app.jar"
NUM_CONSUMERS=3

for i in $(seq 1 $NUM_CONSUMERS); do
    java -Dkafka.consumer.client.id="consumer-instance-$i" \
         -Dkafka.consumer.group.id="$CONSUMER_GROUP-$i" \
         -Dkafka.consumer.topic="$KAFKA_TOPIC" \
         -jar "$JAR_PATH" &
    echo "已启动独立消费组的消费者实例 $i，进程ID: $!"
done

wait

关键补充 #

• 你的Java消费者代码需要兼容读取系统属性，示例代码片段：

  Properties props = new Properties();
  // 优先读取系统属性，无则用默认值
  props.put(ConsumerConfig.GROUP_ID_CONFIG, System.getProperty("kafka.consumer.group.id", "default-group"));
  props.put(ConsumerConfig.CLIENT_ID_CONFIG, System.getProperty("kafka.consumer.client.id", "default-client"));
  props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, System.getProperty("kafka.consumer.bootstrap.servers", "localhost:9092"));
  // 其他Kafka配置...
  

• 如果使用外部配置文件（如consumer.properties），可以在脚本中复制并修改配置后启动：

  cp consumer.properties consumer-instance-$i.properties
  sed -i "s/group.id=.*/group.id=$CONSUMER_GROUP-$i/" consumer-instance-$i.properties
  java -jar $JAR_PATH --config consumer-instance-$i.properties &
  

• 可以添加日志拆分逻辑，避免日志混叠：

  java -Dkafka.consumer.client.id="consumer-instance-$i" \
       -jar "$JAR_PATH" > consumer-instance-$i.log 2>&1 &
  

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