Apache Kafka详解

一、消息队列Message Queue

image.png

两种模式

  1. 点对点模式
    点对点模式是一个基于拉取或轮询的消息传送模型,由消费者主动拉取数据,客户端需要实时开启一个线程监控队列中是否有数据。
  2. 发布/订阅模式
    发布/订阅模式是一个基于推送的消息传送模型,由MQ主动推送消息给所有订阅者,即使当前订阅者不可用。

优点

  • 解耦
    允许你独立的扩展或修改两边的处理过程,只要确保它们遵循同样的接口约束。
  • 冗余
    MQ把数据进行持久化直到它们已经被完全处理,通过这一方式规避了数据丢失风险。
  • 扩展
    因为MQ解耦了你的处理过程,所以增大消息入队和处理的频率是很容易的,只要另外增加处理过程即可,提高了灵活性和峰值处理能力。
  • 可恢复性
    消息队列降低了进程间的耦合度,所以即使一个处理消息的进程挂掉,加入队列中的消息仍然可以在系统恢复后被处理。
  • 顺序保证
    队列本身就遵循FIFO的原则,保证了数据的处理顺序。
  • 缓冲
    有助于控制和优化数据流经过系统的速度,解决生产者和消费者处理速度不一致的问题。
  • 异步通信
    MQ提供了异步处理机制,允许用户把消息放入队列但不立即处理它,直到需要时再去处理。

二、Kafka简介

  1. Apache Kafka是一个开源消息系统,由Scala写成。是由Apache软件基金会开发的一个开源消息系统项目。
  2. Kafka最初是由LinkedIn公司开发,并于2011年初开源。2012年10月从Apache Incubator毕业。该项目的目标是为处理实时数据提供一个统一、高通量、低等待的平台。
  3. Kafka是一个Distributed Message Queue。Kafka对消息保存时根据Topic进行归类,发送消息者称为Producer,消息接收者称为Consumer,此外Kafka集群有多个Kafka实例组成,每个实例称为Broker。
  4. Kafka需要依赖于Zookeeper保存一些Meta信息,来保证系统可用性。
  5. 在流式计算中,Kafka一般用来缓存数据,Storm通过消费Kafka的数据进行计算。

三、Kafka架构

image.png
No. 组件 说明
1 Producer 消息生产者,向Kafka中发消息的客户端。
2 Consumer 消息消费者,从Kafka中取消息的客户端。
3 Broker 一台Kafka服务器就是一个Broker,一个集群由多个Broker组成,一个Broker可以容纳多个Topic。
4 Topic 属于特定类别的消息流称为Topic, 数据存储在Topic中,可以理解为一个队列。
5 Partition 为了实现负载均衡和高并发,一个非常大的Topic可以通过分为多个Partition分布到多个Broker上,每个Partition都是一个有序的队列。
6 Consumer Group(CG) 这是Kafka用来实现一个消息的广播(发给所有Consumer)和单播(发给任意一个Consumer)的手段。一个Topic可以有多个CG,Topic中的消息会复制(不是真的复制,是概念上的)到所有的CG,但每个Partition只会把消息发给该CG中的一个Consumer。如果需要实现广播,只要每个Consumer都有一个独立的CG即可,需要实现单播,只要所有Consumer都在同一个CG即可。用CG还可以将Consumer进行自由分组而不需要重复发送消息到不同的Topic。
7 Offset Partition中的每条消息都会被分配一个有序的id(Offset)。Kafka只保证每个Partition中的顺序,不保证多个Partition的顺序。
8 Leader 负责给定Partition的所有读取和写入的节点。
9 Follower 跟随Leader指令的节点被称为Follower。 如果Leader节点宕机,其中一个Follower将通过选举自动成为新的Leader。

四、Kafka工作流程

Producer生产过程

  1. 写入方式
    Producer采用Push模式将消息发布到Broker,每条消息都被追加(append)到Partition中,属于顺序写磁盘(顺序写磁盘效率比随机写内存要高,保障Kafka吞吐率)。
  2. 分区(Partition)
    消息发送时都被发送到一个topic,其本质就是一个目录,而topic是由一些Partition Logs(分区日志)组成,其组织结构如下图所示:


    image.png

    image.png

    我们可以看到,每个Partition中的消息都是有序的,生产的消息被不断追加到Partition log上,其中的每一个消息都被赋予了一个唯一的offset值。

    1. 分区的原因
      1. 负载均衡,方便在集群中扩展,每个Partition可以通过调整以适应它所在的机器,而一个Topic又可以有多个Partition组成,因此整个集群就可以适应任意大小的数据了。
      2. 可以提高并发,因为可以以Partition为单位读写了。
    2. 分区的原则
      1. 已指定Partition,则直接使用该Partition。
      2. 未指定Partition但指定了Key,则通过对Key进行哈希计算得出一个Partition。
      3. Partition和Key都未指定,则轮询选出一个Partition。
  3. 副本(Replication)
    同一个Partition可能会有多个Replication(对应server.properties配置中的default.replication.factor=N)。没有Replication的情况下,一旦Broker宕机,其上所有Partition的数据都不可被Consumer消费,同时Producer也不能再将数据存于其上的Partition。引入Replication之后,同一个Partition可能会有多个Replication,而这时需要在这些Replication之间选出一个Leader,Producer和Consumer只与这个Leader交互,其它Replication作为Follower从Leader中复制数据。
  4. 写入流程


    image.png

Broker保存过程

  1. 存储方式
    物理上把Topic分成一个或多个Partition(对应server.properties配置中的num.partitions=3),每个Partition物理上对应一个文件夹(该文件夹存储该Partition的所有消息和索引文件),如下:
[atguigu@hadoop102 logs]$ ll
drwxrwxr-x. 2 atguigu atguigu  4096 8月   6 14:37 first-0
drwxrwxr-x. 2 atguigu atguigu  4096 8月   6 14:35 first-1
drwxrwxr-x. 2 atguigu atguigu  4096 8月   6 14:37 first-2
[atguigu@hadoop102 logs]$ cd first-0
[atguigu@hadoop102 first-0]$ ll
-rw-rw-r–. 1 atguigu atguigu 10485760 8月   6 14:33 00000000000000000000.index
-rw-rw-r–. 1 atguigu atguigu      219 8月   6 15:07 00000000000000000000.log
-rw-rw-r–. 1 atguigu atguigu 10485756 8月   6 14:33 00000000000000000000.timeindex
-rw-rw-r–. 1 atguigu atguigu        8 8月   6 14:37 leader-epoch-checkpoint
  1. 存储策略
    无论消息是否被消费,Kafka都会保留所有消息。有两种策略可以删除旧数据:

    1. 基于时间:log.retention.hours=168
    2. 基于大小:log.retention.bytes=1073741824

    需要注意的是,因为Kafka读取特定消息的时间复杂度为O(1),即与文件大小无关,所以这里删除过期文件与提高Kafka性能无关。

  2. Zookeeper存储结构


    image.png

    注意:只有Broker和Consumer在Zookeeper中注册,Producer不在Zookeeper中注册。

Consumer消费过程

  1. 消费者组


    image.png

    Consumer是以Consumer Group的方式工作,由一个或者多个Consumer组成一个Group,共同消费一个Topic。每个Partition在同一时间只能由Group中的一个Consumer读取,但是多个Group可以同时消费同一个Partition。在图中,有一个由三个Consumer组成的Group,有一个Consumer读取Topic中的两个Partition,另外两个Consumer分别读取一个Partition。某个Consumer读取某个Partition,也可以叫做某个Consumer是某个Partition的拥有者。
    在这种情况下,Consumer可以通过水平扩展的方式同时读取大量的消息。另外,如果一个Consumer失败了,那么Group中其它的Consumer会自动负载均衡读取之前失败的Consumer读取的Partition。

  2. 消费方式
    Consumer采用Pull模式从Broker中读取数据。
    Push模式很难适应消费速率不同的Consumer,因为消息发送速率是由Broker决定的。它的目标是尽可能以最快速度传递消息,但是这样很容易造成Consumer来不及处理消息,典型的表现就是拒绝服务以及网络拥塞。而Pull模式则可以根据Consumer的消费能力以适当的速率消费消息。
    对于Kafka而言,Pull模式更合适,它可简化Broker的设计,Consumer可自主控制消费消息的速率,同时Consumer可以自己控制消费方式——即可以批量消费也可以逐条消费,同时还能选择不同的提交方式从而实现不同的传输语义。
    Pull模式不足之处是,如果Kafka没有数据,消费者可能会陷入循环中,一直等待数据到达。为了避免这种情况,可以Pull请求中设置参数,允许Consumer请求在等待数据到达的“长轮询”中进行阻塞(并且可选地等待到给定的字节数,以确保大的传输大小)。

五、Kafka案例

生产者

package com.fas.kafka.service;

import org.apache.log4j.Logger;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;

import com.fas.kafka.util.KafkaUtil;

@Service
@Log4j
public class KafkaProducerService {
    private KafkaTemplate<String, String> kafkaTemplate = KafkaUtil.getKafkaTemplate();

    public void send(String topic, String data) {
        ListenableFuture<SendResult<String, String>> listenableFuture = kafkaTemplate.send(topic, data);
        listenableFuture.addCallback(result -> log.info("Send data to kafka [success] - " + topic + " - " + data), result -> log.info("Send data to kafka [error] - " + topic + " - " + data));
    }
}

消费者

package com.fas.kafka.consumer;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.listener.MessageListener;

import com.alibaba.fastjson.JSONObject;
import com.fas.kafka.entity.DeviceOnlineData;

import lombok.extern.log4j.Log4j;

@Log4j
public class KafkaConsumer implements MessageListener<String, String> {
    public static final BlockingQueue<DeviceOnlineData> DATA_ACCESS_BLOCKING_QUEUE = new LinkedBlockingQueue<>();
    public static final String KAFKA_DATA_ACCESS_TOPIC = PropertiesUtil.KAFKA_PROPERTIES.getProperty("kafka.data_access.topic");
    public static final String KAFKA_TEST_TOPIC = PropertiesUtil.KAFKA_PROPERTIES.getProperty("kafka.test.topic");

    @Override
    public void onMessage(ConsumerRecord<String, String> consumerRecord) {
        String topic = consumerRecord.topic();
        log.info("topic: " + topic);
        String value = consumerRecord.value();
        log.info("value: " + value);

        if (KAFKA_DATA_ACCESS_TOPIC.equals(topic)) {
            JSONObject.parseArray(value, DeviceOnlineData.class).stream().forEach(deviceOnlineData -> {
                try {
                    DATA_ACCESS_BLOCKING_QUEUE.put(deviceOnlineData);
                    log.info("Put into blocking queue");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        } else if (KAFKA_TEST_TOPIC.equals(topic)) {
            log.info("Test success");
        } else {
            log.info("Unknown topic");
        }
    }
}

六、常用命令

  1. 查看所有Topic
    ./bin/kafka-topics.sh --list --zookeeper zk01:2181
  2. 创建Topic
    ./bin/kafka-topics.sh --create --zookeeper zk01:2181 --replication-factor 1 --partitions 1 --topic test
  3. 删除Topic
    ./bin/kafka-topics.sh --delete --zookeeper zk01:2181 --topic test
    需要server.properties中设置delete.topic.enable=true,否则只是标记删除,或者直接重启
  4. 通过Shell命令发送消息
    ./bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test
  5. 通过Shell消费消息
    ./bin/kafka-console-consumer.sh --zookeeper zk01:2181 --from-beginning --topic test
  6. 查看消费位置
    ./bin/kafka-run-class.sh kafka.tools.ConsumerOffsetChecker --zookeeper zk01:2181 --group testGroup
  7. 查看某个Topic的详情
    ./bin/kafka-topics.sh --topic test --describe --zookeeper zk01:2181
  8. 对分区数进行修改
    ./bin/kafka-topics.sh --zookeeper zk01 --alter --partitions 15 --topic utopic
最后编辑于
©著作权归作者所有,转载或内容合作请联系作者
  • 序言:七十年代末,一起剥皮案震惊了整个滨河市,随后出现的几起案子,更是在滨河造成了极大的恐慌,老刑警刘岩,带你破解...
    沈念sama阅读 211,884评论 6 492
  • 序言:滨河连续发生了三起死亡事件,死亡现场离奇诡异,居然都是意外死亡,警方通过查阅死者的电脑和手机,发现死者居然都...
    沈念sama阅读 90,347评论 3 385
  • 文/潘晓璐 我一进店门,熙熙楼的掌柜王于贵愁眉苦脸地迎上来,“玉大人,你说我怎么就摊上这事。” “怎么了?”我有些...
    开封第一讲书人阅读 157,435评论 0 348
  • 文/不坏的土叔 我叫张陵,是天一观的道长。 经常有香客问我,道长,这世上最难降的妖魔是什么? 我笑而不...
    开封第一讲书人阅读 56,509评论 1 284
  • 正文 为了忘掉前任,我火速办了婚礼,结果婚礼上,老公的妹妹穿的比我还像新娘。我一直安慰自己,他们只是感情好,可当我...
    茶点故事阅读 65,611评论 6 386
  • 文/花漫 我一把揭开白布。 她就那样静静地躺着,像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上,一...
    开封第一讲书人阅读 49,837评论 1 290
  • 那天,我揣着相机与录音,去河边找鬼。 笑死,一个胖子当着我的面吹牛,可吹牛的内容都是我干的。 我是一名探鬼主播,决...
    沈念sama阅读 38,987评论 3 408
  • 文/苍兰香墨 我猛地睁开眼,长吁一口气:“原来是场噩梦啊……” “哼!你这毒妇竟也来了?” 一声冷哼从身侧响起,我...
    开封第一讲书人阅读 37,730评论 0 267
  • 序言:老挝万荣一对情侣失踪,失踪者是张志新(化名)和其女友刘颖,没想到半个月后,有当地人在树林里发现了一具尸体,经...
    沈念sama阅读 44,194评论 1 303
  • 正文 独居荒郊野岭守林人离奇死亡,尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 36,525评论 2 327
  • 正文 我和宋清朗相恋三年,在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 38,664评论 1 340
  • 序言:一个原本活蹦乱跳的男人离奇死亡,死状恐怖,灵堂内的尸体忽然破棺而出,到底是诈尸还是另有隐情,我是刑警宁泽,带...
    沈念sama阅读 34,334评论 4 330
  • 正文 年R本政府宣布,位于F岛的核电站,受9级特大地震影响,放射性物质发生泄漏。R本人自食恶果不足惜,却给世界环境...
    茶点故事阅读 39,944评论 3 313
  • 文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹,春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 30,764评论 0 21
  • 文/苍兰香墨 我抬头看了看天上的太阳。三九已至,却和暖如春,着一层夹袄步出监牢的瞬间,已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 31,997评论 1 266
  • 我被黑心中介骗来泰国打工, 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留,地道东北人。 一个月前我还...
    沈念sama阅读 46,389评论 2 360
  • 正文 我出身青楼,却偏偏与公主长得像,于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子,可洞房花烛夜当晚...
    茶点故事阅读 43,554评论 2 349

推荐阅读更多精彩内容