当前位置:   article > 正文

kafka java_Java操作Kafka

kafka java

java操作kafka非常的简单,然后kafka也提供了很多缺省值,一般情况下我们不需要修改太多的参数就能使用。下面我贴出代码。

pom.xml

org.apache.kafka

kafka-clients

0.10.2.0

生产者:

packagecn.duanjt;importjava.util.Properties;importjava.util.Random;importorg.apache.kafka.clients.producer.KafkaProducer;importorg.apache.kafka.clients.producer.ProducerConfig;importorg.apache.kafka.clients.producer.ProducerRecord;importorg.apache.kafka.common.serialization.StringSerializer;public classProducer {public static String topic = "duanjt_test";//定义主题

public static void main(String[] args) throwsInterruptedException {

Properties p= newProperties();

p.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.23.76:9092,192.168.23.77:9092");//kafka地址,多个地址用逗号分割

p.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);

p.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);

KafkaProducer kafkaProducer = new KafkaProducer<>(p);try{while (true) {

String msg= "Hello," + new Random().nextInt(100);

ProducerRecord record = new ProducerRecord(topic, msg);

kafkaProducer.send(record);

System.out.println("消息发送成功:" +msg);

Thread.sleep(500);

}

}finally{

kafkaProducer.close();

}

}

}

注意:

1.kafka如果是集群,多个地址用逗号分割(,)

2.Properties的put方法,第一个参数可以是字符串,如:p.put("bootstrap.servers","192.168.23.76:9092")

3.kafkaProducer.send(record)可以通过返回的Future来判断是否已经发送到kafka,增强消息的可靠性。同时也可以使用send的第二个参数来回调,通过回调判断是否发送成功。

4.p.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);设置序列化类,可以写类的全路径

消费者:

packagecn.duanjt;importjava.util.Collections;importjava.util.Properties;importorg.apache.kafka.clients.consumer.ConsumerConfig;importorg.apache.kafka.clients.consumer.ConsumerRecord;importorg.apache.kafka.clients.consumer.ConsumerRecords;importorg.apache.kafka.clients.consumer.KafkaConsumer;importorg.apache.kafka.common.serialization.StringDeserializer;public classConsumer {public static voidmain(String[] args) {

Properties p= newProperties();

p.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.23.76:9092");

p.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);

p.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);

p.put(ConsumerConfig.GROUP_ID_CONFIG,"duanjt_test");

KafkaConsumer kafkaConsumer = new KafkaConsumer(p);

kafkaConsumer.subscribe(Collections.singletonList(Producer.topic));//订阅消息

while (true) {

ConsumerRecords records = kafkaConsumer.poll(100);for (ConsumerRecordrecord : records) {

System.out.println(String.format("topic:%s,offset:%d,消息:%s", //record.topic(), record.offset(), record.value()));

}

}

}

}

注意:

1.订阅消息可以订阅多个主题

2.ConsumerConfig.GROUP_ID_CONFIG表示消费者的分组,kafka根据分组名称判断是不是同一组消费者,同一组消费者去消费一个主题的数据的时候,数据将在这一组消费者上面轮询。

3.主题涉及到分区的概念,同一组消费者的个数不能大于分区数。因为:一个分区只能被同一群组的一个消费者消费。出现分区小于消费者个数的时候,可以动态增加分区。

4.注意和生产者的对比,Properties中的key和value是反序列化,而生产者是序列化。

声明:本文内容由网友自发贡献,不代表【wpsshop博客】立场,版权归原作者所有,本站不承担相应法律责任。如您发现有侵权的内容,请联系我们。转载请注明出处:https://www.wpsshop.cn/w/黑客灵魂/article/detail/985176
推荐阅读
相关标签
  

闽ICP备14008679号