windows安装kafka配置SASL-PLAIN安全认证

目录

1.Windows安装zookeeper：

1.1下载zookeeper

1.2 解压之后如图二

1.3创建日志文件

1.4复制 “zoo_sample.cfg” 文件

1.5更改 “zoo.cfg” 配置

1.6新建zk_server_jaas.conf

1.7修改zkEnv.cmd

1.8导入相关jar

1.9以上配置就配好啦，接下来启动。

1.10连接测试

2.Windows下安装运行kafka：

2.1下载kafka：

2.2 新建kafka_server_jaas.conf

2.3 新建kafka_client_jaas.conf

2.4 新建kafka_topic_jaas.properties

2.5 修改server.properties配置文件：

2.6 修改consumer.properties和producer.properties

2.7 修改kafka-server-start.bat

2.8 修改kafka-console-consumer.bat和kafka-console-producer.bat

2.9 启动kafka：

2.10 测试 

1）创建主题

2）生产消息

3）消费消息

4) 查看主题配置

5)查看主题消息日志

3.将kafka安装为windows服务

3.1安装NSSM

3.2 kafka安装成win服务

3.3 zookeeper安装为windows服务

3.4 BAT一键安装为Windows服务

3.5 nssm常用命令

3.6总结

4.confluent简介

4.1官网

4.2下载地址

4.3物理机安装参考

4.4 docker安装参考

4.4 Confluent的组件

4.5 Confluent的安装

4.5.1下载地址：

4.5.2下载confluent-7.2.1

4.5.3解压到指定目录下

4.5.4配置zookeeper

4.5.5配置kafka的broker

4.5.6配置rest proxy

4.5.7配置schema registry

4.5.8启动kafka-rest

4.5.9启动zookeeper

4.5.10启动kafka broker

4.5.11启动schema registry

4.6测试使用

4.6.1查看topics

4.6.2查看集群的brokers

4.6.3注册consumer group

4.6.4把topic和消费者my_consumer关联起来

4.6.5通过rest接口向bear push数据

4.6.6通过rest接口消费数据

4.6.7删除注册的consumer实例：

4.6.8更多信息参考

---

1.Windows安装zookeeper：

1.1下载zookeeper

http://zookeeper.apache.org/releases.html，选择自己需要的版本，并解压。

坑一：我下载的是“apache-zookeeper-3.8.2”，配置完成之后启动的时候发现报错，如图一

图一

结果发现从目前的最新版本3.5.5开始，带有bin名称的包才是我们想要的下载可以直接使用的里面有编译后的二进制的包，而之前的普通的tar.gz的包里面是只是源码的包无法直接使用。如果下载3.5.5以后的版本的Zookeeper安装包，需要下载带有bin标识的包就不会有问题了。(参考：https://www.cnblogs.com/zhoading/p/11593972.html)

1.2 解压之后如图二

图二

1.3创建日志文件

在上图目录下创建文件夹 data和log，如图三

　图三

1.4复制 “zoo_sample.cfg” 文件

并重命名为 “zoo.cfg”，如图四

 图四

 

1.5更改 “zoo.cfg” 配置

为自己创建的文件夹 data 和 log 的路径，如图五。

dataDir=D:\\Java\\apache-zookeeper-3.8.2-bin\\data

dataLogDir=D:\\Java\\apache-zookeeper-3.8.2-bin\\log

配置安全认证：

authProvider.1=org.apache.zookeeper.server.auth.SASLAuthenticationProvider

requireClientAuthScheme=sasl

jaasLoginRenew=3600000

  　　　　　　　　　　　　　　　

图五

我一开始用的单斜杠，会出现图六的一个警告。

图六

1.6新建zk_server_jaas.conf

在Zookeeper安装目录conf中，新建zk_server_jaas.conf文件，添加如下内容：

Server {

     org.apache.kafka.common.security.plain.PlainLoginModule required

     username="admin"

     password="admin- secret";

};

1.7修改zkEnv.cmd

在Zookeeper安装目录bin中，打开zkEnv.cmd进行编辑，在该文件中set ZOO_LOG_DIR=%~dp0%..\logs下一行加入如下配置：注意上述配置是斜杠，不是反斜杠

set SERVER_JVMFLAGS=-Djava.security.auth.login.config=D:/Java/apache-zookeeper-3.8.2-bin/config/zk_server_jaas.conf

1.8导入相关jar

在Kafka安装目录libs中分别找到如下jar，复制它们到Zookeeper安装目录的lib中即可

​   kafka-clients-3.5.1.jar

   lz4-java-1.8.0.jar

   slf4j-api-1.7.36.jar

   slf4j-reload4j-1.7.36.jar

   snappy-java-1.1.10.1.jar

1.9以上配置就配好啦，接下来启动。

直接双击bin 目录下的 “zkServer.cmd"，如图七

图七

 启动完成，如图八。

图八

1.10连接测试

双击bin目录下的 zkCli.cmd，如图九

图九

2.Windows下安装运行kafka：

2.1下载kafka：

http://kafka.apache.org/downloads.html  （注意下载binary版本的）

2.2 新建kafka_server_jaas.conf

在Kafka安装目录config中，新建kafka_server_jaas.conf文件，添加如下内容：

KafkaServer {

    org.apache.kafka.common.security.plain.PlainLoginModule required

    username="admin"

    password="admin-secret"

    user_admin="admin-secret";

};

说明：在KafkaServer部分，username和password是broker用于初始化连接到其他的broker在上面配置中，admin用户为broker间的通讯。

    user_userName定义了所有连接到broker和broker验证的所有的客户端连接包括其他broker的用户密码，user_userName必须配置admin用户，否则会报错

2.3 新建kafka_client_jaas.conf

在Kafka安装目录config中，新建kafka_client_jaas.conf文件，添加如下内容：

KafkaClient {

    org.apache.kafka.common.security.plain.PlainLoginModule required

    username="admin"

    password="admin-secret";

};

说明：在KafkaClient部分，username和password是客户端用来配置客户端连接broker的用户

2.4 新建kafka_topic_jaas.properties

在Kafka安装目录config中，新建kafka_topic_jaas.properties文件，添加如下内容：

sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="admin" password="admin-secret";

security.protocol=SASL_PLAINTEXT

sasl.mechanism=PLAIN

2.5 修改server.properties配置文件：

在Kafka安装目录config中，打开server.properties进行编辑，在该文件中加入如下配置：

# 内网ip

listeners=SASL_PLAINTEXT://192.168.0.108:9092

# 外网ip

#advertised.listeners=SASL_PLAINTEXT://192.168.0.108:9092

# 使用的认证协议

security.inter.broker.protocol=SASL_PLAINTEXT

# SASL机制

sasl.enabled.mechanisms=PLAIN

sasl.mechanism.inter.broker.protocol=PLAIN

# 完成身份验证的类

#authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer

# 如果没有找到ACL（访问控制列表）配置，则允许任何操作。

allow.everyone.if.no.acl.found=false

auto.create.topics.enable=true

delete.topic.enable=true

super.users=User:admin

2.6 修改consumer.properties和producer.properties

 在Kafka安装目录config中，打开consumer.properties和producer.properties进行编辑，在该文件中加入如下配置：

security.protocol=SASL_PLAINTEXT

sasl.mechanism=PLAIN

修改consumer.propertie中group.id=TestGroup，若有多个消费客户端需保证group.id不重复，否则无法同时消费消息

2.7 修改kafka-server-start.bat

在Kafka安装目录bin\windows中，打开kafka-server-start.bat进行编辑，在该文件中SetLocal关键字下一行加入如下配置：

set KAFKA_OPTS=-Djava.security.auth.login.config=file:%~dp0../../config/kafka_server_jaas.conf

2.8 修改kafka-console-consumer.bat和kafka-console-producer.bat

在Kafka安装目录bin\windows中，打开kafka-console-consumer.bat和kafka-console-producer.bat进行编辑，在该文件中SetLocal关键字下一行加入如下配置：

set KAFKA_OPTS=-Djava.security.auth.login.config=file:%~dp0../../config/kafka_client_jaas.conf

2.9 启动kafka：

win+R 输入cmd，打开命令窗口，进入安装目录 D:\Java\kafka_2.13-3.5.1，输入以下命令 ( 成功之后不要关闭窗口)

.\bin\windows\kafka-server-start.bat config\server.properties　　

2.10 测试 

1）创建主题

打开一个新的命令窗口，进入安装目录D:\Java\kafka_2.13-3.5.1，输入以下命令 ( 成功之后不要关闭窗口)

.\bin\windows\kafka-topics.bat --create --bootstrap-server 192.168.0.108:9092 --replication-factor 1 --partitions 1 --topic mdlp --command-config config\kafka_topic_jaas.properties

 查看所有主题命令：

.\bin\windows\kafka-topics.bat --list --bootstrap-server 192.168.0.108:9092 --command-config config\kafka_topic_jaas.properties

2）生产消息

打开一个新的命令窗口，进入安装目录 D:\Java\kafka_2.13-3.5.1，输入以下命令 ( 成功之后不要关闭窗口)

.\bin\windows\kafka-console-producer.bat --broker-list 192.168.0.108:9092 --topic mdlp --producer.config config\producer.properties

输入内容：hello word

3）消费消息

打开一个新的命令窗口，进入安装目录 D:\Java\kafka_2.13-3.5.1，输入以下命令 ( 成功之后不要关闭窗口)

.\bin\windows\kafka-console-consumer.bat --bootstrap-server  192.168.0.108:9092 --topic mdlp --consumer.config config\consumer.properties

这里接收生产者的消息。

4)  查看主题配置

打开一个新的命令窗口，进入安装目录 D:\Java\kafka_2.13-3.5.1，输入以下命令 ( 成功之后不要关闭窗口)

.\bin\windows\kafka-configs.bat --bootstrap-server 192.168.0.108:9092 --entity-type topics --entity-name mdlp --describe --command-config config\kafka_topic_jaas.properties

5)  查看主题消息日志

打开一个新的命令窗口，进入安装目录 D:\Java\kafka_2.13-3.5.1，输入以下命令 ( 成功之后不要关闭窗口)

.\bin\windows\kafka-run-class.bat kafka.tools.DumpLogSegments --print-data-log --files kafka-logs\mdlp-0\00000000000000000000.log

3.将kafka安装为windows服务

3.1安装NSSM

NSSM是一个服务封装程序，它可以将普通exe、bat、以及任何程序封装成服务，使之像windows服务一样运行，就像一个服务壳一样，将你的程序包在NSSM里面。

NSSM如何使用

1. 下载NSSM download.
2. 根据自己的平台，将32/64位nssm.exe文件解压至任意文件夹。
3. cmd定位至nssm.exe所在目录。
4. 输入 nssm install {服务名称}，即注册服务的名称。注册服务弹出如下NSSM界面。

3.2 kafka安装成win服务

按照上面的步骤操作就可以了，核心就是下面的截图，也可以添加依赖服务。

Path: D:\Java\kafka_2.13-3.5.1\bin\windows\kafka-server-start.bat
Startup directory: D:\Java\kafka_2.13-3.5.1\bin\windows
Arguments: D:\Java\kafka_2.13-3.5.1\config\server.properties

Path参数为kafka的启动命令，一般在windows目录。Startup directory为启动目录，Arguments是启动参数，只有启动参数需要手工填一下，当你在Path点右边按钮选择后会自动的填上前面2个参数。自己操作体验下就明白了。完了之后点击 Install Service 就可以了。

如果有依赖服务，按下图所示填上即可。

3.3 zookeeper安装为windows服务

zookeeper也是可以安装为windows服务的，操作如下图所示：

没有参数就不用填。如果嫌界面操作比较麻烦，可以制作成批处理文件一次性安装完毕。

3.4 BAT一键安装为Windows服务

@echo off
@echo 安装zookeeper
nssm install zookeeper D:\Java\apache-zookeeper-3.8.2-bin\bin\zkServer.cmd
@echo 安装kafka
nssm install kafka D:\Java\kafka_2.13-3.5.1\bin\windows\kafka-server-start.bat D:\Java\kafka_2.13-3.5.1\config\server.properties
@echo 启动zookeeper服务
nssm start zookeeper
@echo 启动kafka服务
nssm start kafka
pause

上面的命令是通过批处理文件一键安装为windows服务，并且启动zookeeper和kafka服务。你可以直接拿过去使用，注意需要修改下参数，例如将D:\Java\apache-zookeeper-3.8.2-bin\bin\zkServer.cmd修改为你服务器上的zookeeper目录，而D:\Java\kafka_2.13-3.5.1\bin\windows\kafka-server-start.bat D:\Java\kafka_2.13-3.5.1\config\server.properties也是对应的修改为kafka的路径。

如果要卸载的话，使用下面的bat即可。

@echo off
@echo 卸载zookeeper
nssm remove zookeeper confirm
@echo 卸载kafka
nssm remove kafka confirm
pause

将以上代码复制下来，用记事本新建一个文件，粘贴进去，并将后缀改为bat即可，使用时双击运行就完事了。

3.5 nssm常用命令

1. nssm install servername //创建servername服务
2. nssm start servername //启动服务
3. nssm stop servername //暂停服务
4. nssm restart servername //重新启动服务
5. nssm remove servername //删除创建的servername服务

3.6总结

NSSM可以将控制台程序一样的安装为服务，再配合定时任务，可以做好多事情。

4.confluent简介

        confluent是平台化的工具，封装了kafka，让我们可以更方便的安装和使用监控kafka，作用类似于CDH对于Hadoop。

        confluent是由LinkedIn开发出Apache Kafka的团队成员，基于这项技术创立了新公司Confluent，Confluent的产品也是围绕着Kafka做的。

4.1官网

https://www.confluent.io

4.2下载地址

https://www.confluent.io/download/

4.3物理机安装参考

https://docs.confluent.io/platform/current/quickstart/ce-docker-quickstart.html#cos-quickstart

4.4 docker安装参考

https://docs.confluent.io/platform/current/quickstart/ce-docker-quickstart.html#cos-docker-quickstart

对比之后感觉比原生的kafka安装简单很多，容器是docker容器的版本，对于我们在k8s中使用很方便。

4.4 Confluent的组件

Confluent Platform 包括更多的工具和服务，使构建和管理数据流平台更加容易。

Confluent Control Center（闭源）。管理和监控Kafka最全面的GUI驱动系统。

Confluent Kafka Connectors（开源）。连接SQL数据库/Hadoop/Hive

Confluent Kafka Clients（开源）。对于其他编程语言，包括C/C++,Python

Confluent Kafka REST Proxy（开源）。允许一些系统通过HTTP和kafka之间发送和接收消息。

Confluent Schema Registry（开源）。帮助确定每一个应用使用正确的schema当写数据或者读数据到kafka中。

4.5 Confluent的安装

4.5.1下载地址：

http://www.confluent.io/download

打开后，显示最新版本，在右边填写信息后，点击Download下载。

本次我们主要使用REST Proxy，当然底层的broker也是使用confluent的kafka组件，下面简述安装步骤：

4.5.2下载confluent-7.2.1

Wget https://packages.confluent.io/archive/7.2/confluent-community-7.2.1.tar.gz

tar -xvf confluent-community-7.2.1.tar.gz

4.5.3解压到指定目录下

通过查看目录的内容，能够发现，confluent里面是含有kafka的，也就是说，如果你没有安装kafka，那么可以通过confluent直接对kafka进行安装。如果已经安装了kafka，可以使用confluent提供的插件。

自定义配置

我们可以配置自己需要的和对应配置信息

进入解压出来的confluent-7.2.1

cd confluent-7.2.1

4.5.4配置zookeeper

vi etc/kafka/zookeeper.properties

内容如下：

dataDir=/confluent-7.2.1/data/zookeeper

clientPort=2181

maxClientCnxns=0

4.5.5配置kafka的broker

vi etc/kafka/server.properties

内容如下：

broker.id=0

delete.topic.enable=true

listeners=PLAINTEXT://192.168.176.131:9092

num.network.threads=3

num.io.threads=8

socket.send.buffer.bytes=102400

socket.receive.buffer.bytes=102400

socket.request.max.bytes=104857600

log.dirs=/confluent-7.2.1/data/kafka

num.partitions=1

num.recovery.threads.per.data.dir=1

log.retention.hours=168

log.segment.bytes=1073741824

log.retention.check.interval.ms=300000

zookeeper.connect=192.168.176.131:2181

zookeeper.connection.timeout.ms=6000

confluent.support.metrics.enable=true

confluent.support.customer.id=anonymous

4.5.6配置rest proxy

vi etc/kafka-rest/kafka-rest.properties

内容如下：

id=kafka-rest-server

#zookeeper.connect=192.168.176.131:2181

bootstrap.servers=PLAINTEXT://192.168.176.131:9092

4.5.7配置schema registry

vi etc/schema-registry/schema-registry.properties

内容如下：

listeners=http://192.168.176.131:8081

kafkastore.connection.url=192.168.176.131:2181

kafkastore.bootstrap.servers=PLAINTEXT://192.168.176.131:9092

kafkastore.topic=_schemas

debug=false

启动服务

4.5.8启动kafka-rest

bin/kafka-rest-start etc/kafka-rest/kafka-rest.properties

上面的这种方式是前台启动，也可以以后台方式启动。

nohup bin/kafka-rest-start etc/kafka-rest/kafka-rest.properties &

4.5.9启动zookeeper

bin/zookeeper-server-start etc/kafka/zookeeper.properties

4.5.10启动kafka broker

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

4.5.11启动schema registry

bin/schema-registry-start etc/schema-registry/schema-registry.properties

4.6测试使用

4.6.1查看topics

浏览器访问或者curl都可以

http://192.168.176.131:8082/topics

4.6.2查看集群的brokers

curl http://192.168.176.131:8082/brokers

4.6.3注册consumer group

curl -X POST -H "Content-Type:application/vnd.kafka.v2+json" -H "Accept: application/vnd.kafka.v2+json"   --data '{"name": "my_consumer_instance", "format": "json", "auto.offset.reset": "earliest"}'   http://192.168.176.131:8082/consumers/my_test_consumer

4.6.4把topic和消费者my_consumer关联起来

curl -X POST -H "Content-Type:application/vnd.kafka.v2+json" --data '{"topics":["bear"]}'   http://192.168.176.131:8082/consumers/my_test_consumer/instances/my_consumer_instance/subscription

4.6.5通过rest接口向bear push数据

curl -X POST -H "Content-Type:application/vnd.kafka.json.v2+json" --data '{"records":[{"value":{"name": "testUser"}}]}'  "http://192.168.176.131:8082/topics/bear"

4.6.6通过rest接口消费数据

curl -X GET -H "Accept:application/vnd.kafka.json.v2+json" http://192.168.176.131:8082/consumers/my_test_consumer/instances/my_consumer_instance/records

4.6.7删除注册的consumer实例：

curl -X DELETE -H "Accept:application/vnd.kafka.v2+json"     http://192.168.176.131:8082/consumers/my_test_consumer/instances/my_consumer_instance

4.6.8更多信息参考

https://github.com/confluentinc/kafka-rest