kafka用java生产和消费消息问题

向右看齐 2023-06-08 06:22 9阅读 0赞

初次使用kafka踩了好多坑，分享如下：

消费者发送消息看似成功，但在虚拟机liunx上也获取不到。需要在配置config/server.properties配置文件中添加远程监听如下：

advertised.listeners=PLAINTEXT://服务器外网ip:9092 注释掉listeners=PLAINTEXT://:9092

好了，重启之后生成者可以生产消息了。但是又发先消费者无法消费消息，查找了多资料。最后尝试注释掉maven pom文件中对kafka-streams的依赖测试成功。可能kafka结合streams使用方式不是这种方式导致。代码如下：

\----------------------pom文件-----------------------------------

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">  
<modelVersion>4.0.0</modelVersion>  
<groupId>com.project.kafka</groupId>  
<artifactId>kafka</artifactId>  
<version>0.0.1-SNAPSHOT</version>  
<packaging>war</packaging>  
  
<dependencies>  
<dependency>  
<groupId>org.apache.kafka</groupId>  
<artifactId>kafka-clients</artifactId>  
<version>2.3.0</version>  
</dependency>  
  
</dependencies>  
</project>

\--------------------------------------------Producer-------------------------------------------

package com.test;

import java.util.Properties;

import org.apache.kafka.clients.producer.KafkaProducer;  
import org.apache.kafka.clients.producer.ProducerRecord;

public class Producer \{

public static void main(String\[\] args) \{  
Properties props = new Properties();  
props.put("bootstrap.servers", "192.168.56.101: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");

org.apache.kafka.clients.producer.Producer<String, String> producer = new KafkaProducer<String, String>(props);  
for (int i = 0; i < 100; i++) \{  
producer.send(new ProducerRecord<String, String>("testtt", Integer.toString(i), Integer.toString(i)));

System.out.println("==============");  
\}

producer.close();  
\}  
\}

\----------------------------------------------Consumer -----------------------------------------------------------------

package com.test;

import java.time.Duration;  
import java.util.Arrays;  
import java.util.Properties;

import org.apache.kafka.clients.consumer.ConsumerRecord;  
import org.apache.kafka.clients.consumer.ConsumerRecords;  
import org.apache.kafka.clients.consumer.KafkaConsumer;

public class Consumer \{  
public static void main(String\[\] args) \{  
Properties props = new Properties();  
props.setProperty("bootstrap.servers", "192.168.56.101:9092");  
props.setProperty("group.id", "test");  
props.setProperty("enable.auto.commit", "true");  
props.setProperty("auto.commit.interval.ms", "1000");  
props.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");  
props.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");  
KafkaConsumer<String, String> consumer = new KafkaConsumer<String,String>(props);  
consumer.subscribe(Arrays.asList("testtt"));  
while (true) \{  
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));  
for (ConsumerRecord<String, String> record : records)  
System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());  
\}  
\}  
\}