centos Kafka消费者如何设

在 CentOS 上设置 Kafka 消费者，通常分为 命令行方式 和 代码方式（Java/Python 等）。下面给你一个从基础到常用的完整说明。

一、前提条件（CentOS 环境）

1. 已安装 Kafka（假设路径：/opt/kafka）
2. Kafka 服务已启动
3. 知道：
   • Kafka 地址（如 localhost:9092）
   • Topic 名称（如 test-topic）

启动 Kafka（示例）：

# 启动 Zookeeper（老版本）
/opt/kafka/bin/zookeeper-server-start.sh /opt/kafka/config/zookeeper.properties

# 启动 Kafka
/opt/kafka/bin/kafka-server-start.sh /opt/kafka/config/server.properties

二、命令行消费者（最常用）

1️⃣ 从最新位置消费

/opt/kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 \
  --topic test-topic

2️⃣ 从最早消息开始消费

kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 \
  --topic test-topic \
  --from-beginning

3️⃣ 指定消费者组

kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 \
  --topic test-topic \
  --group my-consumer-group

4️⃣ 查看消费者组状态

kafka-consumer-groups.sh \
  --bootstrap-server localhost:9092 \
  --describe \
  --group my-consumer-group

三、消费者核心参数说明（重要）

示例：

kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 \
  --topic test-topic \
  --partition 0 \
  --offset 10

四、Java 消费者示例（生产环境常用）

1️⃣ Maven 依赖

<dependency>
  <groupId>org.apache.kafka</groupId>
  <artifactId>kafka-clients</artifactId>
  <version>3.6.0</version>
</dependency>

2️⃣ Java 消费者代码

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "my-group");
props.put("enable.auto.commit", "true");
props.put("auto.commit.interval.ms", "1000");
props.put("key.deserializer",
        "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer",
        "org.apache.kafka.common.serialization.StringDeserializer");

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Collections.singletonList("test-topic"));

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        System.out.printf("offset=%d, key=%s, value=%s%n",
                record.offset(), record.key(), record.value());
    }
}

五、Python 消费者（可选）

pip install kafka-python

from kafka import KafkaConsumer

consumer = KafkaConsumer(
    'test-topic',
    bootstrap_servers='localhost:9092',
    group_id='my-group',
    auto_offset_reset='earliest'
)

for msg in consumer:
    print(msg.value.decode())

六、CentOS 常见问题排查

1️⃣ 连不上 Kafka

• 检查防火墙

firewall-cmd --list-all

• 检查 server.properties

listeners=PLAINTEXT://0.0.0.0:9092
advertised.listeners=PLAINTEXT://<服务器IP>:9092

2️⃣ 消费不到消息

• 是否用了消费者组（offset 已提交）
• 是否应该加 --from-beginning

如果你能告诉我：

• Kafka 版本
• 是否用消费者组
• 用命令行还是代码

我可以给你更精确的配置示例。