企业级K8s集群部署Kafka消息系统完整教程与最佳实践助您轻松构建高可用大数据处理平台实现业务快速增长和技术突破

引言

在当今大数据时代，构建可靠、可扩展的消息系统已成为企业技术架构的核心组成部分。Apache Kafka作为分布式流处理平台，以其高吞吐量、持久化、分布式特性受到广泛关注。而Kubernetes作为容器编排的事实标准，为Kafka的部署和管理提供了强大的支持。本文将详细介绍如何在企业级Kubernetes集群上部署Kafka消息系统，帮助您构建高可用的大数据处理平台，实现业务的快速增长和技术突破。

Kafka与Kubernetes概述

Apache Kafka简介

Apache Kafka是一个开源的分布式事件流处理平台，最初由LinkedIn开发，现在是Apache软件基金会的一部分。Kafka具有以下核心特性：

高吞吐量：Kafka能够处理每秒数百万条消息
持久化：消息被持久化到磁盘，支持数据回溯
分布式：支持集群部署，提供高可用性和可扩展性
实时处理：支持实时数据流处理

Kubernetes简介

Kubernetes（简称K8s）是一个开源的容器编排平台，用于自动化容器化应用程序的部署、扩展和管理。Kubernetes提供了以下核心功能：

服务发现和负载均衡
存储编排
自动化部署和回滚
自动装箱
自我修复
密钥和配置管理

在Kubernetes上部署Kafka的优势

在Kubernetes上部署Kafka具有以下优势：

简化部署和管理：通过Kubernetes的声明式配置，简化了Kafka集群的部署和管理
弹性伸缩：根据负载自动调整Kafka集群规模
高可用性：利用Kubernetes的自我修复能力，确保Kafka集群的高可用性
资源优化：通过资源限制和请求，优化集群资源使用
运维效率：统一的运维界面和工具链，降低运维复杂度

部署前准备工作

环境要求

在开始部署之前，确保满足以下环境要求：

Kubernetes集群：版本1.19或更高
kubectl：配置好与集群连接的kubectl工具
Helm：版本3.0或更高，用于管理Kafka应用
存储类：配置好持久化存储类（如NFS、Ceph、云存储等）
足够的资源：根据业务需求准备足够的CPU和内存资源

知识储备

为了顺利完成部署，建议具备以下知识：

Kubernetes基础概念（Pod、Service、Deployment、StatefulSet等）
Kafka基本架构和原理
YAML配置文件编写
Helm基本使用方法

使用Helm部署Kafka

安装Helm

如果尚未安装Helm，可以通过以下命令安装：

# 下载Helm安装脚本 curl -fsSL -o get_helm.sh https://raw.githubusercontent.com/helm/helm/main/scripts/get-helm-3 # 执行安装脚本 chmod 700 get_helm.sh ./get_helm.sh # 验证安装 helm version

添加Kafka Helm仓库

我们将使用Bitnami的Kafka Helm chart，这是一个经过充分测试和维护的Kafka部署方案。

# 添加Bitnami仓库 helm repo add bitnami https://charts.bitnami.com/bitnami # 更新仓库 helm repo update

创建Kafka命名空间

为了更好地管理Kafka相关资源，我们创建一个专门的命名空间：

kubectl create namespace kafka

自定义Kafka配置

创建一个自定义的Kafka配置文件kafka-values.yaml，根据企业需求进行配置：

# kafka-values.yaml # 全局配置 global: # 设置镜像仓库 imageRegistry: "" # 设置镜像拉取策略 imagePullPolicy: IfNotPresent # Kafka配置 kafka: # 副本数量，根据集群规模调整 replicaCount: 3 # Kafka配置 configuration: # 设置日志保留时间 log.retention.hours: 168 # 设置日志段大小 log.segment.bytes: 1073741824 # 设置默认分区数 num.partitions: 3 # 设置默认副本因子 default.replication.factor: 3 # 设置最小同步副本数 min.insync.replicas: 2 # 启用删除主题 delete.topic.enable: true # 设置消息最大字节数 message.max.bytes: 10485760 # 设置副本获取最大字节数 replica.fetch.max.bytes: 10485760 # 资源限制 resources: requests: memory: "2Gi" cpu: "1000m" limits: memory: "4Gi" cpu: "2000m" # 持久化存储配置 persistence: # 启用持久化存储 enabled: true # 存储类名称 storageClass: "fast-ssd" # 存储大小 size: "100Gi" # 服务配置 service: # 设置服务类型 type: LoadBalancer # 设置端口 ports: client: 9092 external: 9094 # Pod反亲和性配置，确保Pod分布在不同节点 affinity: podAntiAffinity: requiredDuringSchedulingIgnoredDuringExecution: - labelSelector: matchExpressions: - key: app.kubernetes.io/component operator: In values: - kafka topologyKey: "kubernetes.io/hostname" # ZooKeeper配置 zookeeper: # 副本数量，建议为奇数（3、5等） replicaCount: 3 # 资源限制 resources: requests: memory: "1Gi" cpu: "500m" limits: memory: "2Gi" cpu: "1000m" # 持久化存储配置 persistence: enabled: true storageClass: "fast-ssd" size: "20Gi" # Pod反亲和性配置 affinity: podAntiAffinity: requiredDuringSchedulingIgnoredDuringExecution: - labelSelector: matchExpressions: - key: app.kubernetes.io/component operator: In values: - zookeeper topologyKey: "kubernetes.io/hostname" # Kafka Exporter配置，用于监控 kafkaExporter: enabled: true resources: requests: memory: "256Mi" cpu: "250m" limits: memory: "512Mi" cpu: "500m"

部署Kafka

使用自定义配置文件部署Kafka：

helm install my-kafka bitnami/kafka --namespace kafka -f kafka-values.yaml

验证部署

部署完成后，验证Kafka集群状态：

# 查看Pod状态 kubectl get pods -n kafka # 查看服务状态 kubectl get svc -n kafka # 查看PVC状态 kubectl get pvc -n kafka

如果所有Pod都处于Running状态，说明Kafka集群已成功部署。

配置Kafka高可用性

配置Kafka副本

Kafka的高可用性依赖于主题的副本配置。创建一个具有多个副本的主题：

# 获取Kafka Pod名称 KAFKA_POD=$(kubectl get pods -n kafka -l app.kubernetes.io/component=kafka -o jsonpath='{.items[0].metadata.name}') # 创建一个具有3个副本和3个分区的主题 kubectl exec -it -n kafka $KAFKA_POD -- kafka-topics.sh --create --bootstrap-server localhost:9092 --replication-factor 3 --partitions 3 --topic high-availability-topic

配置Kafka镜像

Kafka镜像机制确保了在Broker故障时数据不会丢失。在kafka-values.yaml中，我们已经设置了min.insync.replicas: 2，这意味着至少需要2个副本同步成功才能确认消息写入。

配置Kafka和ZooKeeper的Pod反亲和性

在kafka-values.yaml中，我们已经配置了Pod反亲和性，确保Kafka和ZooKeeper的Pod分布在不同节点上，提高集群的容错能力。

配置外部访问

配置LoadBalancer服务

在kafka-values.yaml中，我们已经将Kafka服务类型设置为LoadBalancer，这将自动创建一个外部负载均衡器，允许外部客户端访问Kafka集群。

获取外部访问地址：

# 获取Kafka外部服务地址 kubectl get svc -n kafka my-kafka -o jsonpath='{.status.loadBalancer.ingress[0].hostname}' # 或者如果是IP kubectl get svc -n kafka my-kafka -o jsonpath='{.status.loadBalancer.ingress[0].ip}'

配置NodePort服务（可选）

如果不想使用LoadBalancer，可以配置NodePort服务：

# 在kafka-values.yaml中修改服务配置 service: type: NodePort ports: client: 9092 external: 9094 nodePort: 31092

配置Ingress（可选）

如果需要通过域名访问Kafka，可以配置Ingress：

# kafka-ingress.yaml apiVersion: networking.k8s.io/v1 kind: Ingress metadata: name: kafka-ingress namespace: kafka annotations: nginx.ingress.kubernetes.io/backend-protocol: "TCP" spec: ingressClassName: nginx rules: - host: kafka.example.com http: paths: - path: / pathType: Prefix backend: service: name: my-kafka port: number: 9092

应用Ingress配置：

kubectl apply -f kafka-ingress.yaml

监控Kafka集群

部署Prometheus和Grafana

为了监控Kafka集群，我们需要部署Prometheus和Grafana：

# 添加Prometheus社区仓库 helm repo add prometheus-community https://prometheus-community.github.io/helm-charts helm repo update # 创建监控命名空间 kubectl create namespace monitoring # 部署Prometheus helm install prometheus prometheus-community/kube-prometheus-stack --namespace monitoring --set grafana.adminPassword=admin # 验证部署 kubectl get pods -n monitoring

配置Kafka Exporter

在kafka-values.yaml中，我们已经启用了Kafka Exporter。现在，我们需要配置Prometheus来抓取Kafka Exporter的指标。

创建Prometheus抓取配置：

# kafka-prometheus-scrape.yaml apiVersion: v1 kind: ConfigMap metadata: name: prometheus-kafka-scrape namespace: monitoring data: kafka-scrape-config.yaml: | - job_name: 'kafka-exporter' static_configs: - targets: ['my-kafka-kafka-exporter.kafka.svc.cluster.local:9308']

应用配置：

kubectl apply -f kafka-prometheus-scrape.yaml

导入Grafana仪表盘

导入Kafka监控仪表盘到Grafana：

# 获取Grafana密码 kubectl get secret --namespace monitoring prometheus-grafana -o jsonpath="{.data.admin-password}" | base64 --decode ; echo # 端口转发Grafana服务 kubectl port-forward --namespace monitoring svc/prometheus-grafana 3000:80 # 访问Grafana (http://localhost:3000)，使用用户名admin和上面获取的密码登录 # 导入Kafka仪表盘，ID为7589 (Kafka Dashboard by Strimzi)

Kafka客户端连接示例

Java客户端示例

创建一个Java Maven项目，添加Kafka客户端依赖：

<!-- pom.xml --> <dependencies> <dependency> <groupId>org.apache.kafka</groupId> <artifactId>kafka-clients</artifactId> <version>3.4.0</version> </dependency> </dependencies>

生产者示例：

// KafkaProducerExample.java import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.ProducerConfig; import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.common.serialization.StringSerializer; import java.util.Properties; public class KafkaProducerExample { public static void main(String[] args) { // 配置生产者属性 Properties props = new Properties(); // 替换为你的Kafka集群地址 props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka.example.com:9092"); props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); // 设置acks为all，确保消息被所有副本接收 props.put(ProducerConfig.ACKS_CONFIG, "all"); // 设置重试次数 props.put(ProducerConfig.RETRIES_CONFIG, "3"); // 设置启用幂等生产者 props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true"); // 创建生产者 KafkaProducer<String, String> producer = new KafkaProducer<>(props); try { // 发送消息 for (int i = 0; i < 10; i++) { // 创建消息记录 ProducerRecord<String, String> record = new ProducerRecord<>("high-availability-topic", "key-" + i, "message-" + i); // 异步发送消息 producer.send(record, (metadata, exception) -> { if (exception != null) { System.err.println("发送消息失败: " + exception); } else { System.out.printf("发送消息成功: topic=%s, partition=%d, offset=%d%n", metadata.topic(), metadata.partition(), metadata.offset()); } }); } } finally { // 关闭生产者 producer.close(); } } }

消费者示例：

// KafkaConsumerExample.java import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.consumer.ConsumerRecords; import org.apache.kafka.clients.consumer.KafkaConsumer; import org.apache.kafka.common.serialization.StringDeserializer; import java.time.Duration; import java.util.Collections; import java.util.Properties; public class KafkaConsumerExample { public static void main(String[] args) { // 配置消费者属性 Properties props = new Properties(); // 替换为你的Kafka集群地址 props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka.example.com:9092"); props.put(ConsumerConfig.GROUP_ID_CONFIG, "test-group"); props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); // 设置自动提交偏移量 props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true"); // 设置自动提交间隔 props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000"); // 设置从最早的消息开始消费 props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); // 创建消费者 KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props); try { // 订阅主题 consumer.subscribe(Collections.singletonList("high-availability-topic")); // 持续消费消息 while (true) { // 轮询消息 ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100)); // 处理消息 for (ConsumerRecord<String, String> record : records) { System.out.printf("接收到消息: partition=%d, offset=%d, key=%s, value=%s%n", record.partition(), record.offset(), record.key(), record.value()); } } } finally { // 关闭消费者 consumer.close(); } } }

Python客户端示例

安装Kafka Python客户端：

pip install kafka-python

生产者示例：

# kafka_producer.py from kafka import KafkaProducer from kafka.errors import KafkaError import json # 配置生产者 producer = KafkaProducer( bootstrap_servers=['kafka.example.com:9092'], value_serializer=lambda v: json.dumps(v).encode('utf-8'), acks='all', retries=3, max_in_flight_requests_per_connection=1, enable_idempotence=True ) try: # 发送消息 for i in range(10): future = producer.send('high-availability-topic', {'key': f'key-{i}', 'value': f'message-{i}'}) # 等待消息发送确认 try: record_metadata = future.get(timeout=10) print(f"发送消息成功: topic={record_metadata.topic}, partition={record_metadata.partition}, offset={record_metadata.offset}") except KafkaError as e: print(f"发送消息失败: {e}") finally: # 关闭生产者 producer.close()

消费者示例：

# kafka_consumer.py from kafka import KafkaConsumer import json # 配置消费者 consumer = KafkaConsumer( 'high-availability-topic', bootstrap_servers=['kafka.example.com:9092'], group_id='python-test-group', auto_offset_reset='earliest', enable_auto_commit=True, auto_commit_interval_ms=1000, value_deserializer=lambda m: json.loads(m.decode('utf-8')) ) try: # 消费消息 for message in consumer: print(f"接收到消息: partition={message.partition}, offset={message.offset}, key={message.key}, value={message.value}") finally: # 关闭消费者 consumer.close()

Kafka集群维护与扩展

扩展Kafka集群

随着业务增长，可能需要扩展Kafka集群。以下是扩展Kafka Broker的方法：

# 更新Kafka副本数量 helm upgrade my-kafka bitnami/kafka --namespace kafka -f kafka-values.yaml --set kafka.replicaCount=5

重新平衡分区

扩展Broker后，需要重新平衡分区以利用新增加的Broker：

# 获取Kafka Pod名称 KAFKA_POD=$(kubectl get pods -n kafka -l app.kubernetes.io/component=kafka -o jsonpath='{.items[0].metadata.name}') # 创建重新平衡配置文件 cat > rebalance.json << EOF { "version": 1, "partitions": [ { "topic": "high-availability-topic", "partition": 0, "replicas": [1, 2, 3] }, { "topic": "high-availability-topic", "partition": 1, "replicas": [2, 3, 4] }, { "topic": "high-availability-topic", "partition": 2, "replicas": [3, 4, 0] } ] } EOF # 执行重新平衡 kubectl cp -n kafka rebalance.json $KAFKA_POD:/tmp/rebalance.json kubectl exec -it -n kafka $KAFKA_POD -- kafka-reassign-partitions.sh --bootstrap-server localhost:9092 --reassignment-json-file /tmp/rebalance.json --execute

滚动更新Kafka

当需要更新Kafka版本或配置时，可以使用Helm进行滚动更新：

# 更新Kafka配置 helm upgrade my-kafka bitnami/kafka --namespace kafka -f kafka-values.yaml --set kafka.image.tag=3.4.0-debian-11-r12

备份与恢复

备份Kafka数据：

# 获取Kafka和ZooKeeper PVC名称 KAFKA_PVC=$(kubectl get pvc -n kafka -l app.kubernetes.io/component=kafka -o jsonpath='{.items[0].metadata.name}') ZOOKEEPER_PVC=$(kubectl get pvc -n kafka -l app.kubernetes.io/component=zookeeper -o jsonpath='{.items[0].metadata.name}') # 创建临时Pod用于备份 kubectl run -n kafka backup-pod --image=busybox --restart=Never -- sleep infinity # 复制Kafka数据 kubectl exec -n kafka backup-pod -- tar czf /tmp/kafka-backup.tar.gz -C /var/lib/kafka/data . kubectl exec -n kafka backup-pod -- tar czf /tmp/zookeeper-backup.tar.gz -C /var/lib/zookeeper/data . # 将备份数据复制到本地 kubectl cp -n kafka backup-pod:/tmp/kafka-backup.tar.gz ./kafka-backup.tar.gz kubectl cp -n kafka backup-pod:/tmp/zookeeper-backup.tar.gz ./zookeeper-backup.tar.gz # 删除临时Pod kubectl delete pod -n kafka backup-pod

恢复Kafka数据：

# 创建临时Pod用于恢复 kubectl run -n kafka restore-pod --image=busybox --restart=Never -- sleep infinity # 将备份数据复制到Pod kubectl cp -n kafka ./kafka-backup.tar.gz restore-pod:/tmp/kafka-backup.tar.gz kubectl cp -n kafka ./zookeeper-backup.tar.gz restore-pod:/tmp/zookeeper-backup.tar.gz # 恢复数据 kubectl exec -n kafka restore-pod -- mkdir -p /tmp/kafka-data /tmp/zookeeper-data kubectl exec -n kafka restore-pod -- tar xzf /tmp/kafka-backup.tar.gz -C /tmp/kafka-data kubectl exec -n kafka restore-pod -- tar xzf /tmp/zookeeper-backup.tar.gz -C /tmp/zookeeper-data # 将恢复的数据复制到PVC kubectl exec -n kafka restore-pod -- cp -r /tmp/kafka-data/* /var/lib/kafka/data/ kubectl exec -n kafka restore-pod -- cp -r /tmp/zookeeper-data/* /var/lib/zookeeper/data/ # 删除临时Pod kubectl delete pod -n kafka restore-pod

最佳实践与性能优化

硬件选择

存储：使用SSD存储，特别是对于ZooKeeper和Kafka的日志段
网络：确保网络带宽足够，建议使用10GbE或更高速率的网络
内存：为Kafka分配足够的堆内存，但不要超过系统内存的50%
CPU：Kafka对CPU要求不高，但足够的CPU核心可以提高并行处理能力

Kafka配置优化

# 在kafka-values.yaml中添加以下优化配置 kafka: configuration: # 优化网络和IO操作 num.network.threads: 6 num.io.threads: 8 socket.send.buffer.bytes: 1024000 socket.receive.buffer.bytes: 1024000 socket.request.max.bytes: 104857600 # 优化日志刷新 log.flush.interval.messages: 10000 log.flush.interval.ms: 1000 # 优化副本同步 replica.lag.time.max.ms: 10000 # 优化消费者 fetch.purgatory.purge.interval.requests: 1000 producer.purgatory.purge.interval.requests: 1000 # 启用压缩 compression.type: "lz4"

JVM调优

# 在kafka-values.yaml中添加JVM调优配置 kafka: heapOpts: "-Xms2g -Xmx2g" jvmOptions: > -XX:+UseG1GC -XX:MaxGCPauseMillis=20 -XX:InitiatingHeapOccupancyPercent=35 -XX:+ExplicitGCInvokesConcurrent -Djava.awt.headless=true

安全配置

# 在kafka-values.yaml中添加安全配置 kafka: # 启用SASL认证 sasl: enabled: true mechanisms: ["PLAIN", "SCRAM-SHA-256", "SCRAM-SHA-512"] jaas: clientUsers: ["user1", "user2"] clientPasswords: ["password1", "password2"] # 启用SSL加密 tls: enabled: true autoGenerated: true # 或者使用现有证书 # existingSecret: "kafka-tls-secret" # 启用ACL authorizerClassName: "kafka.security.authorizer.AclAuthorizer" allowEveryoneIfNoAclFound: false superUsers: "User:admin"

资源隔离

# 在kafka-values.yaml中添加资源隔离配置 kafka: # 使用节点选择器将Kafka部署到特定节点 nodeSelector: node-role.kubernetes.io/kafka: "true" # 使用容忍度允许Kafka部署到有污点的节点 tolerations: - key: "dedicated" operator: "Equal" value: "kafka" effect: "NoSchedule" # 使用Pod优先级确保Kafka Pod的重要性 priorityClassName: "high-priority"

故障排除

常见问题及解决方案

1. Kafka Pod启动失败

问题：Kafka Pod无法启动，一直处于CrashLoopBackOff状态。

解决方案：

# 查看Pod日志 kubectl logs -n kafka <kafka-pod-name> --previous # 查看Pod描述 kubectl describe pod -n kafka <kafka-pod-name> # 检查PVC是否已绑定 kubectl get pvc -n kafka # 检查存储类是否可用 kubectl get storageclass

2. 生产者连接超时

问题：客户端无法连接到Kafka集群，出现连接超时错误。

解决方案：

# 检查Kafka服务状态 kubectl get svc -n kafka # 检查Kafka Pod状态 kubectl get pods -n kafka # 检查网络策略是否阻止了访问 kubectl get networkpolicy -n kafka # 检查Kafka配置中的监听器设置 kubectl exec -it -n kafka <kafka-pod-name> -- cat /opt/bitnami/kafka/config/server.properties | grep listeners

3. 消息丢失

问题：生产者发送的消息没有被消费者接收。

解决方案：

# 检查主题配置 kubectl exec -it -n kafka <kafka-pod-name> -- kafka-topics.sh --describe --bootstrap-server localhost:9092 --topic <topic-name> # 检查消费者组状态 kubectl exec -it -n kafka <kafka-pod-name> -- kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group <group-name> # 检查生产者配置，确保acks设置为all # 检查消费者配置，确保auto.offset.reset设置正确

4. ZooKeeper连接问题

问题：Kafka无法连接到ZooKeeper。

解决方案：

# 检查ZooKeeper Pod状态 kubectl get pods -n kafka -l app.kubernetes.io/component=zookeeper # 检查ZooKeeper服务状态 kubectl get svc -n kafka -l app.kubernetes.io/component=zookeeper # 测试ZooKeeper连接 kubectl exec -it -n kafka <kafka-pod-name> -- telnet my-kafka-zookeeper 2181 # 检查ZooKeeper日志 kubectl logs -n kafka <zookeeper-pod-name>

性能问题排查

1. 高延迟

问题：消息生产或消费延迟高。

解决方案：

# 检查Kafka指标 kubectl exec -it -n kafka <kafka-pod-name> -- kafka-run-class.sh kafka.tools.JmxTool --jmx-url service:jmx:rmi:///jndi/rmi://:9999/jmxrmi --object-name kafka.server:type=BrokerTopicMetrics,name=RequestQueueSize # 检查磁盘IO kubectl exec -it -n kafka <kafka-pod-name> -- iostat -x 1 # 检查网络流量 kubectl exec -it -n kafka <kafka-pod-name> -- ifstat -i eth0 1 # 检查GC情况 kubectl exec -it -n kafka <kafka-pod-name> -- jstat -gc <pid> 1s

2. 低吞吐量

问题：Kafka集群吞吐量低于预期。

解决方案：

# 检查分区数量 kubectl exec -it -n kafka <kafka-pod-name> -- kafka-topics.sh --describe --bootstrap-server localhost:9092 --topic <topic-name> # 检查消费者滞后情况 kubectl exec -it -n kafka <kafka-pod-name> -- kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group <group-name> # 检查Broker负载 kubectl exec -it -n kafka <kafka-pod-name> -- kafka-run-class.sh kafka.tools.JmxTool --jmx-url service:jmx:rmi:///jndi/rmi://:9999/jmxrmi --object-name kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec