kafka集群搭建（windows环境下）

曾经终败给现在 2023-07-13 15:24 28阅读 0赞

## 一、简介 ##

Kafka 是一个实现了分布式的、具有分区、以及复制的日志的一个服务。它通过一套独特的设计提供了消息系统中间件的功能。它是一种发布订阅功能的消息系统。

### **1、名词介绍** ###

**Message**

消息，就是要发送的内容，一般包装成一个消息对象。

**Topic**

通俗来讲的话，就是放置“消息”的地方，也就是说消息投递的一个容器。假如把消息看作是信封的话，那么 Topic 就是一个邮箱

**Partition && Log**

Partition 分区，可以理解为一个逻辑上的分区，像是我们电脑的磁盘 C:, D:, E: 盘一样,

Kafka 为每个分区维护着一份日志Log文件。

**Producers****（生产者）**

和其他消息队列一样，生产者通常都是消息的产生方。

在 Kafka 中它决定消息发送到指定Topic的哪个分区上。

**Consumers****（消费者）**

消费者就是消息的使用者，在消费者端也有几个名词需要区分一下。

一般消息队列有两种模式的消费方式，分别是** 队列模式** 和 **订阅模式**。

**队列模式**：一对一，就是一个消息只能被一个消费者消费，不能重复消费。一般情况队列支持存在多个消费者，但是对于一个消息，只会有一个消费者可以消费它。

**订阅模式**：一对多，一个消息可能被多次消费，消息生产者将消息发布到Topic中，只要是订阅改Topic的消费者都可以消费。

## **二、安装zookeeper** ##

### **1、简介** ###

Kafka使用zookeeper作为其分布式协调框架，很好的将消息生产、消息存储、消息消费的过程结合在一起。同时借助zookeeper，kafka能够生产者、消费者和broker在内的所以组件在无状态的情况下，建立起生产者和消费者的订阅关系，并实现生产者与消费者的负载均衡。

### **2、下载zookeeper** ###

可以到zookeeper官网下载

[http://zookeeper.apache.org/releases.html][http_zookeeper.apache.org_releases.html]

### **3、配置zookeeper** ###

（1）下载解压完成后，来到conf文件夹下，有一个 **zoo\_sample.cfg** 官方默认的配置文件。复制一份，重命名为 zoo.cfg

（2）配置，打开zoo.cfg 修改配置信息

![复制代码][aHR0cHM6Ly9jb21tb24uY25ibG9ncy5jb20vaW1hZ2VzL2NvcHljb2RlLmdpZg]

#存储内存中数据库快照的位置，如果不设置参数，更新事务日志将被存储到默认位置。
    
    dataDir=../zkData
    
    #日志文件的位置
    
    dataLogDir=../zkLog
    
    #监听端口
    
    clientPort=2181

![复制代码][aHR0cHM6Ly9jb21tb24uY25ibG9ncy5jb20vaW1hZ2VzL2NvcHljb2RlLmdpZg]

（3）集群配置

server.1=127.0.0.1:12888:1388
    
    server.2=127.0.0.1:12889:1389
    
    server.3=127.0.0.1:12887:1387

**格式： server.A = B:C:D**

A:是一个数字，表示第几号服务器

B:服务器IP地址

C:是一个端口号，用来集群成员的信息交换,表示这个服务器与集群中的leader服务器交换信息的端口

D:是在leader挂掉时专门用来进行选举leader所用的端口

###  完整的配置文件如下  ###

![format_png][]

复制两份zookeeper解压好配置后的文件夹，命名为

![format_png 1][]

在对应的文件下下面修改zoo.cfg的监听端口地址

比如：

第一个zookeeper-3.4.6程序 修改zoo.cfg 配置文件

clientPort=2181

第二个zookeeper-3.4.6-2程序 修改zoo.cfg 配置文件

clientPort=2182

第三个zookeeper-3.4.6-2程序 修改zoo.cfg 配置文件

clientPort=2183

### **创建ServerID** ###

在配置的dataDir目录下面新建一个 myid 文件，文件内容就是对应的id号，

比如：

zookeeper-3.4.6程序 myid 文件的内容 为 1

zookeeper-3.4.6-2程序 myid 文件的内容 为 2

zookeeper-3.4.6-3程序 myid 文件的内容 为 3

我这边配置的目录是

![format_png 2][]

### **启动zookeeper** ###

在对应的bin目录下启动

zkServer.cmd

## **三、安装kafka** ##

### （1）下载 ###

去官网 [http://kafka.apache.org/][http_kafka.apache.org]下载即可 这边下载的是

[https://www.apache.org/dyn/closer.cgi?path=/kafka/0.8.2.2/kafka\_2.9.2-0.8.2.2.tgz][https_www.apache.org_dyn_closer.cgi_path_kafka_0.8.2.2_kafka_2.9.2-0.8.2.2.tgz]

这个版本

### （2）配置 ###

解压后到config文件夹下 打开server.properties配置文件进行配置

### （3）配置内容 ###

修改或新增以下配置信息

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 </td> 
 <td> <code>#唯一标识</code> <code>broker.id=0</code> <code>#监听端口</code> <code>port=9092</code> <code>host.name=127.0.0.1</code> <code>#消息最大大小</code> <code>message.max.bytes=50485760</code> <code>#配置副本数量 </code> <code>default.replication.factor=2</code> <code>#获取的最大大小</code> <code>replica.fetch.max.bytes=50485760</code> <code>#队列中消息持久化存放的位置，可以多个目录，用逗号分开</code> <code>log.dirs=/tmp/kafka-logs</code> <code>#默认的分区数</code> <code>num.partitions=2</code> <code>#对应着刚刚配置的zookeeper的三个ip与端口地址</code> <code>zookeeper.connect=127.0.0.1:2181,127.0.0.1:2182,127.0.0.1:2183</code> </td> 
 </tr> 
 </tbody> 
</table>

### （4）集群配置 ###

复制两份解压后的文件，命名如下

![format_png 3][]

**修改部分配置信息**

在**对应**的server.properties中修改

*\#**唯一标识*

broker.id=0

broker.id=1

broker.id=2

*\#**监听端口*

port=9092

port=9093

port=9094

### **启动对应的kafka** ###

进入到bin/windows目录下 启动kafka并指定配置文件

kafka-server-start.bat ../../config/server.properties

启动过程中如果遇到**Kafka****中错误：**

**Unrecognized VM option ‘UseCompressedOops’ Error: Clould not create the Java Vritual Machine. Error: A fatal exception has occurres . Program will exit.**

**解决方案：**

找到bin/windows/kafka-run-class.bat 文件,

找到112行左右

IF ["%KAFKA_JVM_PERFORMANCE_OPTS%"] EQU [""] (
    
      set KAFKA_JVM_PERFORMANCE_OPTS=-server -XX:+UseCompressedOops -XX:+UseParNewGC -XX:+UseConcMarkSweepGC -XX:+CMSClassUnloadingEnabled -XX:+CMSScavengeBeforeRemark -XX:+DisableExplicitGC -Djava.awt.headless=true
    
    )

删除掉 **-XX:+UseCompressedOops** 即可

### **测试集群** ###

（1）创建一个 topic

kafka-topics.bat --create --zookeeper 127.0.0.1:2181 --replication-factor 1 --partitions 1 --topic test

（2）查看是否创建成功

kafka-topics.bat --list --zookeeper localhost:2181

（3）发送消息

kafka-console-producer.bat --broker-list localhost:9092 --topic test
    
    This is a message

![format_png 4][]

（4）接收消息

kafka-console-consumer.bat --zookeeper localhost:2181 --topic test --from-beginning

![format_png 5][]

**不同客户端能接收到消息，说明配置成功**

博主强烈推荐：https://blog.csdn.net/persistencegoing/article/details/84376427

希望大家关注我一波,防止以后迷路,有需要的可以加群讨论互相学习java ,学习路线探讨,经验分享与java求职

群号:721 515 304

[http_zookeeper.apache.org_releases.html]: http://zookeeper.apache.org/releases.html
[aHR0cHM6Ly9jb21tb24uY25ibG9ncy5jb20vaW1hZ2VzL2NvcHljb2RlLmdpZg]: /images/20230529/69eac43fa9184dfc98a328450fff981a.png
[format_png]: /images/20230529/f0d48ddf33644bf8a19b4483196bfca8.png
[format_png 1]: /images/20230529/bd5d8c78223944c882bdbdfdf79ea5a2.png
[format_png 2]: /images/20230529/b66b34f0d0b14f5894c4b85f6b70559e.png
[http_kafka.apache.org]: http://kafka.apache.org/
[https_www.apache.org_dyn_closer.cgi_path_kafka_0.8.2.2_kafka_2.9.2-0.8.2.2.tgz]: https://www.apache.org/dyn/closer.cgi?path=/kafka/0.8.2.2/kafka_2.9.2-0.8.2.2.tgz
[format_png 3]: /images/20230529/3428d3bea5a34b579a68e16705ad7d7a.png
[format_png 4]: /images/20230529/30d993ef0b5543c29dc36030c38c25fd.png
[format_png 5]: /images/20230529/2171b92003b545ce950ef07061c21dc9.png