当前位置：首页 > article >正文

互联网全景消息(9)之Kafka深度剖析(上)

article 2025/1/9 19:04:27

一、应用场景

1.1 kafka场景

Kafka最初是LinkedIn公司采用Scala 语言开发，现在已经捐赠给了Apache基金会。目前Kafka已经定位为一个分布式流式处理平台，它以高吞吐、可持久化、可水平扩展、支持流处理等多种特性而被广泛应用。

Apache Kafka能够支撑海量数据的数据传递，在离线和实时的消息处理业务系统中，Kafka都有广泛的应用。

日志收集：收集各种服务的log，同Kafka以统一接口服务的方式开放给各种consumer，例如：Hadoop、Hbase、Solr等。
消息系统：解耦和生产者和消费者、缓存消息等‘
用户活动跟踪：Kafka经常被用来记录web用户或者app用户的各种活动，如浏览网页、搜索、点击等活动，这些活动被哥哥服务器发布到kafka的topic中，然后订阅者通过订阅这些topic来做实时的监控分析，或者装在到Hadoop、数据仓库中做离线分析和挖掘；
运营指标：Kafka也经常用来记录运营监控数据，包括收集各种分布式应用的数据，生产各种操作的集中反馈，比如报警和报告；
流式处理：比如spark streaming和storm；

1.2 kafka特性

kafka以高吞吐著称，主要有以下特性：

高吞吐、低延迟：kafka每秒可以处理几十万条消息，它的低延迟只有几毫秒；
可扩展：kafka集群支持热扩展；
持久性、可靠性：消息被持久化到本地磁盘，并且支持数据备份防止数据丢失；
容错性：允许集群中节点失败(若副本数量为n，则允许n-1个节点失败)
高并发：支持数千个客户端同时读写；

1.3 消息对比

如果普通的业务消息解耦，消息传输，rabbitMq是首选，它足够简单，管理方便，性能够用；
如果在上诉，日志、消息收集、访问记录等高吞吐，实时性场景下，推荐kafka，它基于分布式，扩容便捷；
如果很重的业务，要做到极高的可靠性，考虑RocketMQ，但是它太重，需要你有足够多的了解；

二、基础组件

2.1 角色

broker：节点，就是你看到的机器；
producer：生产者，发送消息；
consumer：消费者，消费消息；
zookeeper：消息中心，记录kafka的各种信息的地方；
controller：其中的一个broker，作为leader身份来负责管理整个集群，如果挂掉，借助zk重新选主；

2.2 逻辑组件

topic：主题，一个消息的道通；
partition：分区，每个主题可以有多个分区分担数据的传递，多条路并行，吞吐量大；
Replicas：副本，每个分区可以设置多个副本，副本之间数据一致。相当于备份；
leader&follower：主从，上面的这些副本里有1个身份为leader，其他的为follower。leader处理partition的所有读写请求；

2.3 副本集合

AR：所有副本的统称，AR = ISR + OSR;
ISR: (In-sync-Repicas)，同步中的副本，可以参与leader选主，一旦落后太多(数量滞后和时间滞后两个维度)会被提到OSR；
OSR：(out-sync-Repicas)提出同步的副本，一直追赶leader，追上之后就会进入到ISR中；

2.4 消息标记

offset：偏移量，消息消费到那一条了，每个消费者都有自己的偏移量；
HW（high watermark）:副本的高水印值，客户端最多能消费到的位置，HW值为8，代表offset为[0,8]得九条消息都可以被消费到，它们是对消费者可见的，而[9，,12]这4条消息由于未提交，对消费者是不可见的；
LEO(log end offset)：日志末端位移，代表日志文件中下一条待写入消息的offset，这个offset上实际是没有消息的，不管是leader副本还是follower副本，都有这个值；

比如在副本数等于3的情况下，消息发送到leader A之后会更新LEO的值，Follower B和Follower C也会实时拉取Leader A中的消息来更新自己，HW就表示A/B/C三者同时达到的日志位移，也就是A、B、C三者中LEO最小的那个值。由于B、C拉取A消息之间延时问题，所以HW一般会小于LEO，即 LEO >= HW.

三、架构探索

3.1 集群搭建

推荐使用docker-compose一键启动，yml配置如下：

#docker-compose.yml
#注意hostname问题，ip地址：47.94.171.130，换成你自己服务器的
#docker-compose up -d 启动
version: '3'
services:
    zookeeper:
        image: zookeeper:3.4.13

    kafka-1:
        container_name: kafka-1
        image: wurstmeister/kafka:2.12-2.2.2
        ports:
            - 10903:9092
        environment:
            KAFKA_BROKER_ID: 1
            HOST_IP: 47.94.171.130
            KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
            #docker部署必须设置外部可访问ip和端口，否则注册进zk的地址将不可达造成外部无法连接
            KAFKA_ADVERTISED_HOST_NAME: 47.94.171.130
            KAFKA_ADVERTISED_PORT: 10903
        volumes:
            - /etc/localtime:/etc/localtime
        depends_on:
            - zookeeper
    kafka-2:
        container_name: kafka-2
        image: wurstmeister/kafka:2.12-2.2.2
        ports:
            - 10904:9092
        environment:
            KAFKA_BROKER_ID: 2
            HOST_IP: 47.94.171.130
            KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
            KAFKA_ADVERTISED_HOST_NAME: 47.94.171.130
            KAFKA_ADVERTISED_PORT: 10904
        volumes:
            - /etc/localtime:/etc/localtime
        depends_on:
            - zookeeper

3.2 创建主题

[root@iZ2zei8v8whhk4xi8iamv9Z ~]# docker exec -it kafka-1 sh
/ # cd /opt/kafka/bin/
/opt/kafka_2.12-2.2.2/bin # 
/opt/kafka_2.12-2.2.2/bin # kafka-topics.sh --zookeeper zookeeper:2181 --create --topic test --partitions 2 --replication-factor 1
Created topic test.
/opt/kafka_2.12-2.2.2/bin #

3.3 查看主题

kafka-topics.sh --zookeeper zookeeper:2181 --list

3.4 主题详情

kafka-topics.sh --zookeeper zookeeper:2181 --describe --topic test

#分析输出： #注意格式：Partition: （分区号，从0开始），Leader、 Replicas、Isr: （机器号，也就是启动时配置的broker_id）

3.5 消息收发

#使用docker连接任意集群中的一个容器
docker exec -it kafka-1 sh
#进入kafka的容器内目录
cd /opt/kafka/bin
#客户端监听
./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test
#另起一个终端，验证发送
./kafka-console-producer.sh --broker-list localhost:9092 --topic test

3.6 分组消费

#启动两个consumer时，如果不指定group信息，消息被广播 #指定相同的group，让多个消费者分工消费

./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test -- group aaa

#结果：在发送方，连续发送 1-4 ，4条消息，同一group下的两台consumer交替消费，并发执行