2. 环境准备：Flink集群搭建（Standalone模式）、Kafka消息队列安装与配置

实时风控巡检系统依赖稳定的流计算引擎与高吞吐消息管道。本章将指导你完成 Flink Standalone 集群 与 Kafka 消息队列 的安装与配置，为后续 CEP 规则部署与告警阈值测试奠定基础。

2.1 环境概览与版本选型

本课程采用以下组件版本组合，经生产级压测验证，兼容性良好：

2.2 Flink Standalone 集群搭建

Standalone 模式适合开发测试与中小规模生产场景。以下步骤以 1 台 Master + 2 台 Worker 为例。

2.2.1 节点规划与前置条件

• Master 节点（192.168.1.10）：运行 JobManager、Dispatcher、ResourceManager。
• Worker 节点 1（192.168.1.11）：运行 TaskManager，分配 4 个 Slot。
• Worker 节点 2（192.168.1.12）：运行 TaskManager，分配 4 个 Slot。
• 所有节点已配置 SSH 免密登录，且 JDK 环境变量正确设置。

2.2.2 安装与配置步骤

1. 下载解压：在 Master 节点执行 wget https://archive.apache.org/dist/flink/flink-1.17.1/flink-1.17.1-bin-scala_2.12.tgz，解压至 /opt/flink。
2. 配置 flink-conf.yaml：

# /opt/flink/conf/flink-conf.yaml
jobmanager.rpc.address: 192.168.1.10
jobmanager.bind-host: 0.0.0.0
taskmanager.bind-host: 0.0.0.0
taskmanager.host: 192.168.1.11
taskmanager.numberOfTaskSlots: 4
parallelism.default: 2
rest.address: 192.168.1.10
rest.bind-address: 0.0.0.0

3. 配置 workers 文件：

# /opt/flink/conf/workers
192.168.1.11
192.168.1.12

4. 分发至 Worker 节点：使用 scp -r /opt/flink root@192.168.1.11:/opt/ 同步配置。
5. 启动集群：在 Master 节点执行 /opt/flink/bin/start-cluster.sh。
6. 验证：访问 http://192.168.1.10:8081，确认 Web UI 显示 2 个 TaskManager 共 8 个 Slot。

2.3 Flink Standalone 集群架构图

下图展示了 Standalone 模式下 JobManager 与 TaskManager 的通信关系，以及 Slot 资源分配方式：

2.4 Kafka 消息队列安装与配置

Kafka 负责接收交易事件流，作为 Flink Source 的输入管道。本课程采用 单节点 ZooKeeper + 单节点 Kafka Broker 的轻量部署，生产环境建议扩展为多 Broker 集群。

2.4.1 ZooKeeper 安装

1. 下载 ZooKeeper 3.8.1 并解压至 /opt/zookeeper。
2. 复制配置文件：cp conf/zoo_sample.cfg conf/zoo.cfg。
3. 修改 dataDir=/var/lib/zookeeper，并创建该目录。
4. 启动：/opt/zookeeper/bin/zkServer.sh start。
5. 验证：echo ruok | nc localhost 2181 返回 imok。

2.4.2 Kafka Broker 安装与配置

1. 下载 Kafka 3.4.0 并解压至 /opt/kafka。
2. 编辑 config/server.properties：

# /opt/kafka/config/server.properties
broker.id=0
listeners=PLAINTEXT://0.0.0.0:9092
advertised.listeners=PLAINTEXT://192.168.1.10:9092
log.dirs=/var/lib/kafka-logs
zookeeper.connect=localhost:2181
num.partitions=3
default.replication.factor=1
offsets.topic.replication.factor=1

3. 创建日志目录：mkdir -p /var/lib/kafka-logs。
4. 启动 Kafka：/opt/kafka/bin/kafka-server-start.sh -daemon config/server.properties。
5. 验证：创建测试主题 test-topic，使用生产者与消费者发送/接收消息，确认无报错。

2.5 实时交易事件流拓扑

下图展示了从交易事件产生到 Flink 消费的完整数据流路径：

2.6 连通性验证

完成安装后，执行以下命令验证 Flink 与 Kafka 的集成：

# 1. 在 Kafka 中创建风控主题
/opt/kafka/bin/kafka-topics.sh --create \
  --bootstrap-server localhost:9092 \
  --topic risk-trade-events \
  --partitions 3 \
  --replication-factor 1

# 2. 启动 Flink 自带 SocketWindowWordCount 示例（替换为 Kafka Source）
/opt/flink/bin/flink run \
  /opt/flink/examples/streaming/SocketWindowWordCount.jar \
  --host 192.168.1.10 --port 9999

# 3. 使用 kafka-console-producer 发送测试消息
echo '{"tradeId":"T001","amount":15000,"userId":"U100"}' | \
  /opt/kafka/bin/kafka-console-producer.sh \
  --broker-list localhost:9092 \
  --topic risk-trade-events

2.7 环境检查清单

使用下表逐项确认环境就绪：