从零开始学习Kafka-白红宇

简介

kafka是一个分布式消息队列。具有高性能、持久化、多副本备份、横向扩展能力。生产者往队列里写消息，消费者从队列里取消息进行业务逻辑。一般在架构设计中起到解耦、削峰、异步处理的作用。

Kafka核心组件-intsmaze

Topic：消息根据Topic进行归类，可以理解为一个队里。

Producer：消息生产者，就是向kafka broker发消息的客户端。

Consumer：消息消费者，向kafka broker取消息的客户端。

broker：每个kafka实例(server)，一台kafka服务器就是一个broker，一个集群由多个broker组成，一个broker可以容纳多个topic。

Zookeeper：依赖集群保存meta信息。大家先看kafka的介绍或者教程啊，上来都显示一堆长篇大论，各自文字图片，看着很懵逼，头晕。搞程序的，要让ta跑起来，再针对可运行的成果，慢慢了解ta。所以本文会由浅入深，先实践后理论，结合实践讲理论。

Kafka安装配置

下载

wget http://mirror.bit.edu.cn/apache/kafka/2.2.0/kafka_2.11-2.2.0.tgz复制代码

解压

tar -zxvf kafka_2.11-2.2.0.tgz 复制代码

修改 kafka-server 的配置文件

cd kafka_2.11-2.2.0 vim  config/server.properties复制代码

修改其中的：

# The id of the broker. This must be set to a unique integer for each broker.broker.id=1# A comma separated list of directories under which to store log fileslog.dirs=/data/kafka-logs复制代码

启动zk【默认端口2181】

bin/zookeeper-server-start.sh config/zookeeper.properties复制代码

启动Kafka

使用 kafka-server-start.sh 启动 kafka 服务：

bin/kafka-server-start.sh config/server.properties复制代码

测试使用

创建 topic

使用 kafka-topics.sh 创建单分区单副本的 topic demo：

bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic demo复制代码

查看 topic 列表：

bin/kafka-topics.sh --list --zookeeper localhost:2181复制代码

发送消息【生产者】

使用 kafka-console-producer.sh 发送消息：

bin/kafka-console-producer.sh --broker-list localhost:9092 --topic demo复制代码

读取消息【消费者】

使用 kafka-console-consumer.sh 接收消息并在终端打印：

 bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic demo --from-beginning复制代码

注意不要使用

bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning，高版本已经不支持

查看描述 topics 信息

bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic demo复制代码

[root@localhost kafka_2.11-2.2.0]# bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic demoTopic:demo      PartitionCount:1        ReplicationFactor:1     Configs:        Topic: demo     Partition: 0    Leader: 1       Replicas: 1     Isr: 1复制代码

第一行给出了所有分区的摘要，每个附加行给出了关于一个分区的信息。由于我们只有一个分区，所以只有一行。

“Leader”: 是负责给定分区的所有读取和写入的节点。每个节点将成为分区随机选择部分的领导者。

“Replicas”: 是复制此分区日志的节点列表，无论它们是否是领导者，或者即使他们当前处于活动状态。

“Isr”: 是一组“同步”副本。这是复制品列表的子集，当前活着并被引导到领导者

扩展-集群配置

Kafka 支持两种模式的集群搭建：可以在单机上运行多个 broker 实例来实现集群，也可在多台机器上搭建集群，下面介绍下如何实现单机多 broker 实例集群，其实很简单，只需要如下配置即可。

单机多broker 集群配置

利用单节点部署多个 broker。不同的 broker 设置不同的 id，监听端口及日志目录。例如：

cp config/server.properties config/server-2.propertiesvi config/server-2.properties复制代码

修改内容：

broker.id=2listeners = PLAINTEXT://127.0.0.1:9093log.dirs=/data/kafka-logs2复制代码

同样，配置第三个broker：

cp config/server-2.properties config/server-3.propertiesvi config/server-3.properties复制代码

修改内容：

broker.id=2listeners = PLAINTEXT://127.0.0.1:9093log.dirs=/data/kafka-logs2复制代码

listeners 申明此kafka服务器需要监听的端口号，默认会使用localhost的地址，如果是在远程服务器上运行则必须配置，例如： listeners=PLAINTEXT:// 192.168.180.128:9092 并确保服务器的9092端口能够访问

启动2/3 borker

bin/kafka-server-start.sh config/server-2.properties &bin/kafka-server-start.sh config/server-3.properties &复制代码

至此，单机多broker实例的集群配置完毕。

扩展-多机多borker集群

分别在多个节点按上述方式安装 Kafka，配置启动多个 Zookeeper 实例。

假设三台机器 IP 地址是： 192.168.153.135， 192.168.153.136， 192.168.153.137

分别配置多个机器上的 Kafka 服务，设置不同的 broker id，zookeeper.connect 设置如下:

config/server.properties里面的 zookeeper.connect

zookeeper.connect=192.168.153.135:2181,192.168.153.136:2181,192.168.153.137:2181复制代码

使用 Kafka Connect 来导入/导出数据

从控制台写入数据并将其写回控制台是一个方便的起点，但您可能想要使用其他来源的数据或将数据从 Kafka 导出到其他系统。对于许多系统，您可以使用 Kafka Connect 来导入或导出数据，而不必编写自定义集成代码。

Kafka Connect 是 Kafka 包含的一个工具，可以将数据导入和导出到 Kafka。它是一个可扩展的工具，运行连接器，实现与外部系统交互的自定义逻辑。在这个快速入门中，我们将看到如何使用简单的连接器运行 Kafka Connect，这些连接器将数据从文件导入到 Kafka topic，并将数据从 Kafka topic 导出到文件。

参考：

http://www.54tianzhisheng.cn/2018/01/04/Kafka/ 复制代码

http://kafka.apache.org/10/documentation/streams/quickstart复制代码

http://kafka.apache.org/20/documentation.html#quickstart复制代码

代码测试

准备测试kafka

cp config/server.properties config/server-idea.propertiesvi config/server-idea.properties broker.id=999listeners = PLAINTEXT://192.168.1.177:9999log.dirs=/data/kafka-logs-999复制代码

192.168.1.177为kafka所在机器的ip地址，9999端口号是对外提供的端口，下文会使用到

Springboot 发送消息、接受消息源码

很简单的一个小demo，可以直接拷贝使用。

KafkaApplication.java:

import org.apache.kafka.clients.consumer.ConsumerRecord;import org.apache.kafka.clients.producer.KafkaProducer;import org.springframework.boot.SpringApplication;import org.springframework.boot.autoconfigure.SpringBootApplication;import org.springframework.context.ConfigurableApplicationContext;import org.springframework.kafka.annotation.EnableKafka;import org.springframework.kafka.annotation.KafkaListener;import org.springframework.kafka.core.KafkaTemplate;@SpringBootApplicationpublic class KafkaApplication {    public static void main(String[] args) {        ConfigurableApplicationContext context = SpringApplication.run(KafkaApplication.class, args);        KafkaTemplate kafkaTemplate = context.getBean(KafkaTemplate.class);        for (int i = 0; i < 10; i++) {            //调用消息发送类中的消息发送方法            kafkaTemplate.send("mytopic", System.currentTimeMillis() + "发送" + i);            try {                Thread.sleep(3000);            } catch (InterruptedException e) {                e.printStackTrace();            }        }    }    @KafkaListener(topics = {
   "mytopic"},groupId = "halburt-demo2")    public void consumer1(String message) {        System.out.println("consumer1收到消息:" + message);    }    @KafkaListener(topics = {
   "mytopic"} ,groupId = "halburt-demo")    public void consumer2(ConsumerRecord
     record) {        System.out.println("consumer2收到消息");        System.out.println("    topic" + record.topic());        System.out.println("    key:" + record.key());        System.out.println("    value:"+record.value());    }}复制代码

application.yml:

server:  port: 8090spring:  kafka:    consumer:      auto-commit-interval: 100      bootstrap-servers: 192.168.1.177:9999      enable-auto-commit: true      group-id: halburt-demo      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer      max-poll-records: 1      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer    listener:      concurrency: 5    producer:      bootstrap-servers: 192.168.1.177:9999      key-serializer: org.apache.kafka.common.serialization.StringSerializer      value-serializer: org.apache.kafka.common.serialization.StringSerializer复制代码

192.168.1.177:9999即为kafka的配置文件中配置

pom.xml依赖:

依赖版本：

spring-boot.version：2.1.3.RELEASE spring-kafka.version：2.2.0.RELEASE

【此处有坑】此处依赖版本可以不用这2个版本，但是一定要注意springboot和kafka的版本对应


            
                 
      
       org.springframework.kafka
                  
      
       spring-kafka
                  
      
       2.2.0.RELEASE
              
                 
                 
      
       org.springframework.boot
                  
      
       spring-boot-starter-web
                  
      
       2.1.3.RELEASE
              
         
    复制代码

启动kafka并run Application.java

bin/kafka-server-start.sh config/server-idea.properties &复制代码

已经启动了zk，此处不用再启动，如果未启动，需要启动zk。

cd /home/hd/kafka_2.11-2.2.0/bin/zookeeper-server-start.sh config/zookeeper.properties&复制代码

kafka启动成功之后，run Application，会看到日志如下：

已经接收到消息了。

如果你是跟着本文从头开始的，一定注意此处有个坑

如果你是从头开始跟这个本文学习的，那么你直接启动的话，会发现消息发出去了，但是没有接收到。我也是查了好久，看了很多教程，别人都行我就不行。如果你的zk有其他的topic节点的话，会收不到消息，直接上解决方案：删除所有的zk节点。怎么删除？

上码：

/** * zookeeper znode递归删除节点 * @author Halburt * */public class DeleteZkNode {    //zookeeper的地址     private static final String connectString = "192.168.1.177:2181";    private static final int sessionTimeout = 2000;    private static ZooKeeper zookeeper = null;    /**     * main函数     * @param args     * @throws Exception     */    public static void main(String[] args) throws Exception {        //调用rmr,删除所有目录        rmr("/");    }    /**     * 递归删除 因为zookeeper只允许删除叶子节点，如果要删除非叶子节点，只能使用递归     * @param path     * @throws IOException     */    public static void rmr(String path) throws Exception {        ZooKeeper zk = getZookeeper();        //获取路径下的节点        List
    
      children = zk.getChildren(path, false);        for (String pathCd : children) {            //获取父节点下面的子节点路径            String newPath = "";            //递归调用,判断是否是根节点            if (path.equals("/")) {                newPath = "/" + pathCd;            } else {                newPath = path + "/" + pathCd;            }            rmr(newPath);        }        //删除节点,并过滤zookeeper节点和 /节点        if (path != null && !path.trim().startsWith("/zookeeper") && !path.trim().equals("/")) {            zk.delete(path, -1);            //打印删除的节点路径            System.out.println("被删除的节点为：" + path);        }    }    /**     * 获取Zookeeper实例     * @return     * @throws IOException     */    public static ZooKeeper getZookeeper() throws IOException {        zookeeper = new ZooKeeper(connectString, sessionTimeout, new Watcher() {            @Override            public void process(WatchedEvent event) {            }        });        return zookeeper;    }}复制代码

终端命令查看消息

bin/kafka-console-consumer.sh --bootstrap-server 192.168.1.177:9999 --topic mytopic  --from-beginning复制代码

安利一下可视化工具Kafka Tool 2

下载地址

Kafka Tool 2是一款Kafka的可视化客户端工具，可以非常方便的查看Topic的队列信息以及消费者信息以及kafka节点信息。下载地址：

使用

先创建连接

下载安装之后会弹出一个配置连接的窗口，我们可以看到这个窗口左上角为Add Cluster(添加集群)，但没关系，对应单节点的Kafka实例来说也是可以的，因为这个软件监控的是Zookeeper而不是Kafka，Kafka的集群搭建也是依赖Zookeeper来实现的，所以默认情况下我们都是直接通过Zookeeper去完成大部分操作。

创建完成之后，连接

我们可以看到已经创建好的Topic。这个软件默认显示数据的类型为Byte，可以在设置里面找到对应的修改选项

接下来就自己探索吧

理论学习

kafka单节点的结构如下：

单节点broker包含多个topic主题，而每个topic则包含多个partition副本，每个partition会有序的存储消息。

kafka的总体数据流

kafka对外使用topic的概念，生产者往topic里写消息，消费者从topic读消息。为了做到水平扩展，一个topic实际是由多个partition组成的，遇到瓶颈时，可以通过增加partition的数量来进行横向扩容。单个parition内是保证消息有序。每新写一条消息，kafka就是在对应的文件append写，所以性能非常高。kafka的总体数据流是这样的：

Producers往Brokers里面的指定Topic中写消息，Consumers从Brokers里面拉去指定Topic的消息，然后进行业务处理。

名词解析

Producer

消费者： Producer将消息发布到指定的Topic中,同时Producer也能决定将此消息归属于哪个partition;比如基于"round-robin"方式或者通过其他的一些算法等.

Consumer

每个consumer属于一个consumer group;反过来说,每个group中可以有多个consumer.发送到Topic的消息,只会被订阅此Topic的每个group中的一个consumer消费（对于一条消息来说，同一组的消费者只会有一个消费者去消费）.

如果所有的consumer都具有相同的group,这种情况和queue模式很像;消息将会在consumers之间负载均衡. 如果所有的consumer都具有不同的group,那这就是"发布-订阅";消息将会广播给所有的消费者.

在kafka中,一个partition中的消息只会被group中的一个consumer消费;每个group中consumer消息消费互相独立;我们可以认为一个group是一个"订阅"者,一个Topic中的每个partions,只会被一个"订阅者"中的一个consumer消费,不过一个consumer可以消费多个partitions中的消息.kafka只能保证一个partition中的消息被某个consumer消费时,消息是顺序的。事实上,从Topic角度来说，消息仍不是有序的。

Topics

一个Topic可以认为是一类消息，每个topic将被分成多个partition(区),每个partition在存储层面是append log文件。任何发布到此partition的消息都会被直接追加到log文件的尾部，每条消息在文件中的位置称为offset（偏移量），offset为一个long型数字，它是唯一标记一条消息。它唯一的标记一条消息。kafka并没有提供其他额外的索引机制来存储offset，因为在kafka中几乎不允许对消息进行“随机读写”。

Partition

topic物理上的分组，一个topic可以分为多个partition，每个partition是一个有序的队列

以下是单个生产者和消费者从两个分区主题读取和写入的简单示例。

此图显示了一个producer向2个partition分区写入日志，以及消费者从相同日志中读取的内容。日志中的每条记录都有一个相关的条目号，称之为偏移量offset。消费者使用此偏移来记录其在partitiond读取日志的位置。

当然如果存在多个消费者的话，根据groupId分组，同一组的消费者不会重复读取日志。

换句话说：订阅topic是以一个消费组来订阅的，一个消费组里面可以有多个消费者。同一个消费组中的两个消费者，不会同时消费一个partition。换句话来说，就是一个partition，只能被消费组里的一个消费者消费，但是可以同时被多个消费组消费。因此，如果消费组内的消费者如果比partition多的话，那么就会有个别消费者一直空闲。 复制代码

其实consumer可以使用任意顺序消费日志消息,它只需要将offset重置为任意值.(offset将会保存在zookeeper中，kafka集群几乎不需要维护任何consumer和producer状态信息,这些信息有zookeeper保存）

partition有多个.最根本原因是kafka基于文件存储.通过分区,可以将日志内容分散到多个partition上,来避免文件大小达到单机磁盘的上限,每个partiton都会被当前server(kafka实例)保存;可以将一个topic切分多任意多个partitions,来消息保存/消费的效率.此外越多的partitions意味着可以容纳更多的consumer,有效提升并发消费的能力.

使用场景

消息系统、消息队列

对于一些常规的消息系统,kafka是个不错的选择;partitons/replication和容错,可以使kafka具有良好的扩展性和性能优势.不过到目前为止,我们应该很清楚认识到,kafka并没有提供JMS中的"事务性""消息传输担保(消息确认机制)""消息分组"等企业级特性;kafka只能使用作为"常规"的消息系统,在一定程度上,尚未确保消息的发送与接收绝对可靠(比如,消息重发,消息发送丢失等)

日志聚合

kafka的特性决定它非常适合作为"日志收集中心";application可以将操作日志"批量""异步"的发送到kafka集群中,而不是保存在本地或者DB中;kafka可以批量提交消息/压缩消息等,这对producer端而言,几乎感觉不到性能的开支.此时consumer端可以使hadoop等其他系统化的存储和分析系统.

网站活动追踪、调用链系统、事件采集

可以将网页/用户操作等信息发送到kafka中.并实时监控,或者离线统计分析等

等等其他场景

server.properties配置文件解读

############################# Server Basics ############################## 节点的ID，必须与其它节点不同broker.id=0# 选择启用删除主题功能，默认false#delete.topic.enable=true############################# Socket Server Settings ############################## 套接字服务器坚挺的地址。如果没有配置，就使用java.net.InetAddress.getCanonicalHostName()的返回值# FORMAT:# listeners = listener_name://host_name:port# EXAMPLE:# listeners = PLAINTEXT://your.host.name:9092#listeners=PLAINTEXT://:9092# 节点的主机名会通知给生产者和消费者。如果没有设置，如果配置了"listeners"就使用"listeners"的值。# 否则就使用java.net.InetAddress.getCanonicalHostName()的返回值#advertised.listeners=PLAINTEXT://your.host.name:9092# 将侦听器的名称映射到安全协议，默认情况下它们是相同的。有关详细信息，请参阅配置文档#listener.security.protocol.map=PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL# 服务器用来接受请求或者发送响应的线程数num.network.threads=3# 服务器用来处理请求的线程数，可能包括磁盘IOnum.io.threads=8# 套接字服务器使用的发送缓冲区大小socket.send.buffer.bytes=102400# 套接字服务器使用的接收缓冲区大小socket.receive.buffer.bytes=102400# 单个请求最大能接收的数据量socket.request.max.bytes=104857600############################# Log Basics ############################## 一个逗号分隔的目录列表，用来存储日志文件log.dirs=/tmp/kafka-logs# 每个主题的日志分区的默认数量。更多的分区允许更大的并行操作，但是它会导致节点产生更多的文件num.partitions=1# 每个数据目录中的线程数，用于在启动时日志恢复，并在关闭时刷新。num.recovery.threads.per.data.dir=1############################# Internal Topic Settings ############################## 内部主题设置# 对于除了开发测试之外的其他任何东西，group元数据内部主题的复制因子“__consumer_offsets”和“__transaction_state”，建议值大于1，以确保可用性(如3)。offsets.topic.replication.factor=1transaction.state.log.replication.factor=1transaction.state.log.min.isr=1############################# Log Flush Policy ############################## 在强制刷新数据到磁盘之前允许接收消息的数量#log.flush.interval.messages=10000# 在强制刷新之前，消息可以在日志中停留的最长时间#log.flush.interval.ms=1000############################# Log Retention Policy ############################## 以下的配置控制了日志段的处理。策略可以配置为每隔一段时间删除片段或者到达一定大小之后。# 当满足这些条件时，将会删除一个片段。删除总是发生在日志的末尾。# 一个日志的最小存活时间，可以被删除log.retention.hours=168# 一个基于大小的日志保留策略。段将被从日志中删除只要剩下的部分段不低于log.retention.bytes。#log.retention.bytes=1073741824# 每一个日志段大小的最大值。当到达这个大小时，会生成一个新的片段。log.segment.bytes=1073741824# 检查日志段的时间间隔，看是否可以根据保留策略删除它们log.retention.check.interval.ms=300000############################# Zookeeper #############################zookeeper.connect=localhost:2181# 连接到Zookeeper的超时时间zookeeper.connection.timeout.ms=6000############################# Group Coordinator Settings #############################group.initial.rebalance.delay.ms=0复制代码

参考文章

如有表述不当之处，敬请指正。