1. 查看生产者日志
生产者通常会记录它的活动日志，包括连接尝试和连接成功或失败的消息。在生产者应用的日志中搜索连接相关的信息。
2. 使用Kafka工具查看生产者活动
使用Kafka命令行工具或管理工具查看主题的活动情况：
使用kafka-consumer-groups工具
如果生产者正在生产消息，消费者组也会有活动。你可以查看消费者组的偏移量变化。
kafka-consumer-groups.sh --bootstrap-server <kafka-broker>:<port> --describe --group <consumer-group>
使用kafka-topics工具
查看主题的描述信息，检查分区的消息偏移量是否在增加。
kafka-topics.sh --bootstrap-server <kafka-broker>:<port> --describe --topic <your-topic>
3. 使用JMX监控
Kafka生产者和Broker都支持通过JMX（Java Management Extensions）暴露各种监控指标。你可以通过以下步骤查看连接情况：
- 启动Kafka时启用JMX监控：
export KAFKA_JMX_OPTS="-Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.port=9999 -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false"
- 使用JMX工具（如JConsole或VisualVM）连接到JMX端口，查看生产者的连接指标。
4. 检查Kafka Broker的连接数
通过Kafka Broker的JMX监控，可以查看当前活跃的连接数。检查是否有来自生产者的连接。
5. 编写生产者代码进行连接测试
编写一个简单的生产者程序，发送测试消息并捕获异常以验证连接情况。
import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.clients.producer.RecordMetadata; import java.util.Properties; public class ProducerConnectionTest { public static void main(String[] args) { Properties props = new Properties(); props.put("bootstrap.servers", "your-kafka-broker:port"); props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); KafkaProducer<String, String> producer = new KafkaProducer<>(props); ProducerRecord<String, String> record = new ProducerRecord<>("your-topic", "key", "value"); try { RecordMetadata metadata = producer.send(record).get(); System.out.printf("Sent record to partition %d with offset %d%n", metadata.partition(), metadata.offset()); } catch (Exception e) { e.printStackTrace(); } finally { producer.close(); } } }
6. 使用Kafka管理工具
工具如Kafka Manager、Confluent Control Center、Lenses.io等，可以帮助你监控Kafka集群，包括生产者的连接情况。
通过以上方法，可以确认Kafka生产者是否成功连接到Kafka服务。
public class BlogEnding { public static void main(String[] args) { encourageEngagement(); } public static void encourageEngagement() { System.out.println("🚀 感谢您阅读本文！如果您觉得有收获，请一键三连：点赞 ❤️️、转发 🔁、评论 💬，并加关注哦！"); } }
