【运维基础篇】-新版Kafka部署

新版Kafka部署

目录

消息队列

一、消息队列

消息队列

，缩写为

MQ

，从字面上来理解，消息队列是一种用来存储消息的队列，拥有先进先出(FIFO)的特性,主要用于不同线程和进程间的通信，处理一系列的输入请求。
消息队列采用的的是

异步通信机制

，即消息的发送者和接收者无须同时与消息队列进行数据交互，消息会一直保存在队列中，直到被接收者读取。每条消息记录都包含了详细的数据说明，如(数据产生时间、数据类型、特定的输入参数等)。

消息队列主要为解决

应用耦合

、

异步处理

、

流量削峰

等问题。

当前使用较多的消息队列有RabbitMQ、RocketMQ、ActiveMQ、Kafka、ZeroMQ、MetaMQ、部分数据库如Redis、MySQL等也可实现消息队列的功能。

消息队列通过将

消息的发送和接收分离

来实现应用程序的异步和解耦，但消息队列真正的目的是解决了异步的通信问题，消息队列屏蔽了底层复杂的通信协议。
消息队列分

有Broker的消息队列

（重Topic: Kafka、ActiveMQ、RocketMQ，轻Topic: 如RabbitMQ(或者说是AMQP)）和

无Broker的消息队列

（ZeroMQ）。

Kafka简介

Kafka在

2.8

版本之前，Kafka强依赖zookeeper来来负责集群元数据的管理，这也导致当Zookeeper集群性能发生抖动时，Kafka的性能也会收到很大的影响。2.8版本之后，kafka3.x开始提供KRaft（Kafka Raft，依赖Java 8+ ）模式，开始去除对zookeeper的依赖。最新的3.5版本中，Kafka依然兼容zookeeper Controller，但Kafka

Raft元数据模式

，已经可以在不依赖zookeeper的情况下独立启动Kafka了。

kafka(kraft)部署

安装文件下载地址：https://dlcdn.apache.org/kafka/

1.下载对应3.xx 版本安装包并解压

tar-xvf kafka_2.13-3.5.2.tgz -C /data
cd /data/kafka_2.13-3.5.2

2.生成对应的UUID

cd bin
sh kafka-storage.sh random-uuid > uuid
cat uuid
[root@centos bin]# cat uuid
Fm8c0IcvQ96qPW8fGkMsDQ

3.单机版本部署
修改配置文件

vim config/kraft/server.properties

具体配置项如下：

process.roles=broker,controller
[email protected]:9093
listeners=SSL://:9092,CONTROLLER://:9093
advertised.listeners=SSL://10.11.0.6:9092
controller.listener.names=CONTROLLER
listener.security.protocol.map=CONTROLLER:SSL,PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL
log.dirs=/data/kafka_2.13-3.5.2/data_logs

SSL配置

#sslssl.keystore.location=/data/kafka_2.13-3.5.2/ssl/server.keystore.jks
ssl.keystore.password=秘钥密码
ssl.key.password=秘钥密码
ssl.truststore.location=/data/kafka_2.13-3.5.2/ssl/server.truststore.jks
ssl.truststore.password=秘钥密码
#重要ssl.client.auth=none
security.inter.broker.protocol=SSL
#重要ssl.endpoint.identification.algorithm=ssl.enabled.protocols=TLSv1.2,TLSv1.1,TLSv1
ssl.keystore.type=JKS
ssl.truststore.type=JKS

4.使用前面生成的uuid格式化Kafka存储目录

sh kafka-storage.sh format-t`cat uuid`-c../config/kraft/server.properties
[root@centos bin]# sh kafka-storage.sh format -t `cat uuid` -c ../config/kraft/server.properties
Formatting /data/kafka_2.13-3.5.2/kraft-combined-logs

5.启动kafka服务

sh kafka-server-start.sh -daemon../config/kraft/server.properties

1. 查看进程、查看日志

ps-ef|grep kafka

tail-f../logs/server.log

7.集群部署
修改配置，如下：
第一台

process.roles=broker,controller
[email protected]:9093,[email protected]:9093,[email protected]:9093

listeners=SSL://:9092,CONTROLLER://:9093

advertised.listeners=SSL://10.11.0.6:9092

controller.listener.names=CONTROLLER

listener.security.protocol.map=CONTROLLER:SSL,PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL

第二台

process.roles=broker,controller

[email protected]:9093,[email protected]:9093,[email protected]:9093

listeners=SSL://:9092,CONTROLLER://:9093

advertised.listeners=SSL://10.11.0.7:9092

controller.listener.names=CONTROLLER

listener.security.protocol.map=CONTROLLER:SSL,PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL

第三台

process.roles=broker,controller
[email protected]:9093,[email protected]:9093,[email protected]:9093

listeners=SSL://:9092,CONTROLLER://:9093

advertised.listeners=SSL://10.11.0.8:9092

controller.listener.names=CONTROLLER

listener.security.protocol.map=CONTROLLER:SSL,PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL

SSL配置与单机一致

配置集群ID
一样要先生成集群ID，根据上面操作，只在

第一台

生成uuid就好。后面两台使用和第一台一样的ID。操作省略

格式化 kafka 存储目录（三台节点）
三台都要执行，uuid使用第一台生成的（要把第一台生成的uuid文件拷贝到另外的机器）

bin/kafka-storage.sh format-t`cat uuid`-c config/kraft/server.properties

启动三台服务

sh kafka-server-start.sh -daemon../config/kraft/server.properties

kafka操作命令：

# 创建topic
bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --topic test --command-config /data/kafka_2.13-3.5.2/config/client.properties
 
# 查看topic列表
bin/kafka-topics.sh --bootstrap-server localhost:9092 --list  --command-config /data/kafka_2.13-3.5.2/config/client.properties

# 查看消息详情
bin/kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic test --command-config /data/kafka_2.13-3.5.2/config/client.properties
 
# 生产消息
bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic test  --producer.config /data/kafka_2.13-3.5.2/config/client.properties
 
# 消费消息
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning  --consumer.config /data/kafka_2.13-3.5.2/config/client.properties

# 查看消费者组列表
bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --command-config /data/kafka_2.13-3.5.2/config/client.properties

client.properties

security.protocol=SSL
ssl.truststore.location=/data/kafka_2.13-3.5.2/ssl/server.truststore.jks
ssl.truststore.password=密钥密码
ssl.truststore.type=JKS
ssl.keystore.location=/data/kafka_2.13-3.5.2/ssl/server.keystore.jks
ssl.keystore.password=密钥密码
ssl.keystore.type=JKS
#重要重要重要，否则连不上
ssl.endpoint.identification.algorithm=

证书生成方式：生成CA数字证书