网站栏目建设征求意见,网页设计基础课程设计问题反馈,建站63年来第一次闭站?北京站辟谣,客户说做网站没效果怎么回答好目录 一、独立消费者消费某一个主题中某个分区数据案例1.1、案例需求1.2、案例代码1.3、测试 一、独立消费者消费某一个主题中某个分区数据案例
1.1、案例需求
创建一个独立消费者#xff0c;消费firstTopic主题 0 号分区的数据#xff0c;所下图所示#xff1a;
1.2、案… 目录 一、独立消费者消费某一个主题中某个分区数据案例1.1、案例需求1.2、案例代码1.3、测试 一、独立消费者消费某一个主题中某个分区数据案例
1.1、案例需求
创建一个独立消费者消费firstTopic主题 0 号分区的数据所下图所示
1.2、案例代码 生产者往firstTopic主题 0 号分区发送数据代码 package com.xz.kafka.producer;import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;
import java.util.Properties;public class CustomProducerCallback {public static void main(String[] args) throws InterruptedException {//1、创建 kafka 生产者的配置对象Properties properties new Properties();//2、给 kafka 配置对象添加配置信息bootstrap.serversproperties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,192.168.136.27:9092,192.168.136.28:9092,192.168.136.29:9092);//3、指定对应的key和value的序列化类型 key.serializer value.serializerproperties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());//4、创建 kafka 生产者对象KafkaProducerString, String kafkaProducer new KafkaProducer(properties);//5、调用 send 方法,发送消息for (int i 0; i 5; i) {kafkaProducer.send(new ProducerRecord(firstTopic, 0,,hello kafka i), new Callback() {Overridepublic void onCompletion(RecordMetadata metadata, Exception exception) {if (exception null){System.out.println(主题 metadata.topic() 分区 metadata.partition());}}});Thread.sleep(2);}// 3 关闭资源kafkaProducer.close();}
} 消费者消费firstTopic主题 0 分区数据代码 package com.xz.kafka.consumer;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.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;import java.time.Duration;
import java.util.ArrayList;
import java.util.Properties;public class CustomConsumerPartition {public static void main(String[] args) {// 配置Properties properties new Properties();// 连接properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,192.168.136.27:9092,192.168.136.28:9092,192.168.136.29:9092);// 反序列化properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());// 组idproperties.put(ConsumerConfig.GROUP_ID_CONFIG,test);// 1 创建一个消费者KafkaConsumerString, String kafkaConsumer new KafkaConsumer(properties);// 2 订阅主题对应的分区ArrayListTopicPartition topicPartitions new ArrayList();topicPartitions.add(new TopicPartition(firstTopic,0));kafkaConsumer.assign(topicPartitions);// 3 消费数据while (true){ConsumerRecordsString, String consumerRecords kafkaConsumer.poll(Duration.ofSeconds(1));for (ConsumerRecordString, String consumerRecord : consumerRecords) {System.out.println(consumerRecord);}}}
}1.3、测试
在 IDEA 中执行消费者程序如下图 在 IDEA 中执行生产者程序 在控制台观察生成几个 0号分区的数据如下图 在 IDEA 控制台观察接收到的数据只能消费到 0 号分区数据表示正确。
