文章目录
- 一、代码
- 1、添加依赖
- 2、配置kafka
- 3、创建生产者
- 4、创建消费者
- 5、测试
- 二、遇到问题
- 1、could not be established. Broker may not be available
- 2、Error while fetching metadata with correlation id xxx
一、代码
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、创建生产者
设置发送消息服务类
@Component
public class KafkaProducer {@Autowiredprivate 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);}
}
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);}
}
5、测试
编写测试类，发送消息
@Autowired
private KafkaProducer kafkaProducer;@Test
public 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即可
