SpringBoot集成系列--Kafka_could not be established. broker may not be availa

一、代码

1、添加依赖

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

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>
1
2
3
4

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
1
2
3
4

参数说明：

• 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
1
2
3
4
5
6

参数说明：

• 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、创建生产者

设置发送消息服务类

@Component
public class KafkaProducer {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    public void testSend(String topic, String message) throws ExecutionException, InterruptedException {
        SendResult<String, String> stringStringSendResult = kafkaTemplate.send(topic, message).get();
        System.out.println(stringStringSendResult);
    }
}
1
2
3
4
5
6
7
8
9
10
11

4、创建消费者

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;

@Service
public class KafkaConsumer {
	@KafkaListener(topics = "forlan_topic")
	public void listen(String message) {
		System.out.println("Kafka收到消息：" + message);
	}
}
1
2
3
4
5
6
7
8
9
10

5、测试

编写测试类，发送消息

@Autowired
private KafkaProducer kafkaProducer;

@Test
public void testKafka() throws ExecutionException, InterruptedException {
	kafkaProducer.testSend("forlan_topic", "Forlan测试发送Kafka消息"+ LocalDateTime.now());
}
1
2
3
4
5
6
7

执行测试类，效果如下：

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

二、遇到问题

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.
1

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

KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
1

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

KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://192.168.56.100:9092
1

其实就是把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}
1

重启kafka即可