# 一、安全组
安全组：防火墙，屏蔽或开启，端口和ip地址

阿里云安全组默认规则是：白名单，默认拒绝

# 二、创建、使用

登录阿里云后，点击控制台

![image.png](/static/img/f7ee74e421c14c6b3bb02c3f04013644.image.webp)


点左上角
![image.png](/static/img/12c20ee06f9cd95b294a2448d3ccec22.image.webp)

![image.png](/static/img/75ce7f2ed46e3a3c5cfd41abe6b13e28.image.webp)


![image.png](/static/img/f890376c8c490ddf339b4573c7420722.image.webp)

选择自己专有网络的地区创建
![image.png](/static/img/0a0ea1a44465108ef35e36902c20f0bd.image.webp)

![image.png](/static/img/ed6676511388bfca2e41a018081c68b1.image.webp)

![image.png](/static/img/21051b77cbac7d65fec4a25a1ca14690.image.webp)

# 三、VPC
VPC虚拟专有网络，虚拟局域网，可以自定义网段

![image.png](/static/img/0e4bb41d3ccc83255c2dcf6bb04b5ec6.image.webp)

可以根据地区去创建对应的VPC
![image.png](/static/img/d27808ae5aacc21fb2cc4bef4bcf10e4.image.webp)

创建VPC
![image.png](/static/img/4f1d6f79008fe227d6659414f384b66a.image.webp)

1.安全组&VPC

# 一、ansible介绍
- Ansible 是一款 自动化运维工具，核心作用是通过代码（YAML 格式的剧本 / Playbook）批量管理多台服务器，实现「自动化部署、配置管理、任务执行」等操作，减少人工重复劳动，提高运维效率和一致性。
- 简单说，它能帮你做这些事：
- 批量在 100 台服务器上安装软件（比如统一装 Nginx）；
- 远程修改多台机器的配置文件（比如批量调整内核参数）；
- 一键启停 / 重启所有服务器上的某个服务（比如统一重启数据库）；
- 自动部署应用（从代码拉取到服务启动全流程自动化）；
- 收集所有主机的硬件 / 系统信息（比如内存、IP、系统版本）。
- 它的特点是 无需在远程主机安装客户端（基于 SSH 通信），配置简单，用 YAML 语法编写剧本，上手门槛较低，广泛用于中小型企业的自动化运维场景。

# 二、ansible模块
- command/shell：远程执行命令（shell 支持管道、重定向等语法，command 不支持）
- copy：将本地文件 / 目录推送到远程主机，可设置权限、属主等
- yum：远程通过 yum 安装、卸载、更新软件（适用于 CentOS/RHEL 等）
- service：远程管理服务启停、重启、重载，兼容 SysVinit 和 systemd
- script：本地脚本直接在远程主机执行（无需先传文件）
- file：远程管理文件 / 目录（创建、删除、权限）、软链接 / 硬链接
- group：远程创建、删除用户组，修改组 ID（GID）
- user：远程创建、删除用户，设置密码、所属组、家目录等
- cron：远程添加、修改、删除定时任务（crontab）
- mount：远程挂载文件系统（磁盘、NFS 等），并写入 /etc/fstab 实现开机自动挂载
- get_url：远程主机下载文件（从 URL 获取）
- systemd：通过 systemd 管理服务（启停、开机自启等，适用于 systemd 系统）
- selinux：远程开启、关闭或配置 SELinux 模式
- setup：收集远程主机的系统信息（如硬件、系统版本等）


# 三、ansible安装

```js
yum install -y epel-release
yum  install -y ansible
```

# 四、基本配置
## 进入到ansible目录下，修改hosts即可配置管理对应的主机
```js
vi /etc/ansible/hosts
```
方式一：操作时是根据模块去操作的，直接填写服务器账号密码
```yml
[elk1]
192.168.183.112

[elk2]
192.168.183.113

# 因为两个组的登录信息完全相同，可合并为一个公共组变量（更简洁）
[elk1:vars]
ansible_user=root
ansible_password=12345678

[elk2:vars]
ansible_user=root
ansible_password=12345678
```
![image.png](/static/img/8c4a37b60eb76b76df1d8cbe82b020e6.image.webp)

方式二就是做服务器之间的免密登录


## /etc/ansible/ansible.cfg（主配置文件）

作用：设置 Ansible 的全局默认参数（如远程用户、超时时间、提权配置等），覆盖默认行为。
```js
vi /etc/ansible/ansible.cfg
```


```yml
[defaults]
# 1. 主机清单路径（默认就是/etc/ansible/hosts，可省略）
inventory      = /etc/ansible/hosts

# 2. 默认远程登录用户（如果hosts文件没指定，就用这个）
remote_user    = root  # 比如你远程主机用root登录，就设为root

# 3. SSH私钥路径（如果用密钥登录，指定私钥位置）
#private_key_file = /root/.ssh/id_rsa  # 控制机上的私钥路径

# 4. 关闭主机密钥检查（首次连接无需手动确认yes，适合自动化）
host_key_checking = False

# 5. 连接超时时间（秒），避免卡太久
timeout        = 10

# 6. 并行任务数（同时操作的主机数量，默认5，主机多就调大）
forks          = 20

# 7. 不生成重试文件（避免冗余的.retry文件）
retry_files_enabled = False

# 8. 关闭过时功能警告（减少输出噪音）
deprecation_warnings = False


[privilege_escalation]
# 1. 是否自动提权（比如需要执行sudo命令时，自动切换到root）
become = False

# 2. 提权方式（常用sudo，也可以是su）
#become_method = sudo

# 3. 提权到目标用户（默认root）
#become_user = root

# 4. 是否需要输入sudo密码（如果远程主机sudo免密，设为False）
#become_ask_pass = False
```

测试
```js
重启
systemctl restart sshd

# 测试elk1组的连通性
ansible elk1 -m ping

# 测试在elk2主机执行sudo命令（比如查看/root目录，需要root权限）
ansible elk2 -m command -a 'ls /root'
```


![image.png](/static/img/09bbd2cbf35da1d33ef83dfa06d035ca.image.webp)

1.Ansible介绍和安装

1.kafka中的broker是干什么的?
- 每个kafka的节点都称为一个broker，broker是消息的代理，Producers往Brokers里面的指定Topic中写消息，Consumers从Brokers里面拉取指定Topic的消息，然后进
- 行业务处理，broker在中间起到一个代理保存消息的中转站。

2.kafka中的zookeeper起到什么作用，可以不用zookeeper么？
- zookeeper是一个分布式的协调组件，早期版本的kafka用zk做meta元数据信息存储，consumer的消费状态，group的管理以及ofset的值。
- 考虑到zk本身的一些因素以及整个架构较大概率存在单点问题，新版本中逐渐弱化了zookeeper的作用。
- 新的consumer使用了kafka内部的group coordination协议，也减少了对zookeeper的依赖，
- 但是broker依然依赖于zK，zookeeper在kafka中还用来选举controller和检测broker是否存活等等。

3.kafka中follower的作用？
- 读写都是由leader处理，follower只是作备份数据功能，从leader那将数据同步过来进行备份，不对外提供服务。

:::tip{title="提示"}
1. 关于 broker：除了作为消息的中转存储节点，broker 之间是相互独立的，共同组成 Kafka 集群。每个 broker 会存储部分 topic 的分区数据，且通过集群机制实现负载均衡和高可用。同时，broker 还负责处理生产者的写入请求、消费者的读取请求，并与 ZooKeeper 协作作用于集群元数据的维护。
2. 关于 ZooKeeper：虽然新版本 Kafka（如 2.8.0+）引入了 KRaft 模式，实现了不依赖 ZooKeeper 的部署（通过内置的控制器集群管理元数据），但在传统模式下，ZooKeeper 核心作用还包括：记录分区的 leader/follower 分配、维护 broker 集群的成员关系、存储 topic 的配置信息（如分区数、副本数等）。KRaft 模式的出现正是为了彻底摆脱对 ZooKeeper 的依赖，简化架构并提升稳定性。
3. 关于 follower：除了备份数据，follower 还有一个关键作用：当 leader 节点故障时，follower 会通过选举成为新的 leader，从而保证分区的可用性。follower 持续从 leader 同步数据，只有当数据同步到一定程度（如 ISR 列表中的 follower），才具备成为新 leader 的资格，这确保了故障转移时的数据一致性。
:::

6.kafka面试题

# 一、环境部署


```js
准备的机器
192.168.201.112
192.168.201.113
```
都安装好java jdk
```js
rpm -ivh  jdk-21_linux-x64_bin.rpm
```

都解压压缩包

```js
tar -zxvf apache-zookeeper-3.9.4-bin.tar.gz
tar -zxvf kafka_2.12-3.6.2.tgz
```

# 二、修zookeepe改配置文件
创建存储数据目录

```js
mkdir -p /kafka/zookeeper_data
```
修改192.168.201.112、192.168.201.113机器的配置文件
```js
vi  /kafka/apache-zookeeper-3.9.4-bin/conf/zoo.cfg
```

```js
# ZooKeeper 的时间基准单位（毫秒），所有时间相关配置都以它为倍数
# 比如后续的 initLimit、syncLimit 都是 "tickTime 的倍数"
tickTime=2000

# 从节点（Follower）启动后，与主节点（Leader）同步初始数据的最大超时时间
# 计算方式：initLimit × tickTime = 10 × 2000ms = 20秒
# 超过这个时间未同步成功，从节点会被判定为连接失败
initLimit=10

# 运行中，从节点与主节点之间通信的最大超时时间（如请求响应超时）
# 计算方式：syncLimit × tickTime = 5 × 2000ms = 10秒
# 超过这个时间未收到响应，主节点会认为从节点失联
syncLimit=5

# 数据存储目录（核心！必须改）
# 存储内容：内存快照（snapshot）、事务日志（记录所有写操作，用于恢复数据）
# 注意：/tmp 是临时目录，Linux 重启后会自动清空，生产环境必须改成永久目录
# 示例：dataDir=/data/zookeeper（需提前创建该目录并授权）
dataDir=/kafka/zookeeper_data

# 客户端连接 ZooKeeper 的端口号（默认2181，需确保防火墙开放）
# 外部程序（如Kafka）通过此端口连接，例如 Kafka 配置 zookeeper.connect=IP:2181
clientPort=2181

# 限制单个客户端（如一个Kafka实例）的最大并发连接数（默认60）
# 若出现"连接数超限"报错，可取消注释并调大，例如：maxClientCnxns=100
#maxClientCnxns=60

# 以下是自动清理旧数据的配置（生产环境建议开启，避免磁盘占满）
# 请先阅读官方维护指南再开启：https://zookeeper.apache.org/doc/current/zookeeperAdmin.html#sc_maintenance

# 自动保留的最新快照文件数量（取消注释后生效）
# 例如设置为3，表示只保留最近3个快照，更早的会被删除
#autopurge.snapRetainCount=3

# 自动清理任务的执行频率（单位：小时，取消注释后生效）
# 设置为0表示禁用自动清理；设置为1表示每1小时清理一次
#autopurge.purgeInterval=1

## 监控指标配置（对接Prometheus时使用，默认关闭）
#
# Prometheus监控 exporter 配置
#metricsProvider.className=org.apache.zookeeper.metrics.prometheus.PrometheusMetricsProvider
# 监控数据暴露的IP（0.0.0.0表示允许所有IP访问）
#metricsProvider.httpHost=0.0.0.0
# 监控数据暴露的端口（需开放此端口给Prometheus）
#metricsProvider.httpPort=7000
# 是否导出JVM信息（如内存、CPU使用情况）
#metricsProvider.exportJvmInfo=true

# 添加集群编号
server.0=192.168.201.112:2888:3888
server.1=192.168.201.113:2888:3888
```
指定集群编号192.168.201.112

```js
touch /kafka/zookeeper_data/myid
echo 0 > /kafka/zookeeper_data/myid
```
指定集群编号192.168.201.113

```js
touch /kafka/zookeeper_data/myid
echo 1 > /kafka/zookeeper_data/myid
```

然后这两台机器进入目录启动zookeeper
```js
cd /kafka/apache-zookeeper-3.9.4-bin/bin
./zkServer.sh start
```

这样就一个领导者一个工作者。（ leader领导者，follower：跟随者）
![image.png](/static/img/0d764541487333196e7f85cb571b9c2b.image.webp)

![image.png](/static/img/2f8e7c85cad8c6cf5a7884d4b2a8b882.image.webp)

# 三、修改Kakfa集群

创建Kafka的数据存储目录

```js
mkdir -p /kafka/data
```
修改配置文件
```js
vi /kafka/kafka_2.12-3.6.2/config/server.properties
```

服务器192.168.201.112配置
```js
############################# 服务器基本配置（ZooKeeper 模式） #############################
# 当前节点的唯一 ID（集群中不可重复）
broker.id=0

# ZooKeeper 集群连接地址（填你的 ZooKeeper 实际 IP:端口，单机就填一个）
zookeeper.connect=192.168.201.112:2181,192.168.201.113:2181
# 连接 ZooKeeper 的超时时间（默认 6000ms，无需修改）
zookeeper.connection.timeout.ms=6000


############################# socket 服务器设置（保留，无需修改） #############################
# 服务器监听的网络地址和端口（仅保留 PLAINTEXT 即可，删除 CONTROLLER 监听器）
listeners=PLAINTEXT://:9092
# broker 之间通信使用的监听器名称
inter.broker.listener.name=PLAINTEXT
# 向客户端暴露的连接地址（填 Kafka 实际 IP）
advertised.listeners=PLAINTEXT://192.168.201.112:9092
# 监听器与安全协议的映射（删除 CONTROLLER 相关映射）
listener.security.protocol.map=PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL

# 网络和 IO 线程数（保留默认）
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600


############################# 日志存储基本配置（保留，无需修改） #############################
# 消息日志存储目录（你的配置是 /kafka/data，没问题）
log.dirs=/kafka/data
num.partitions=1
num.recovery.threads.per.data.dir=1


############################# 内部主题、日志刷新/保留策略（全部保留，无需修改） #############################
offsets.topic.replication.factor=1
share.coordinator.state.topic.replication.factor=1
share.coordinator.state.topic.min.isr=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1

#log.flush.interval.messages=10000
#log.flush.interval.ms=1000

log.retention.hours=168
#log.retention.bytes=1073741824
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
```

服务器192.168.201.113配置

```js
############################# 服务器基本配置（ZooKeeper 模式） #############################
# 当前节点的唯一 ID（集群中不可重复）
broker.id=1

# ZooKeeper 集群连接地址（填你的 ZooKeeper 实际 IP:端口，单机就填一个）
zookeeper.connect=192.168.201.112:2181,192.168.201.113:2181
# 连接 ZooKeeper 的超时时间（默认 6000ms，无需修改）
zookeeper.connection.timeout.ms=6000


############################# socket 服务器设置（保留，无需修改） #############################
# 服务器监听的网络地址和端口（仅保留 PLAINTEXT 即可，删除 CONTROLLER 监听器）
listeners=PLAINTEXT://:9092
# broker 之间通信使用的监听器名称
inter.broker.listener.name=PLAINTEXT
# 向客户端暴露的连接地址（填 Kafka 实际 IP）
advertised.listeners=PLAINTEXT://192.168.201.113:9092
# 监听器与安全协议的映射（删除 CONTROLLER 相关映射）
listener.security.protocol.map=PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL

# 网络和 IO 线程数（保留默认）
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600


############################# 日志存储基本配置（保留，无需修改） #############################
# 消息日志存储目录（你的配置是 /kafka/data，没问题）
log.dirs=/kafka/data
num.partitions=1
num.recovery.threads.per.data.dir=1


############################# 内部主题、日志刷新/保留策略（全部保留，无需修改） #############################
offsets.topic.replication.factor=1
share.coordinator.state.topic.replication.factor=1
share.coordinator.state.topic.min.isr=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1

#log.flush.interval.messages=10000
#log.flush.interval.ms=1000

log.retention.hours=168
#log.retention.bytes=1073741824
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
```


然后进入bin目录启动kafka

```js
cd /kafka/kafka_2.12-3.6.2/bin/
```

```js
./kafka-server-start.sh -daemon ../config/server.properties
```

# 三、生产者和消费者测试
创建集群主题
```js
./kafka-topics.sh --bootstrap-server 192.168.201.112:9092,192.168.201.113:9092 --create --topic test_topic --partitions 3 --replication-factor 1
```
查看所有主题
```js
./kafka-topics.sh --bootstrap-server 192.168.201.112:9092,192.168.201.113:9092 --list
```

启动一个手动输入消息的生产者，向 test_topic 发送消息
```js
./kafka-console-producer.sh --bootstrap-server 192.168.201.112:9092,192.168.201.113:9092 --topic test_topic
```
启动消费者，订阅 test_topic 接收消息

```js
./kafka-console-consumer.sh --bootstrap-server 192.168.201.112:9092,192.168.201.113:9092 --topic test_topic --from-beginning
```

5.Kafka集群部署

# 一、使用kafka自带的命令创建主题
创建一个名为 test_topic 的主题，设置 3 个分区（提高并行度）、1 个副本（单机模式只能设 1，集群模式可设 2+）：

 创建主题命令：--bootstrap-server 指定 Kafka 地址，--topic 是主题名，--partitions 分区数，--replication-factor 副本数
```shell
./kafka-topics.sh --bootstrap-server 192.168.201.113:9092 --create --topic test_topic --partitions 3 --replication-factor 1
```

执行成功后会提示：Created topic test_topic.

![image.png](/static/img/08325bb87de48d727527e31bf5dfcf62.image.webp)

查看主题详情，确认创建成功：

```js
./kafka-topics.sh --bootstrap-server 192.168.201.113:9092 --describe --topic test_topic
```

![image.png](/static/img/cdaf844d2945fc1cfc0e00600c18049a.image.webp)

# 二、启动命令行生产者（发送消息）

启动一个手动输入消息的生产者，向 test_topic 发送消息（需要新开一个终端窗口执行，避免和后续消费者冲突）：
启动生产者命令：--bootstrap-server 指定 Kafka 地址，--topic 指定要发送的主题
```js
./kafka-console-producer.sh --bootstrap-server 192.168.201.113:9092 --topic test_topic
```
执行后会进入输入模式（显示 > 提示符），直接输入任意内容（比如 “hello kafka”“测试消息 123”），按回车就会发送消息；

![image.png](/static/img/f7dfa7aa0233d4a0773b7fd0a546ec94.image.webp)
想停止生产者，按 Ctrl+C 即可。

# 三、启动命令行消费者（接收消息）
再新开一个终端窗口，启动消费者，订阅 test_topic 接收消息（支持实时接收生产者发送的新消息，也能接收历史消息）：
启动消费者命令：--from-beginning 表示“从最早的消息开始接收”（不加则只收新消息）
```js
./kafka-console-consumer.sh --bootstrap-server 192.168.201.113:9092 --topic test_topic --from-beginning
```
执行后，消费者会自动接收 test_topic 里的所有消息（包括之前生产者发送的历史消息）；
此时回到生产者窗口，再输入新消息（比如 “这是新消息”），消费者窗口会实时显示这条消息，说明消息收发正常；


![image.png](/static/img/af812d3859f52a22a7ac0ad9b86c7205.image.webp)
想停止消费者，按 Ctrl+C 即可。

# 四、删除测试主题
如果后续不需要 test_topic，可以执行命令删除：

```js
./kafka-topics.sh --bootstrap-server 192.168.201.113:9092 --delete --topic test_topic
```