java——spring boot集成kafka——kafka集群中controller的作用

Posted 2023-04-02 小白龙白龙马

 

 

 

 

Kafka集群中的broker在zk中创建临时序号节点，序号最⼩的节点（最先创建的节点）将作为集群的controller，负责管理整个集群中的所有分区和副本的状态：

 

 

1、当某个分区的leader副本出现故障时，由控制器负责为该分区选举新的leader副本。

 

2、当检测到某个分区的ISR集合发⽣变化时，由控制器负责通知所有broker更新其元数据信息。

 

3、当使⽤kafka-topics.sh脚本为某个topic增加分区数量时，同样还是由控制器负责让新分区被其他节点感知到。

 

spring boot集成kafka开发，接收kafka消息，Java

在这篇文章（1）基础上：

kafka连接zookeeper配置搭建环境、安装和启动，Windows环境下_Zhang Phil-CSDN博客kafka，分布式集群架构下，高性能的流式事件数据（主要是消息）集成、发布（生产）和订阅（分发、消费）组件（中间件）。kafka依赖zooeeper（数据后端），这里有Windows下安装配置启动zookeeper的 文章（1）：Windows 10环境zookeeper单机伪集群部署和配置_Zhang Phil-CSDN博客Windows 10环境zookeeper单机伪集群部署和配置1，首先到zookeeper项目主页地址下载项目包，https://archive.apache.org/disthttps://blog.csdn.net/zhangphil/article/details/123086790

实现一个简单功能：通过kafka控制台发送消息，然后在spring boot编程实现的java代码程序里面接收消息。

按照文章（1）启动zookeeper和kafka，使用命令：

kafka-topics.bat --list --bootstrap-server 127.0.0.1:9092

查看Windows环境下，列出主机127.0.0.1上kafka的主题topic。本例选择其中一个topic主题，使用kafka命令在控制台发送消息，接收方则用spring框架里面的java程序实现。

下面开始实现spring框架的kafka消息接收java代码。

（1）在application.properties里面配置spring环境中的kafka参数：

server.port=7999
spring.kafka.bootstrap-servers=127.0.0.1:9092
spring.kafka.consumer.bootstrap-servers=127.0.0.1:9092
spring.kafka.consumer.group-id=zhangphil-group
spring.kafka.consumer.key-deserializer=org.springframework.kafka.support.serializer.StringOrBytesSerializer
spring.kafka.consumer.value-deserializer=org.springframework.kafka.support.serializer.JsonSerializer
spring.kafka.producer.bootstrap-servers=127.0.0.1:9092
spring.kafka.producer.key-serializer=org.springframework.kafka.support.serializer.StringOrBytesSerializer
spring.kafka.producer.value-serializer=org.springframework.kafka.support.serializer.JsonSerializer

(2)添加注解@EnableKafka， 打开spring application的kafka配置：

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.kafka.annotation.EnableKafka;

@EnableKafka
@SpringBootApplication
public class SpringKafkaApplication 
    public static void main(String[] args) 
        SpringApplication.run(SpringKafkaApplication.class, args);
    

（3）写一个kafka消息接收器：

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

@Component
public class MyKafka 
    @KafkaListener(topics = "zhangphil_demo", groupId = "$spring.kafka.consumer.group-id")
    public void msg_listener(ConsumerRecord<?, ?> record) 
        System.out.println(record.topic() + "," + record.offset() + "," + record.value());
        System.out.println(record);
    

以上java程序写完后，即可启动spring application。

然后在kafka控制台，用命令:

kafka-console-producer.bat --topic zhangphil_demo --bootstrap-server localhost:9092

在主题topic为zhangphil_demo里面发送测试消息 hello,world!2022,运行输出如图：