初试kafka消息队列中间件二（采用java代码收发消息）

今天的案例主要是将采用命令行收发信息改成使用java代码实现，根据上一篇的接着写；

先启动Zookeeper，然后启动Kafka，再创建消息主题；

以上三步我就不重复了，不会的看上一篇即可

maven依赖

    <dependency>

      <groupId>org.apache.kafka</groupId>

      <artifactId>kafka-clients</artifactId>

      <version>2.3.</version>

    </dependency>

发送消息的代码

				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 Test01 {

				    public static void main(String[] args){

				        Properties properties = new Properties();

				        properties.put("bootstrap.servers", "192.168.31.223:9092");

				        properties.put("acks", "all");

				        properties.put("retries", 0);

				        properties.put("batch.size", 16384);

				        properties.put("linger.ms", 1);

				        properties.put("buffer.memory", 33554432);

				        properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");

				        properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

				        Producer<String, String> producer = null;

				        try {

				            producer = new KafkaProducer<String, String>(properties);

				                String msg = "Message111122 " ;

				               //此处的msg1代表消息的主题

				                producer.send(new ProducerRecord<String, String>("msg1", msg));

				                System.out.println("Sent:" + msg);

				        } catch (Exception e) {

				            e.printStackTrace();

				        } finally {

				            producer.close();

				        }

				    }

				}

接受消息的代码

				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 Test02 {

				    public static void main(String[] args){

				        Properties properties = new Properties();

				        properties.put("bootstrap.servers", "192.168.31.223:9092");

				        properties.put("group.id", "group-1");

				        properties.put("enable.auto.commit", "true");

				        properties.put("auto.commit.interval.ms", "1000");

				        properties.put("auto.offset.reset", "earliest");

				        properties.put("session.timeout.ms", "30000");

				        properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

				        properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

				        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);

				        kafkaConsumer.subscribe(Arrays.asList("msg1"));

				        while (true) {

				            ConsumerRecords<String, String> records = kafkaConsumer.poll(100);

				            for (ConsumerRecord<String, String> record : records) {

				                System.out.printf("offset = %d, value = %s", record.offset(), record.value());

				                System.out.println();

				            }

				        }

				    }

				}

结果：

期间遇到过一些错误

1、启动报错

错误信息：

				SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".

				SLF4J: Defaulting to no-operation (NOP) logger implementation

				SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.

处理：

缺少maven依赖

   <dependency>

      <groupId>org.slf4j</groupId>

      <artifactId>slf4j-nop</artifactId>

      <version>1.7.</version>

    </dependency>

2、创建消息主题错误

错误信息：命令行创建消息主题提示已经存在

说明以前创建的主题并没有删除，已存在的可以不用再创建，想要删除的需要去改配置，自行百度即可；

秒客网

初试kafka消息队列中间件二（采用java代码收发消息）

相关文章