Kafka(四) ：Java 编写简单的kafka生产者和消费者

水深无声 2023-07-22 09:18 2阅读 0赞

### 一、准备工作 ###

*  框架：springboot
 *  电脑查看网关是否开启工具：[telnet 安装（Mac）][telnet _Mac]
 *  引入依赖：

<dependency>
       <groupId>org.springframework.kafka</groupId>
       <artifactId>spring-kafka</artifactId>
     </dependency>

### 二、生产者 ###

public static void main(String[] args) {
            //创建生产者配置信息
            Properties properties = new Properties();
            //设置key
            properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
            //设置重试次数
            properties.put(ProducerConfig.RETRIES_CONFIG,10);
            //设置value
            properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());
            //设置服务地址
            properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"http://106.52.174.56:9092");
    
            //设置kafka
            KafkaProducer<String, String> producers = new KafkaProducer<String, String>(properties);
            //设置消息
            ProducerRecord<String,String> producerRecords = new ProducerRecord<>("kafkalearn","learn-info","hello word!");
            //发送消息
            producers.send(producerRecords);
            //关闭
            producers.close();
    
        }

### 三、消费者 ###

public static void main(String[] args) {
    
            //创建消费者配置信息
            Properties properties = new Properties();
            //设置key
            properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
            //设置消费组
            properties.put(ConsumerConfig.GROUP_ID_CONFIG,"group.demo");
            //设置value
            properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName());
            //设置服务地址
            properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"http://106.52.174.56:9092");
    
            //设置kafka
            KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(properties);
            //订阅消息
            consumer.subscribe(Collections.singletonList("kafkalearn"));
            //监听消息,间隔2秒轮询一次
            while (true){
                ConsumerRecords<String, String> poll = consumer.poll(Duration.ofMillis(2000));
                poll.forEach(item->{
                    System.out.println("结果呢：----"+item.key()+"--"+item.value());
                });
    
            }

### 五、出现的问题 ###

1.  centeros上本地连接不上，报错，

*  **  解决办法：****[无法远程连接kafka][kafka]**

[telnet _Mac]: https://www.cnblogs.com/zhuminghui/p/10309003.html
[kafka]: https://blog.csdn.net/maoyuanming0806/article/details/80555979