弹性 MapReduce Kafka 使用

一、命令方式

bin/kafka-console-producer.sh –bootstrap-server node86:9092 –topic t_cdr

二、Java 代码方式

@Component@Slf4jpublic class KafkaConsumer { @KafkaListener(topics = KafkaProducer.TOPIC_TEST, groupId = KafkaProducer.TOPIC_GROUP1) public void topic_test(ConsumerRecord<?, ?> record, Acknowledgment ack, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) { Optional message = Optional.ofNullable(record.value()); if (message.isPresent()) { Object msg = message.get(); log.info("topic_test 消费了： Topic:" + topic + ",Message:" + msg); ack.acknowledge(); } } @KafkaListener(topics = KafkaProducer.TOPIC_TEST, groupId = KafkaProducer.TOPIC_GROUP2) public void topic_test1(ConsumerRecord<?, ?> record, Acknowledgment ack, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) { Optional message = Optional.ofNullable(record.value()); if (message.isPresent()) { Object msg = message.get(); log.info("topic_test1 消费了： Topic:" + topic + ",Message:" + msg); ack.acknowledge(); } }}

三、命令方式

bin/kafka-console-consumer.sh –bootstrap-server node86:9092 –topic t_cdr –from-beginning

四、命令方式

bin/kafka-topics.sh –bootstrap-server node86:9092 –create –topic t_cdr –partitions 30 –replication-factor 2

详细使用可参考 kafka 产品文档。

五、Kafka3.x 连接方式

Kafka 3.x 已移除命令行工具的 –zookeeper 参数，所有脚本统一使用 –bootstrap-server 连接 Broker，端口默认为 9092。–bootstrap-server 后填写 Kafka Broker 节点的内网 IP 或主机名（例如 node86:9092），多个 Broker 之间用英文逗号分隔。

客户端代码只需将 bootstrap.servers 配置为 Kafka Broker 地址（例如 node86:9092），无需再配置 ZooKeeper 地址。