SpringBoot集成系列--Kafka

文章目录

一、代码

1、添加依赖

在pom.xml文件中添加Kafka的依赖

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

2、配置kafka

在

生产者项目

的application.properties文件中配置kafka

spring.kafka.bootstrap-servers=192.168.56.100:9092
spring.kafka.producer.client-id=forlan-client-id
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer

参数说明：

• spring.kafka.bootstrap-servers：指定Kafka集群的地址。这是一个以逗号分隔的服务器列表，每个服务器都以IP地址和端口号的形式表示。
• spring.kafka.producer.client-id：指定Kafka生产者的客户端ID。这是一个用于标识生产者的唯一字符串。在诊断和监控时，这个ID可以帮助识别发送给Kafka的哪些消息是由哪个生产者发送的。
• spring.kafka.producer.key-serializer：定义Kafka的消息的键的序列化器。
• spring.kafka.producer.value-serializer：定义Kafka的消息的值的序列化器。

在

消费者项目

的application.properties文件中配置kafka消费者

spring.kafka.bootstrap-servers=192.168.56.100:9092
spring.kafka.consumer.client-id=forlan-client-id
spring.kafka.consumer.group-id=forlan-group-id
spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer

参数说明：

• spring.kafka.bootstrap-servers：指定Kafka集群的地址。这是一个以逗号分隔的服务器列表，每个服务器都以IP地址和端口号的形式表示。
• spring.kafka.consumer.client-id：指定Kafka消费者的客户端ID。这是一个用于标识消费者的唯一字符串。在诊断和监控时，这个ID可以帮助识别从Kafka接收到的哪些消息是由哪个消费者接收的。
• spring.kafka.consumer.group-id：指定Kafka消费者的组ID。在消费者组中，成员之间可以共享消息消费进度，以便在需要时进行重新消费或进行故障转移。
• spring.kafka.consumer.key-deserializer：指定用于反序列化从Kafka接收的消息的键的类。
• spring.kafka.consumer.value-deserializer：指定用于反序列化从Kafka接收的消息的值的类。

3、创建生产者

设置发送消息服务类

@ComponentpublicclassKafkaProducer{@AutowiredprivateKafkaTemplate<String,String> kafkaTemplate;publicvoidtestSend(String topic,String message)throwsExecutionException,InterruptedException{SendResult<String,String> stringStringSendResult = kafkaTemplate.send(topic, message).get();System.out.println(stringStringSendResult);}}

4、创建消费者

importorg.springframework.kafka.annotation.KafkaListener;importorg.springframework.stereotype.Service;@ServicepublicclassKafkaConsumer{@KafkaListener(topics ="forlan_topic")publicvoidlisten(String message){System.out.println("Kafka收到消息："+ message);}}

5、测试

编写测试类，发送消息

@Autowired
private KafkaProducer kafkaProducer;@Testpublic void testKafka() throws ExecutionException, InterruptedException {
    kafkaProducer.testSend("forlan_topic","Forlan测试发送Kafka消息"+ LocalDateTime.now());
}

执行测试类，效果如下：

消费者后，会收到消息，如下：

二、遇到问题

1、could not be established. Broker may not be available

[Consumer clientId=consumer-forlan-group-id-1, groupId=forlan-group-id] Connection to node 1001 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.

由于我们的kafka是装在容器内，使用下面这种方式，localhost指的是容器的ip，会有问题

KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092

正确的做法，应该是改为我们宿主机的ip，如下：

KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://192.168.56.100:9092

其实就是把localhost设置为具体的ip，宿主机的ip。

2、Error while fetching metadata with correlation id xxx

[Producer clientId=forlan-client-id-1] Error while fetching metadata with correlation id 3 : {forlan-topic=LEADER_NOT_AVAILABLE}

重启kafka即可