一文教你如何监控Kafka Topic的生产者客户端

# 下载 Kafka（以 3.7.0 为例）
wget https://downloads.apache.org/kafka/3.7.0/kafka_2.13-3.7.0.tgz
tar -xzf kafka_2.13-3.7.0.tgz
cd kafka_2.13-3.7.0

# 确保 Java 已安装（Kafka 依赖 Java 运行）
sudo apt install openjdk-17-jdk  # Ubuntu/Debian
sudo yum install java-17-openjdk-devel  # CentOS/RHEL

# 验证安装
bin/kafka-topics.sh --version

# 列出所有消费者组（部分生产者信息可能在此显示）
bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list

# 获取 Topic 的写入偏移量（间接判断生产者活跃情况）
bin/kafka-run-class.sh kafka.tools.GetOffsetShell \
  --broker-list localhost:9092 \
  --topic my-topic \
  --time -1

# 启动 Kafka 时启用 JMX
export JMX_PORT=9999
bin/kafka-server-start.sh config/server.properties &

# 使用 jcmd 查看指标（需 Java 11+）
jcmd <Kafka_PID> PerfCounter.print | grep Producer

# 查看最近的生产者连接
grep "ProducerId" logs/server.log

# 按客户端 IP 过滤
grep "Accepted connection from" logs/server.log | awk '{print $NF}'

import org.apache.kafka.clients.admin.*;

public class KafkaProducerMonitor {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        
        try (AdminClient admin = AdminClient.create(props)) {
            ListConsumerGroupsResult groups = admin.listConsumerGroups();
            groups.all().get().forEach(System.out::println);
        }
    }
}

# 使用 tcpdump 抓取 Kafka 流量（9092 端口）
sudo tcpdump -i eth0 port 9092 -A | grep "PRODUCE"