Kafka是一个能够支持高并发以及流式信息处置的信息两边件，并且Kafka天生就是支持集群的，当天就关键来引见一下如何搭建Kafka集群。

Kafka目前支持经常使用Zookeeper形式搭建集群以及KRaft形式（即无Zookeeper）形式这两种形式搭建集群，这两种形式各有各的好处，当天就来区分引见一下这两种形式。

1，Kafka集群中的节点类型

咱们首先要求了解一下，一个Kafka集群是由下列几种类型的节点导致的，它们充任着不同的作用：

Broker节点：即代理节点，是Kafka中的上班节点，充任信息队列的角色，担任贮存和处置信息，每个Broker都是一个独立的Kafka主机，可以在不同的机器上运转，除此之外Broker还担任分区（partition）的治理，将主题（topic）划分为多个分区，并散布在集群的不同Broker上

Controller节点：即控制器节点，是集群中的不凡节点，担任贮存和治理整个集群元数据和形态，它能够监控整个集群中的Broker，在要求时还能够启动平衡操作

混合节点：即同时担任Broker和Controller节点角色的节点

2，两种形式集群的搭建形式

接上去，我就来引见一下两种形式的集群架构和搭建形式，即Zookeeper形式集群和KRaft形式集群。

(1)Zookeeper形式集群

这是一种比拟繁难，相对传统的搭建形式了！在这种形式下，每个Kafka节点都是依赖于Zookeeper的，经常使用Zookeeper存储集群中一切节点的元数据。

只需一切的Kafka节点衔接到同一个Zookeeper上方（或许同一个Zookeeper集群），这些Kafka节点就导致了一个集群。所以说就算是只要一个Kafka节点在运转，这一个节点也可以称作一个集群。

在Zookeeper形式集群中，Zookeeper节点（或许集群）就充任了Controller的角色，而一切的Kafka节点就充任着Broker的角色。

上方就来引见一下搭建环节，这里我在4台虚构机上区分运转Zookeeper和Kafka来模拟一个集群，一共一个Zookeeper节点和三个Kafka节点导致，如下：

上述地址例如kafka1等等，是经过修正虚构机的主机名（hostname）成功的，这样虚构机之间可以间接经过这些主机名相互访问，这个主机名咱们就可以视作实践在主机上方搭建时，主机的外网地址或许域名，这里就不再赘述如何修正虚构机的主机名了，要求保障上述一切虚构机在一个虚构机网段中并且能够相互ping通，即上述一切虚构机要求两两之间可以经过网络相互访问。

运转Kafka和Zookeeper都要求8及其以上运转环境，大家要首先在虚构机中装置并性能好。

①搭建Zookeeper

首先咱们要运转起一个Zookeeper节点，这里就不再赘述Zookeeper节点如何搭建了！搭建可以检查官方文档，或许经常使用的形式搭建。

搭建成功并运转Zookeeper之后，咱们会把一切的Kafka节点都性能到这一个Zookeeper节点上。

②性能并运转一切Kafka节点

首先去Kafka官方下载最新版并解压，而后将解压出来的Kafka区分复制到三台虚构机中。

而后修正每台虚构机的Kafka目录中的性能文件，性能文件位于解压的Kafka文件夹中的config/server.properties，经常使用文本编辑器关上，并找到下列性能项启动性能：

留意上述advertised.listeners这特性能项自动状况下是被注释掉了的，大家要求去细心找一下并去掉注释（扫尾的#）而后再启动性能。

三台虚构机性能成功后，区分经常使用终端进入到Kafka目录下并启动，口头下列命令：

bin/kafka-server-start.shconfig/server.properties

在上述三台虚构机上方都经过这个命令启动Kafka，如图则启动成功：

到此，整个集群就搭建成功了！大家要求保障上述三台虚构机中的终端不被封锁。

③创立话题测试

咱们先在kafka1的虚构机上方再开一个终端并进入Kafka目录，口头下列命令创立Topic：

bin/kafka-server-start.shconfig/server.properties

而后去kafka2的虚构机上方再开一个终端并进入Kafka目录，口头下列命令列出Topic：

bin/kafka-topics.sh--list--bootstrap-serverlocalhost:9092

可见咱们在第一个节点上创立了话题，然而在第二个节点上依然可以失掉这个话题，说明集群创立成功，数据在集群之间可以共享。

(2)KRaft形式集群

在上述传统打算中，Kafka要求依赖Zookeeper成功元数据寄存和共享，这样也就暴显露了一些疑问：

搭建Kafka集群时还要求额外搭建Zookeeper，参与了运维老本

Zookeeper是强分歧性的组件（合乎CP通常），假设集群中数据出现变动，那么肯定要等到其它节点都同步，至少超越一半同步成功，这样节点数多性能差

那么KRaft形式是新版本Kafka中推出的集群形式，这种形式下就完全不要求Zookeeper了！只要求数个Kafka节点就可以间接导致集群，在这时集群中的Kafka节点既有或许是Controller节点也或许是Broker节点，在这个形式中，咱们不只可以手动性能某个节点的角色（是Controller还是Broker），还可以使其同时担任Broker和Controller角色（混合节点）。

在KRaft形式中，集群的节点会经过投票选举的形式，选用出一个关键的Controller节点，这个节点也称作指导者，它将担任保养整个集群的元数据和形态信息，那么其它的Controller节点或许混合节点就称之为追寻者，它们会从指导者同步集群元数据和形态信息。假设指导者宕机了，一切的节点会从新投票选举一个新的指导者。

在选举环节中，一切的节点都会介入投票环节，而候选节点只会是Controller节点或许混合节点（即Broker节点不会被选举为指导者）。

要求留意的是，在自动状况下Kafka集群中的Broker节点和Controller节点通常会监听不同的端口：

所以要求依据实践状况性能网络设置和防火墙规定，以确保Kafka集群中的节点能够在正确的端口上启动通讯。上述提到的两种端口也是可以修正的，当然不倡导修正。

雷同地，就算是你只是搭建了一个Kafka节点，这一个节点也依然被视为一个Kafka集群，并且KRaft形式下假设只要求建设一个节点，那么这个节点肯定是混合节点。

上方雷同是开启三台虚构机，搭建三个Kafka节点导致的KRaft形式集群如下：

这里就不再赘述下载Kafka的环节了！

①修正性能文件

在KRaft形式下，性能文件位于Kafka目录中的config/kraft/server.properties，经常使用文本编辑器关上并找到下列性能以修正：

node.id示意这个节点的id，一个集群中每个节点id不能重复，要求是不小于1的整数，这里三台虚构机的性能区分为1，2和3（相似上述Zookeeper的broker.id性能）

controller.quorum.voters设定投票者列表，即要求性能一切的Controller节点id及其地址端口，性能格局为节点1的id@节点1地址:节点1端口,节点2的id@节点2地址:节点2端口,节点3的id@节点3地址:节点3端口...，这里的端口要求是控制器端口，自动都是9093，上方也提到过了，自动不要求修正，我这里三台虚构机的都性能为1@kafka1:9093,2@kafka2:9093,3@kafka3:9093（实践在主机上搭建时交流为主机的外网地址或许域名）

advertised.listeners示意这个Kafka节点的外网地址，这里区分性能为PLAINTEXT://kafka1:9092，PLAINTEXT://kafka2:9092和PLAINTEXT://kafka3:9092（和上述Zookeeper形式中的一样，实践在主机上搭建时交流为主机的外网地址或许域名）

上述是肯定要启动性能的，还有上方性能是可以选用性性能的：

process.roles示意设定这个节点的类型，设定为broker示意设定这个节点为Broker节点，雷同地设定controller示意设定为Controller节点，自动是broker,controller示意这个节点会智能切换节点类型，这里先坚持自动不变，上方再来具体探讨

②生成集群ID并经常使用集群ID格局化数据目录

在KRaft形式下，一个集群要求设定一个id，咱们可以经常使用自带的命令生成，先进入上述恣意一台虚构机并经常使用终端进入Kafka目录中，口头下列命令生成一个UUID：

bin/kafka-storage.shrandom-uuid

咱们这里记载下这个ID以备用。

这个集群ID理想上是一个长度16位的字符串经过Base64编码后得来的，因此你也可以不经常使用上述命令，间接自定义一个16位长度的纯英文和数字组成的字符串，而后将这个字符串编码为Base64格局作为这个集群ID也可以。可以经常使用菜鸟工具中的在线Base64编码工具。

而后在上述三台虚构机中，都经常使用终端进入Kafka目录后，口头下列命令：

bin/kafka-storage.shformat-t生成的集群ID-cconfig/kraft/server.properties

这样，三个Kafka节点都经常使用了这一个ID成功了集群元数据性能，示意这三个Kafka节点导致一个集群。

③启动Kafka

雷同地，在三台虚构机中，都经常使用终端进入Kafka目录后，口头下列命令：

bin/kafka-server-start.shconfig/kraft/server.properties

三台虚构机所有启动后，这个集群才启动终了。

④创立话题测试

雷同地，如今第一个虚构机的Kafka目录下口头下列命令：

bin/kafka-topics.sh--create--topicmy-topic-kraft--bootstrap-serverlocalhost:9092

而后在第二个虚构机的Kafka目录下检查话题：

bin/kafka-topics.sh--list--bootstrap-serverlocalhost:9092

可见集群节点之间可以相互通讯。

无论是在虚构机还是主机上，都要保障9092和9093端口开明，且一切虚构机/主机之间都能够两两相互访问（网络连通）！

3，关键性能引见

无论是那种形式的集群，咱们都触及到了许多性能项，大家经过上述的搭建示例也能够了解到每特性能项的意义，这里就专门来着重引见一下，Kafka中一些关键的性能项。

(1)listeners

这特性能项用于指定Kafka主机监听客户端衔接的地址和端口，当Kafka主机启动时，它将监听listeners性能项中指定的地址和端口，期待客户端的衔接恳求。

普通状况下这特性能以PLAINTEXT://或许CONTROLLER://扫尾，意义如下：

这特性能项通常不要求修正，上方给出几特性能示例

(2)advertise.listeners

这特性能容易和listeners混杂，理想上它们是有较大的区别的。

该性能项指定Kafka主机广播给客户端的地址和端口，通常性能为Kafka所在主机的外网地址。

当客户端（消费者或消费者）尝试衔接到Kafka主机时，它首先会失掉Kafka主机广播的地址和端口，也就是advertise.listeners性能所指定的地址和端口，而后才会经常使用advertise.listeners性能所指定的地址和端口来建设与Kafka主机的衔接。

置信这时大家会有个疑问：既然客户端要衔接Kafka（例如SpringBoot集成Kafka客户端），那肯定是曾经知道了Kafka对外的地址端口了，那为什么衔接的时刻还要求失掉一下广播的地址端口再启动衔接呢？这样是不是有一些多此一举？

理想上，Kafka设计这特性能是为了处置上方较为复杂的网络场景：

总之，这特性能设置为Kafka主机所在的外网地址即可！例如PLAINTEXT://69.54.112.239:9092。

(3)process.roles

这是KRaft形式下专门的性能，用于性能这个节点的类型，可以性能为下列值：

假设没有性能这个选项，则Kafka会以Zookeeper形式运转。

这里有下列留意事项：

则不能性能advertised.listeners，可以将其注释掉或许删掉

listeners要求性能为CONTROLLER://扫尾，倡导性能为CONTROLLER://:9093

则要求性能advertised.listeners为主机外网地址和端口，这和Zookeeper形式中相反

listeners要求性能为PLAINTEXT://扫尾，倡导性能为PLAINTEXT://:9092

雷同要求性能advertised.listeners为主机外网地址和端口

listeners要求同时性能CONTROLLER://和PLAINTEXT://，倡导性能为PLAINTEXT://:9092,CONTROLLER://:9093

在开发环境或许小规模集群，可以所有经常使用混合节点，假设是消费环境就倡导设定好每个节点的类型了！并且通常要求先启动Controller节点再启动Broker节点。

理想上，咱们发现Kafka的KRaft性能目录config/kraft下有三特性能文件，其中server.properties是混合节点的性能模板，而broker.properties和controller.properties区分是Broker节点和Controller节点的性能模板，大家假设要设定节点类型，可以间接经常使用对应的性能文件，将对应性能文件要求修正的局部修正一下，而后将上述格局化数据目录命令和启动命令中的性能文件门路扭转一下即可，这样可以省略咱们设定process.roles和listeners或许控制器节点删除advertise.listeners性能的操作。

(4)controller.quorum.voters

该性能项用于性能集群中Controller节点选举环节中的投票者，集群中一切的Controller节点都要求被列举在这特性能项中，其性能格局为id1@host1:port1,id2@host2:port2,id3@host3:port3...。

有的同窗或许以为这里要求把集群中一切节点都写出来，理想上这是失误的，这里只要求写一切的Controller节点和混合节点的id、地址和端口即可，这特性能中性能的端口当然是控制器端口。

上述集群搭建的例子中，因为一切的节点都是混合节点，因此就所有写在其中了！假设咱们手动设定每个节点的类型，例如：

那么一切节点的controller.quorum.voters都要求性能为1@kafka1:9093。

理想上，一切的节点都是经过这特性能中的节点列表，来得悉一切的控制器节点信息（以失掉集群元数据）并失掉投票候选者的，因此集群中一切节点，不论是Broker还是Controller，还是混合节点，都要求性能这一项。

(5)其它性能

除了上述咱们触及到的一些性能之外，还有下列性能大家可以启动修正：

上述无论是哪个形式的集群，都可以在性能文件中找到这些性能，假设找不到可手动参与。除了修正性能文件之外，咱们还可以在启动Kafka的命令中指定性能和值，例如：

bin/kafka-server-start.shconfig/server.properties--overridezookeeper.connect=127.0.0.1:2181--overridebroker.id=1

上述命令在启动时经过命令指定了zookeeper.connect性能值为127.0.0.1:2181，以及broker.id为1，可见在前面追加--override性能名=值即可，留意命令行中指定的性能值会笼罩掉性能文件中的性能值！

---

Kafka简介+Kafka Tool使用简介+使用实例

详细安装访问：macOS 可以用homebrew快速安装,访问地址:

原文链接：查看topic列表：

创建topic：

--create ：创建命令；--topic ：后面指定topic名称；--replication-factor ：后面指定副本数；--partitions ：指定分区数，根据broker的数量决定；--zookeeper ：后面指定 的zk链接 查看某个topic：

Kafka 作为消息系统的一种， 当然可 以像其他消 息中 间件一样作为消息数据中转的平台。 下面以 Java 语言为例，看一下如何使用 Kafka 来发送和接收消息。 1、引入依赖

2、消息生产者

示例 中用 KafkaProducer 类来创建一个消息生产者,该类的构造函数入参是一系列属性值。下面看一下这些属性具体都是什么含义。

表示 Kafka 集群 。 如果集群中有多台物理服务器，则服务器地址之间用逗号分隔， 比如” 192.168.1.1 :9092,192.168.1.2:9092” 。 localhost 是笔者电脑的地址，9092 是 Kafka 服务器默认监听的端口号。

和 表示消息的序列化类型 。 Kafka 的消息是以键值对的形式发送到 Kafka 服务器的，在消息被发送到服务器之前，消息生产者需要把不同类型的 消息序列化为 二 进制类型，示例中是发送文本消息到服务器 ， 所以使用的是StringSerializer。

和 表示消息的反序列化类型。把来自 Kafka 集群的二进制消 息反序列 化 为指定 的 类型,因为序列化用的是String类型，所以用StringDeserializer 来反序列化。 用于指定 Kafka 连接 ZooKeeper 的 URL ，提供了基于 ZooKeeper 的集群服务器自动感知功能， 可以动态从 ZooKeeper 中读取 Kafka 集群配置信息。

有 了 消息生产者之后 ， 就可以调用 send 方法发送消息了。该方法的入参是 ProducerRecord类型对象 ， ProducerRecord 类提供了多种构造函数形参，常见的有如下三种 ：

ProducerRecord(topic,partition,key,value)；

ProducerRecord(topic,key,value)；

ProducerRecord(topic, value) ;

其中 topic 和 value 是必填的， partition 和 key 是可选的 。如果指定了 pa时tion，那么消息会被发送至指定的 partition ；如果没指定 partition 但指定了 Key，那么消息会按照 hash(key）发送至对应的 partition： 如果既没指定 partition 也没指定 key，那么 消息会按照 round-robin 模式发送（即以轮询的方式依次发送〉到每一个 partition。示例中将向 test-topic 主题发送三条消息。

3、消息消费者

和消息生产者类似，这里用 KafkaConsumer 类来创建一个消息消费者，该类的构造函数入参也是一系列属性值。

bootstrap. servers和生产者一样，表示 Kafka 集群。

表示消费者的分组 ID。

表示 Consumer 的 offset 是否自 动提交 。

用于设置自动提交 offset 到 ZooKeeper 的时间间隔，时间单位是毫秒。

key. deserializer和 表示用字符串来反序列化消息数据。

消息消费者使用 subscribe 方法 订阅了 Topic 为 test-topic的消息。 Consumer 调用poll 方法来轮询 Kafka 集群的消息， 一直等到 Kafka 集群中没有消息或达到超时时间（示例中设置超时时间为 100 毫秒）为止 。 如果读取到消息，则打印出消息记录的 pa此ition, offset、key 等。

Kafka集群部署（Docker容器的方式）

文章主要介绍以docker容器的方式部署kafka集群。 上述配置文件中的server.x，数字x对应到data/myid文件中的值。 三台机器x的值分别就是1,2,3。 参数详细说明请参考 官网文档 。 1.--net=host: 容器与主机共享同一Network Namespace，即容器与网络看到的是相同的网络视图（host模式存在一定的风险，对安全要求很高的生产环境最好不要用host模式，应考虑除此之外的其他几种模式） 2.-v: 指定主机到容器的目录映射关系 这样就以容器的方式启动了zookeeper的服务，可以通过 docker exec -it zookeeper bash 命令进入容器中进行一些操作，例如查看服务启动是否正常。 也可以通过查看2181端口是否被监听判断zookeeper的服务是否运行 详细的参数配置说明请参考 官方文档 ，参数不仅可以通过上述文件的方式来配置，也可以通过容器环境变量的方式来配置，这里结合两种方式使用。 _ADVERTISED_HOST_NAME、KAFKA_BROKER_ID的值要结合每台机器自身设置 2./etc/hosts文件中最好配置ip与hostname的映射关系，否则会报出如下错误 Error: Exception thrown by the agent : : Local host name unknown: : node0: node0: System error 3.通过-e 指定的环境变量与在中配置的选项其效果是一样的 4.配置文件中的选项若要通过环境变量来指定，方式为：如对应KAFKA_BROKER_ID，类似的对应KAFKA_LOG_DIRS _HEAP_OPTS=-Xmx6G -Xms6G指java堆内存大小的设置，6G大小是kafka官网给出的数值，此数值要结合机器的内存大小给出。 超过6G的内存，可以设置为6G；若机器的内存低于6G而设置6G，则会报错。 5.启动成功后，可以通过docker logs kafka命令查看日志 _HOSTS：ZooKeeper访问地址（需指定机器的ip，localhost:2181或127.0.0.1:2181均会报 : Connection refused 异常）