本文主要介绍kafka安装和部署的详细步骤。主要应用场景有：日志采集系统和消息系统。这篇文章很详细的介绍了你，对你的学习或者工作有一定的参考价值。有需要的朋友可以参考一下。

目录

概述第一步：下载代码第二步：启动服务第三步：创建主题第四步：发送消息第五步：消费一条消息第六步：设置多个代理集群(单机伪集群的配置)第七步：测试集群的容错性7.1向集群发布消息7.2消费消息7.3杀死领导者，测试集群容错性7.4再次消费消息，确认消息不丢失。

概述

Kafka最初由Linkedin开发，是由zookeeper协调的分布式、分区、多副本、多订户的分布式日志系统(也可作为MQ系统)。常用于web/nginx日志、访问日志、消息服务等。Linkedin在2010年向Apache Foundation捐款，成为顶级开源项目。

主要应用场景有：日志采集系统和消息系统。

卡夫卡的主要设计目标如下：

消息持久能力以时间复杂度为O(1)的方式提供，即使对于TB级别以上的数据也能保证恒定的时间访问性能。

高吞吐量。即使在非常便宜的商用机器上，单台机器也能支持每秒100K条消息的传输。

支持Kafka服务器和分布式消费之间的消息分区，保证消息在各个分区的顺序传输。

支持离线数据处理和实时数据处理。

横向扩展：支持在线横向扩展

Step 1: 下载代码

可以登录Apache kafka，正式下载。

http://kafka.apache.org/downloads.html

注意：2.11-1.1.0版本兼容JDK1.7，否则后续版本需要JDK1.8。

Step 2: 启动服务

运行kafka需要Zookeeper，所以需要先启动Zookeeper。如果没有Zookeeper，可以使用kafka自带的打包配置的Zookeeper(PS:在kafka包里)。

//这是前台启动。启动后，目前不能进行其他操作(不推荐)。/zookeeper-server-start.sh./config/zookeeper.properties

//在后台启动(推荐)。/zookeeper-server-start.sh./config/zookeeper . properties 1/dev/null 21

现在就开始卡夫卡

配置/服务器1.properties:

broker.id=0

listeners=明文：//192.168.10.130:9092

log.dirs=卡夫卡-logs

zookeeper . connect=localhost:2181

//在后台启动卡夫卡。/kafka-server-start.sh./config/server . properties 1/dev/null 21

Step 3：创建一个主题

创建一个名为“test”的主题，只有一个分区和备份(2181是zookeeper的默认端口)。/Kafka-topics . sh-create-zookeeper localhost:2181 -config max . message . bytes=12800000-config flush . messages=1 -复制-因子1-分区1-主题测试

命令分辨率：

- create:指定创建主题的动作。

- topic:指定新主题的名称。

- zookeeper:指定kafka连接zk的连接url，与server.properties文件中的配置项{zookeeper.connect}相同。

- config:指定当前主题的有效参数值。参数表的参考文件是：http://kafka.apache.org/082/documentation.html#brokerconfigs.

- partitions:指定当前创建的kafka分区的数量，默认情况下为1。

- replication-factor:为每个分区指定复制因子的数量，默认情况下为一个。

创建后，您可以通过运行以下命令来查看创建的主题信息：/Kafka-topics . sh-list-zookeeper localhost:2181

试验

或者，除了手动创建主题之外，您还可以配置您的代理在发布不存在的主题时自动创建主题。

补充：

(1)查看对应题目的描述信息。/Kafka-topics . sh-describe-zookeeper localhost:2181-topic test0

命令分辨率：

- describe:指定这是一个显示详细信息的命令。

- zookeeper:指定kafka连接zk的连接url，与server.properties文件中的配置项{zookeeper.connect}相同。

- topic:指定要显示数据的主题的名称。

(2)主题信息修改

bin/Kafka-topics . sh-zookeeper 192 . 168 . 187 . 146:2181-alter-topic test0-config max . message . bytes=128000

bin/Kafka-topics . sh-zookeeper 192 . 168 . 187 . 146:2181-alter-topic test0-delete-config max . message . bytes

bin/Kafka-topics . sh-zookeeper 192 . 168 . 187 . 146:2181-alter-topic test 0-partitions 10

bin/Kafka-topics . sh-Zookeeper 192 . 168 . 187 . 146:2181-ALTER-TOPIC TEST 0-PARTITIONS 3 # # Kafka分区的数量只能增加，不能减少。

(3)话题删除

默认情况下，卡夫卡的话题不能直接删除，需要配置相关参数。

bin/Kafka-topics . sh-delete-topic test0-zookeeper 192 . 168 . 187 . 146:2181

加粗样式

注意：如果删除，这不会有任何影响。话题。Enable未设置为true。# #默认情况下，删除是标记删除，此话题并未实际删除；如果运行删除主题，有两种方法：

方法一：通过delete命令删除后，手动删除zk上的本地磁盘和相关主题信息。

方法2:配置server.properties文件，并使用参数delete重新启动kafka服务。Topic.enable=true。此时，执行删除命令意味着可以删除该主题。

Step 4: 发送消息

Kafka提供了一个命令行工具，可以从输入文件或命令行读取消息，并发送给Kafka集群。每一行都是一个信息。

运行producer，然后在控制台中向服务器输入几条消息。

备注：这里的localhost:9092不是固定的，需要根据server.properties中配置的地址来写这里的地址！

[root@administrator bin]#。/Kafka-console-producer . sh-broker-list localhost:9092-主题测试

这是一条信息

这是另一条消息

//按“Ctrl C”终止输入

Step 5: 消费消息

Kafka还提供了一个命令行工具来消费消息和输出存储的信息。

备注：这里的localhost:9092不是固定的，需要根据server.properties中配置的地址来写这里的地址！

[root@administrator bin]#。/Kafka-console-consumer . sh-bootstrap-server localhost:9092-主题测试-从头开始

这是一条信息

这是另一条消息

//按“Ctrl C”停止阅读消息

如果您有两个不同的终端运行上述命令，那么当您运行生产者时，消费者可以使用生产者发送的消息。

Step 6: 设置多个broker集群（单机伪集群的配置）

到现在为止，我们只经营一家券商，没什么意思。对于卡夫卡来说，一个经纪人不过是一个集群的大小，那我们就多成立几个经纪人吧。

首先，为每个代理创建一个配置文件：

cp config/server.properties配置/server-1.properties

cp配置/server.properties配置/server-2.properties

现在编辑这些新创建的文件，并设置以下属性：

vim配置/服务器.属性

配置/服务器1.properties:

broker.id=0

listeners=明文：//192.168.10.130:9092

log.dirs=卡夫卡-logs

zookeeper . connect=localhost:2181

config/server-1 .属性：

broker.id=1

listeners=明文：//192.168.10.130:9093

log.dirs=kafka-logs-1

zookeeper . connect=localhost:2181

配置/server-2.properties:

broker.id=2

listeners=明文：//192.168.10.130:9094

log.dirs=kafka-logs-2

zookeeper . connect=localhost:2181

备注1:侦听器必须配置为IP地址；如果配置为服务器的localhost或主机名，那么使用Java:org . Apache . Kafka . common . errors发送数据时会抛出异常，超时异常：batched expired。因为没有配置advertised.host.name，所以Kafka并没有像官方文档宣称的那样，而是广播了我们配置的host.name，而是广播了主机配置的主机名。远程客户端没有配置主机，所以自然不能连接到这个主机名。

备注2:使用java客户端访问远程kafka时，一定要打开集群中的所有端口，否则连接会超时。

/sbin/iptables -I输入-p tcp - dport 9092 -j接受

/sbin/iptables -I输入-p tcp - dport 9093 -j接受

/sbin/iptables -I输入-p tcp - dport 9094 -j接受

/etc/rc.d/init.d/iptables保存

Broker.id是集群中每个节点的唯一永久名称。我们修改端口和日志目录，因为我们现在运行在同一台机器上。我们希望防止broker在同一个端口上注册和覆盖彼此的数据。

我们已经运行了zookeeper和一个kafka节点，所以我们只需要启动两个新的kafka节点。/kafka-server-start.sh./config/server-1 . properties 1/dev/null 21。/kafka-server-start.sh./config/server-2。属性1/dev/null 21

现在，我们创建一个新题目，把备份设置为：3。/卡夫卡-话题。sh-create-zookeeper localhost:2181-replication-factor 3-partitions 1-topic my-replication-topic

好了，现在我们已经有了一个集群了，我们怎么知道每个集群在做什么呢？运行命令"描述主题"。/卡夫卡-话题。sh-describe-zookeeper localhost:2181-topic my-replicated-topic

//所有分区的摘要

主题：我的已复制主题分区计数：1复制因子：3配置：

//提供一个分区信息，因为我们只有一个分区，所以只有一行。

主题：我的已复制主题分区：0领导者：1副本：1，2，0 Isr: 1，2，0

"领导":该节点负责该分区的所有的读和写，每个节点的领导者都是随机选择的。

"副本":备份的节点列表，无论该节点是否是领导者或者目前是否还活着，只是显示。

" isr ":"同步备份"的节点列表，也就是活着的节点并且正在同步领导者

其中复制品和中断服务程序(中断服务程序的缩写)中的1,2,0就对应着3个经纪人他们的broker.id属性！

我们运行这个命令，看看一开始我们创建的那个节点：/卡夫卡-话题。sh-describe-zookeeper localhost:2181-主题测试

主题：测试分区计数：1复制因子：1配置：

主题：测试分区：0领导者：0副本：0 Isr: 0

这并不奇怪，刚才创建的主题没有复制品，并且在服务器"0"上，我们创建它的时候，集群中只有一个服务器，所以是"0"。

Step 7: 测试集群的容错能力

7.1发布消息到集群

[root@administrator bin]# ./卡夫卡-控制台-制作人。上海经纪人名单192。168 .10 .130:9092-主题我的重复主题

集群消息一

集群消息2

//Ctrl C终止产生消息

7.2消费消息

[root@administrator bin]# ./卡夫卡-控制台-消费者。引导服务器192。168 .10 .130:9093-从头开始-主题我的重复主题

集群消息一

集群消息2

//Ctrl C终止消费消息

7.3干掉leader，测试集群容错

首先查询谁是领导者。/卡夫卡-话题。sh-describe-zookeeper localhost:2181-topic my-replicated-topic

//所有分区的摘要

主题：我的已复制主题分区计数：1复制因子：3配置：

//提供一个分区信息，因为我们只有一个分区，所以只有一行。

主题：我的已复制主题分区：0领导者：1副本：1，2，0 Isr: 1，2，0

可以看到领导者的broker.id为1,找到对应的经纪人

[root@administrator bin]# jps -m

卡夫卡街5130号./config/server。属性

4861号./config/zookeeper.properties

1231引导启动启动

卡夫卡7420号./config/server-2。属性

卡夫卡7111号./config/server-1。属性

9139名太平绅士

通过以上查询到领导者的PID(卡夫卡./config/server-1.properties)为7111,杀掉该进程

//杀掉该进程

kill -9 7111

//再查询一下，确认新的领导者已经产生，新的领导者为broker.id=0

[root@administrator bin]# ./卡夫卡-话题。sh-describe-zookeeper localhost:2181-topic my-replicated-topic

主题：我的已复制主题分区计数：1复制因子：3配置：

//备份节点之一成为新的领导，而经纪人一已经不在同步备份集合里了

主题：我的已复制主题分区：0领导者：0副本：1，0，2 Isr: 0，2

7.4再次消费消息，确认消息没有丢失

[root@administrator bin]# ./卡夫卡-控制台-消费者。sh-zookeeper localhost:2181-from-beging-topic my-replication-topic

集群消息一

集群消息2

消息依然存在，故障转移成功！

到此这篇关于卡夫卡安装部署的文章就介绍到这了，更多相关卡夫卡安装部署内容请搜索我们以前的文章或继续浏览下面的相关文章希望大家以后多多支持我们！

郑重声明：本文由网友发布，不代表盛行IT的观点，版权归原作者所有，仅为传播更多信息之目的，如有侵权请联系，我们将第一时间修改或删除，多谢。