依赖
引入以下依赖
<!-- https://mvnrepository.com/artifact/org.apache.kafka/kafka-clients --><dependency><groupId>org.apache.kafka</groupId><artifactId>kafka-clients</artifactId><version>4.1.0</version></dependency>
生产者
- 文档： https://kafka.apache.org/documentation/#producerapi
- 设置服务配置，例如序列化，IP端口等
final Properties props = new Properties() {{// 服务IP端口配置put(BOOTSTRAP_SERVERS_CONFIG, "study.fedora01.com:9092,study.fedora02.com:9092,study.fedora03.com:9092");// 序列化put(KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getCanonicalName());put(VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getCanonicalName());}};
- 发送消息
2.1 单向发送，不关心是否发送成功
try (final Producer<String, String> producer = new KafkaProducer<>(props)) {final Random rnd = new Random();final int numMessages = 10;for (int i = 0; i < numMessages; i++) {String user = users[rnd.nextInt(users.length)];String item = items[rnd.nextInt(items.length)];// 单向发送, 不关心是否发送成功producer.send(new ProducerRecord<>(topic, user, item));System.out.printf("%s events were produced to topic %s%n", numMessages, topic);}
2.2 同步发送, 发送完成后获取服务端发送情况，如果一直未发送成功那么就会进行阻塞
try (final Producer<String, String> producer = new KafkaProducer<>(props)) {final Random rnd = new Random();final int numMessages = 10;for (int i = 0; i < numMessages; i++) {String user = users[rnd.nextInt(users.length)];String item = items[rnd.nextInt(items.length)];// 同步发送, 发送完成后获取服务端发送情况，如果一直未发送成功那么就会进行阻塞RecordMetadata recordMetadata = producer.send(new ProducerRecord<>(topic, user, item),(event, ex) -> {if (ex != null)ex.printStackTrace();elseSystem.out.printf("Produced event to topic %s: key = %-10s value = %s%n", topic, user, item);}).get();}
2.3 异步发送, 回调函数处理异步响应信息
try (final Producer<String, String> producer = new KafkaProducer<>(props)) {final Random rnd = new Random();final int numMessages = 10;for (int i = 0; i < numMessages; i++) {String user = users[rnd.nextInt(users.length)];String item = items[rnd.nextInt(items.length)];// 异步发送, 回调函数处理异步响应信息producer.send(new ProducerRecord<>(topic, user, item),(event, ex) -> {if (ex != null)ex.printStackTrace();elseSystem.out.printf("Produced event to topic %s: key = %-10s value = %s%n", topic, user, item);});}System.out.printf("%s events were produced to topic %s%n", numMessages, topic);}
消费者
- 文档: https://kafka.apache.org/documentation/#consumerapi
- 设置配置信息
String topic = "producer-topic";Properties props = new Properties();// 设置服务地址props.put("bootstrap.servers", "study.fedora01.com:9092,study.fedora02.com:9092,study.fedora03.com:9092");// 消费者端需要设置一个group id来消费信息props.put("group.id", "test-group-2"); // 序列化props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");Consumer<String, String> consumer = new KafkaConsumer<>(props);
- 监听topic
consumer.subscribe(Collections.singletonList(topic));System.out.println("开始消费topic: " + topic);
- 处理消息，每隔1000毫秒拉取一次
while (true) {ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));if (records.isEmpty()) {System.out.println("还没有读取到消息...");}for (ConsumerRecord<String, String> record : records) {System.out.printf("Key=%s, Value=%s, Partition=%d, Offset=%d%n",record.key(), record.value(), record.partition(), record.offset());}}