怎么在Java中利用kafka发送消息

这期内容当中小编将会给大家带来有关怎么在Java中利用kafka发送消息，文章内容丰富且以专业的角度为大家分析和叙述，阅读完这篇文章希望大家可以有所收获。

1. maven依赖包

<dependency>   <groupId>org.apache.kafka</groupId>   <artifactId>kafka-clients</artifactId>   <version>0.9.0.1</version>  </dependency>

2. 生产者代码

package com.lnho.example.kafka;   import org.apache.kafka.clients.producer.KafkaProducer;  import org.apache.kafka.clients.producer.Producer;  import org.apache.kafka.clients.producer.ProducerRecord;    import java.util.Properties;    public class KafkaProducerExample {   public static void main(String[] args) {    Properties props = new Properties();    props.put("bootstrap.servers", "master:9092");    props.put("acks", "all");    props.put("retries", 0);    props.put("batch.size", 16384);    props.put("linger.ms", 1);    props.put("buffer.memory", 33554432);    props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");    props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");      Producer<String, String> producer = new KafkaProducer<>(props);    for(int i = 0; i < 100; i++)     producer.send(new ProducerRecord<>("topic1", Integer.toString(i), Integer.toString(i)));      producer.close();   }  }

3. 消费者代码

package com.lnho.example.kafka;    import org.apache.kafka.clients.consumer.ConsumerRecord;  import org.apache.kafka.clients.consumer.ConsumerRecords;  import org.apache.kafka.clients.consumer.KafkaConsumer;  import java.util.Arrays;  import java.util.Properties;    public class KafkaConsumerExample {   public static void main(String[] args) {    Properties props = new Properties();    props.put("bootstrap.servers", "master:9092");    props.put("group.id", "test");    props.put("enable.auto.commit", "true");    props.put("auto.commit.interval.ms", "1000");    props.put("session.timeout.ms", "30000");    props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");    props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");    KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);    consumer.subscribe(Arrays.asList("topic1"));    while (true) {     ConsumerRecords<String, String> records = consumer.poll(100);     for (ConsumerRecord<String, String> record : records)      System.out.printf("offset = %d, key = %s, value = %s\n", record.offset(), record.key(), record.value());    }   }  }

上述就是小编为大家分享的怎么在Java中利用kafka发送消息了，如果刚好有类似的疑惑，不妨参照上述分析进行理解。如果想知道更多相关知识，欢迎关注亿速云行业资讯频道。