[大数据]springboot整合kafka

简介

Apache kafka是一个分布式的发布-订阅消息系统,能够支撑海量的数据传递,在离线和实时的消息处理业务系统中,kafka都有广泛的应用,kafka将消息持久化到磁盘中,并对消息创建了备份保证数据的安全性,kafka在保证了较高处理的速度的同时,又能保证数据处理的低延迟和数据的零丢失

特性

1. 高吞吐量,低延迟:kafka每秒可以处理几十万消息,延迟最低只有几毫秒,每个主题可以分多个分区,消费组对分区进行消费操作
2. 可扩展性: kafka集群支持热扩展
3. 持久性,可靠性: 消息被持久化到本地磁盘,并且支持数据备份防止数据丢失
4. 容错性: 允许集群中节点失败(若副本数量为n,则允许n-1个节点失败)
5. 高并发: 支持上千个客户端同时读写

概念

Producer

生产者即为数据的发布者,该角色将消息发布到kafka的topic中,broker接收到生产者发送的消息后,broker将该消息追加到当前用于追加数据的segment文件中,生产者发送的消息,存储到一个partition中,生产者也可以指定数据存储的partition

Consumer

消费者可以从broker中读取数据,消费者可以消费多个topic中的数据

Topic

在kafka中,使用一个类别属性来划分数据的所属类,划分数据的这个类成为topic,如果把kafka看成一个数据库,那么topic就可以理解为数据库中的一张表,topic的名字即为表名

Partition

topic中的数据分割为一个或多个partition,每个topic至少有一个partition,每个partition中的数据使用多个segment文件存储,单个partition中的数据是有序的,但若一个topic有多个partition,那么消费时就不能保证消费的顺序,在需要严格保证消息的消费顺序的场景下,需要将topic的partition数量设为1

Partition offset

每条消息都有一个当前partition下唯一的64字节的offset,指明了这条消息的起始位置

Replicas of partition

副本是一个分区的备份,副本不会被消费者消费,副本只用于防止数据丢失,即消费者不会从follow的partition中消费数据,只会从leader的partition中消费数据,副本之间是一主多从的关系

Broker

kafka集群包含一个或多个服务器,服务器节点称为broker,broker中存储topic中的数据,如果某个topic有N个partition,集群有N个broker,那么每个broker存储该topic的一个partition,如果某topic有N个partition,集群却有(N+M)个broker,那么其中N个broker存储该topic的一个partition,剩下的M个broker不存储该topic的partition数据,如有topic有N个partition,集群中broker的数量却少于N个,那么一个broker存储该topic的一个或多个partition数据,在实际的生产环境下,尽量避免这样的情况发生,这种情况容易导致kafka集群数据的不均衡

Leader Follow

follow跟随leader,所有写请求都通过leader路由,数据变更会广播给所有follow,follow与leader保持数据同步,如果leader失效,则会从follow中选举出一个新的leader,当follow与leader挂掉,卡住或者同步太慢,leader会把这个follow从ISR(in sync replicas)列表中删除,重新创建一个follow

zookeeper

zookeeper负责维护和协调broker,当kafka系统中新增了broker或者某个broker发生故障失效时,由zookeeper通知生产者和消费者,生产者和消费者依据zookeeper的broker状态信息与broker协调数据的发布订阅任务

AR(Assigned Replicas)

分区中的所有副本称之AR

ISR(In-Sync Replicas)

所有与leader部分保持一定程度的副本(包括leader副本在内)组成ISR

OSR(Out-of-Sync-Replicas)

与leader副本同步滞后过多的副本

HW(High Watermark)

高水位,标识了一个特定的offset,消费者只能拉取到这个offset之前的消息

LEO(Log End Offset)

日志末端位移,记录了该副本底层日志中下一条消息的位移值,如果LEO=10,标识该副本保存了10条消息,位移范围是[0,9]

使用docker-compose安装zookeeper,kafka,kafka-manager

拉取镜像

# Zookeeper
sudo docker pull wurstmeister/zookeeper
# Kafka
sudo docker pull wurstmeister/kafka
# kafka-manager
sudo docker pull sheepkiller/kafka-manager

创建一个空目录,并且创建一个名为docker-compose.yaml文件,文件内容如下:

services:
    zookeeper:
        image: wurstmeister/zookeeper #镜像
        ports:
            - 2181:2181 #对外暴露端口号
    kafka1:
        image: wurstmeister/kafka
        depends_on:
            - zookeeper
        ports:
            - 9093:9093
        environment:
            KAFKA_ADVERTISED_HOST_NAME: kafka1
            KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181/kafka
            KAFKA_LISTENERS: PLAINTEXT://:9093
            KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://192.168.3.58:9093
            KAFKA_BROKER_ID: 1
        volumes:
            - /var/run/docker.sock:/var/run/docker.sock #挂载位置
    kafka2:
        image: wurstmeister/kafka
        depends_on:
            - zookeeper
        ports:
            - 9094:9094
        environment:
            KAFKA_ADVERTISED_HOST_NAME: kafka2
            KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181/kafka
            KAFKA_LISTENERS: PLAINTEXT://:9094
            KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://192.168.3.58:9094
            KAFKA_BROKER_ID: 2
        volumes:
            - /var/run/docker.sock:/var/run/docker.sock
    kafka3:
        image: wurstmeister/kafka
        depends_on:
            - zookeeper
        ports:
            - 9095:9095
        environment:
            KAFKA_ADVERTISED_HOST_NAME: kafka3
            KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181/kafka
            KAFKA_LISTENERS: PLAINTEXT://:9095
            KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://192.168.3.58:9095
            KAFKA_BROKER_ID: 3
        volumes:
            - /var/run/docker.sock:/var/run/docker.sock
    kafka-manager:
        image: sheepkiller/kafka-manager  # 镜像：开源的web管理kafka集群的界面
        ports:
            - 9000:9000
        environment:
            ZK_HOSTS: zookeeper:2181   # 修改:宿主机IP

参数说明:

KAFKA_ADVERTISED_HOST_NAME：
    广播主机名称，一般用IP指定
KAFKA_ZOOKEEPER_CONNECT：
    Zookeeper连接地址，格式：zoo1：port1,zoo2:port2:/path
KAFKA_LISTENERS：
    Kafka启动所使用的的协议及端口
KAFKA_ADVERTISED_LISTENERS：
    Kafka广播地址及端口，也就是告诉客户端，使用什么地址和端口能连接到Kafka，这个很重要，如果不指定，宿主机以外的客户端将无法连接到Kafka，
    比如我这里因为容器与宿主机做了端口映射，所以广播地址采用的是宿主机的地址及端口，告诉客户端只要连接到宿主机的指定端口就行了
KAFKA_BROKER_ID：
    指定BrokerId，如果不指定，将会自己生成

创建容器并启动:

# 创建并启动容器，可以使用 -f 参数指定docker-compose.yml文件
sudo docker-compose up -d
# 查看状态
sudo docker-compose ps
# 停止并删除容器，可以使用 -f 参数指定docker-compose.yml文件
sudo docker-compose down　

常见指令如下:

//进入容器
docker exec -it 容器id /bin/bash

//创建topic
kafka-topics.sh --zookeeper 192.168.3.58:2181/kafka --create --topic xiaozhuang --partitions 2 --replication-factor 1
    --zookeeper：指定了Kafka所连接的Zookeeper服务地址
    --topic：指定了所要创建主题的名称
    --partitions：指定了分区个数
    --replication-factor：指定了副本因子
    --create：创建主题的动作指令
    
//展示所有主题
kafka-topics.sh --zookeeper 192.168.3.58:2181/kafka --list

//查看主题详情
kafka-topics.sh --zookeeper 192.168.3.58:2181/kafka --describe --topic xiaozhuang

//生产端发送消息到topic为xiaozhuang上
kafka-console-producer.sh --broker-list localhost:9092 --topic xiaozhuang

//启用消费端,监听topic为xiaozhuang的消息
kafka-console-consumer.sh --bootstrap-server 192.168.3.58:9093 --topic xiaozhuang 

Springboot整合kafka

创建springboot项目,引入以下依赖项

<dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka-test</artifactId>
            <scope>test</scope>
        </dependency>
</dependencies>

编写配置文件application.properties

###########【Kafka集群】###########
spring.kafka.bootstrap-servers=localhost:9093,localhost:9094,localhost:9095
###########【初始化生产者配置】###########
# 重试次数
spring.kafka.producer.retries=0
# 应答级别:多少个分区副本备份完成时向生产者发送ack确认(可选0、1、all/-1)
spring.kafka.producer.acks=1
# 批量大小
spring.kafka.producer.batch-size=16384
# 提交延时
spring.kafka.producer.properties.linger.ms=0
# 当生产端积累的消息达到batch-size或接收到消息linger.ms后,生产者就会将消息提交给kafka
# linger.ms为0表示每接收到一条消息就提交给kafka,这时候batch-size其实就没用了

# 生产端缓冲区大小
spring.kafka.producer.buffer-memory = 33554432
# Kafka提供的序列化和反序列化类
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer
# 自定义分区器
# spring.kafka.producer.properties.partitioner.class=com.felix.kafka.producer.CustomizePartitioner
###########【初始化消费者配置】###########
# 默认的消费组ID
spring.kafka.consumer.properties.group.id=defaultConsumerGroup
# 是否自动提交offset
spring.kafka.consumer.enable-auto-commit=true
# 提交offset延时(接收到消息后多久提交offset)
spring.kafka.consumer.auto-commit-interval=1000
# 当kafka中没有初始offset或offset超出范围时将自动重置offset
# earliest:重置为分区中最小的offset;
# latest:重置为分区中最新的offset(消费分区中新产生的数据);
# none:只要有一个分区不存在已提交的offset,就抛出异常;
spring.kafka.consumer.auto-offset-reset=latest
# 消费会话超时时间(超过这个时间consumer没有发送心跳,就会触发rebalance操作)
spring.kafka.consumer.properties.session.timeout.ms=120000
# 消费请求超时时间
spring.kafka.consumer.properties.request.timeout.ms=180000
# Kafka提供的序列化和反序列化类
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer
# 消费端监听的topic不存在时，项目启动会报错(关掉)
spring.kafka.listener.missing-topics-fatal=false
# 设置批量消费
# spring.kafka.listener.type=batch
# 批量消费每次最多消费多少条消息
# spring.kafka.consumer.max-poll-records=50

创建ProducerController.java,编写发送消息接口:

@RestController
public class ProducerController {
    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;

    @GetMapping("/api/v1/send/msg/{msg}")
    public void sendMsg(@PathVariable("msg") String msg) {
        String topic = "xiaozhuang";
        kafkaTemplate.send(topic, msg);
    }
}

创建消费端Consumer.java

@Component
public class Consumer {
    @KafkaListener(topics = {"xiaozhuang"})
    public void getMsg(ConsumerRecord<String, Object> record) {
        System.out.println("topic: " + record.topic());
        System.out.println("partition: " + record.partition());
        System.out.println("offset: " + record.offset());
        System.out.println("获取到record的值: " + record.value());
    }
}

启动项目,完成测试....