kafka用java收发消息-编程知识

用java客户端代码来对kafka收发消息
具体代码如下

package com.cool.interesting.kafka;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.clients.producer.*;
import org.apache.kafka.common.TopicPartition;import java.time.Duration;
import java.util.Collections;
import java.util.Properties;
import java.util.concurrent.Future;public class KafkaTest {private static final String BOOTSTRAP_SERVERS = "192.168.47.145:9092";private static final String TOPIC_NAME = "test";public static void main(String[] args) {// 生产者示例produceMessage();// 消费者示例consumeMessage();//从指定偏移量消费消息consumeOffsetMessage();}//生产者代码private static void produceMessage() {Properties props = new Properties();//acks是保证消息的发送机制，有以下几个值//acks = 0：表示生产端发送消息后立即返回，不等待broker端的响应结果。通常此时生产端吞吐量最高，消息发送的可靠性最低。//acks = 1: 表示leader副本成功写入就会响应Producer，而无需等待ISR（同步副本）集合中的其他副本写入成功。这种方案提供了适当的持久性，保证了一定的吞吐量。默认值即是1。//acks = all或-1: 表示不仅要等leader副本成功写入，还要求ISR中的其他副本成功写入，才会响应Producer。这种方案提供了最高的持久性，但也提供了最差的吞吐量。//调优建议：建议根据实际情况设置，如果要严格保证消息不丢失，请设置为all或-1；如果允许存在丢失，建议设置为1；一般不建议设为0，除非无所谓消息丢不丢失。props.put(ProducerConfig.ACKS_CONFIG,1);props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);//key和value序列化props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");//其他配置参数详见org.apache.kafka.clients.producer.ProducerConfig类try (Producer<String, String> producer = new KafkaProducer<>(props)) {for (int i = 0; i < 10; i++) {String message = "Message " + i;//异步发送Future<RecordMetadata> send = producer.send(new ProducerRecord<>(TOPIC_NAME,  message));System.out.println("Sent message: " + message);}}}//正常消费者代码private static void consumeMessage() {Properties props = new Properties();props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);//将订阅的topic绑定到一个消费者（这个group_id 是自己定义的）props.put(ConsumerConfig.GROUP_ID_CONFIG, "test99");props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);//订阅一个topicconsumer.subscribe(Collections.singletonList(TOPIC_NAME));while (true) {//设置kafak从broker拉取消息的超时时间// （这意味着 poll() 方法将在等待最多 2秒的时间内尝试从 Kafka 集群拉取消息，如果在超时时间内没有拉取到消息，将返回一个空的 ConsumerRecords 对象）ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(2));for (ConsumerRecord<String, String> record : records) {System.out.println("Received_message: " + record.value());}}}//指定偏移量开始消费private static void consumeOffsetMessage() {Properties props = new Properties();props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);//将订阅的topic绑定到一个消费者（这个group_id 是自己定义的）props.put(ConsumerConfig.GROUP_ID_CONFIG, "test");props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);//订阅一个topicconsumer.subscribe(Collections.singletonList(TOPIC_NAME));//如果要指定偏移量，必须先poll一次，不然代码报错ConsumerRecords<String, String> poll = consumer.poll(0);System.out.println("poll:"+poll.isEmpty());//创建一个分区（参数为topic_name，和分区序号）TopicPartition topicPartition = new TopicPartition(TOPIC_NAME, 0);// 指定要消费的偏移量long offset = 3;//从指定偏移量开始消息消息consumer.seek(topicPartition, offset);while (true) {//设置kafka从broker拉取消息的超时时间// （这意味着 poll() 方法将在等待最多 2秒的时间内尝试从 Kafka 集群拉取消息，如果在超时时间内没有拉取到消息，将返回一个空的 ConsumerRecords 对象）ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(2));for (ConsumerRecord<String, String> record : records) {System.out.println("Received_message: " + record.value());}}}}

安装kafka的可视化工具：offset explorer
offset explorer 是一个用于查看和管理 Kafka 消费者组的工具，它允许你检查消费者组的偏移量（offset），并且可以查看每个消费者组在每个分区上的偏移量情况。这对于监控和调试 Kafka 消费者组非常有用。
下载地址为：https://www.kafkatool.com/download.html
如下图所示：
在这里插入图片描述