3、环境准备：Docker与Docker Compose搭建、Kafka单机/集群部署、Flink本地开发环境配置

说实话，做实时流水线最怕什么？不是业务逻辑复杂，而是环境折腾半天跑不起来。我见过太多人卡在“装软件”这一步，最后连 Hello World 都没见到就放弃了。

这一章，咱们就把地基打牢。我会带你一步步搞定 Docker、Kafka 和 Flink 的环境。嗯，都是我在项目中反复踩坑后总结出来的“最优解”。

3.1 Docker 与 Docker Compose：一切的基础

为什么非要用 Docker？说白了，就是为了“环境一致性”。你想想看，你在本地调好的配置，到测试环境就崩了，多闹心。Docker 把整个运行环境打包成镜像，走到哪都一样。

3.1.1 安装 Docker Engine

我个人习惯在 Linux 上做开发，但 macOS 和 Windows 也一样。这里以 Ubuntu 22.04 为例：

# 卸载旧版本
sudo apt-get remove docker docker-engine docker.io containerd runc

# 安装依赖
sudo apt-get update
sudo apt-get install ca-certificates curl gnupg lsb-release

# 添加 Docker 官方 GPG 密钥
sudo mkdir -p /etc/apt/keyrings
curl -fsSL https://download.docker.com/linux/ubuntu/gpg | sudo gpg --dearmor -o /etc/apt/keyrings/docker.gpg

# 设置仓库
echo \
  "deb [arch=$(dpkg --print-architecture) signed-by=/etc/apt/keyrings/docker.gpg] https://download.docker.com/linux/ubuntu \
  $(lsb_release -cs) stable" | sudo tee /etc/apt/sources.list.d/docker.list > /dev/null

# 安装 Docker Engine
sudo apt-get update
sudo apt-get install docker-ce docker-ce-cli containerd.io docker-compose-plugin

# 验证安装
sudo docker run hello-world

sudo usermod -aG docker $USER
newgrp docker

3.1.2 Docker Compose：编排利器

Docker Compose 不是必须的，但有了它，你就不用一个一个敲 docker run 命令了。尤其是后面我们要启动 Kafka 集群，一个 yaml 文件搞定所有服务。

我用的是 Docker Compose V2，它已经集成到 docker 命令里了：

# 验证是否已安装
docker compose version

# 如果没装，手动安装
sudo apt-get install docker-compose-plugin

来看一个最简单的 docker-compose.yml 示例：

version: '3.8'
services:
  hello:
    image: hello-world
    container_name: hello-demo

运行：docker compose up，看到打印信息就说明环境通了。

3.2 Kafka 单机与集群部署

Kafka 是整个实时流水线的“主动脉”。消息从源头进来，经过 Kafka 缓冲，再被 Flink 消费。我在项目中遇到过 Kafka 集群因为配置不当导致消息丢失的情况，那叫一个惨痛。所以这一节，咱们把部署细节讲透。

3.2.1 单机部署（开发测试用）

单机部署最简单，用 Docker Compose 几行配置就搞定：

version: '3.8'
services:
  zookeeper:
    image: confluentinc/cp-zookeeper:7.4.0
    container_name: zookeeper
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000
    ports:
      - "2181:2181"

  kafka:
    image: confluentinc/cp-kafka:7.4.0
    container_name: kafka
    depends_on:
      - zookeeper
    ports:
      - "9092:9092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1

启动命令：docker compose up -d。等几秒钟，用 docker compose ps 查看状态，两个服务都显示 healthy 就对了。

# 进入 Kafka 容器
docker exec -it kafka bash

# 创建一个测试 topic
kafka-topics --create --topic test-topic --bootstrap-server localhost:9092 --partitions 3 --replication-factor 1

# 列出所有 topic
kafka-topics --list --bootstrap-server localhost:9092

3.2.2 集群部署（生产模拟）

单机版只能用来玩玩，真要模拟生产环境，至少得 3 个 broker。我曾经在生产环境吃过单点故障的亏，一个 broker 挂了，整个集群不可用。所以集群部署是必修课。

来看这个 3 节点集群的配置：

version: '3.8'
services:
  zookeeper:
    image: confluentinc/cp-zookeeper:7.4.0
    container_name: zookeeper
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000
    ports:
      - "2181:2181"

  kafka1:
    image: confluentinc/cp-kafka:7.4.0
    container_name: kafka1
    depends_on:
      - zookeeper
    ports:
      - "9092:9092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 3

  kafka2:
    image: confluentinc/cp-kafka:7.4.0
    container_name: kafka2
    depends_on:
      - zookeeper
    ports:
      - "9093:9092"
    environment:
      KAFKA_BROKER_ID: 2
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9093
      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 3

  kafka3:
    image: confluentinc/cp-kafka:7.4.0
    container_name: kafka3
    depends_on:
      - zookeeper
    ports:
      - "9094:9092"
    environment:
      KAFKA_BROKER_ID: 3
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9094
      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 3

启动后，用 docker compose ps 确认三个 broker 都正常运行。然后创建一个带副本的 topic 测试：

docker exec -it kafka1 bash
kafka-topics --create --topic cluster-topic --bootstrap-server localhost:9092 --partitions 3 --replication-factor 3
kafka-topics --describe --topic cluster-topic --bootstrap-server localhost:9092

你会看到每个 partition 都有 3 个副本，分布在不同的 broker 上。这就是高可用的基础。

3.3 Flink 本地开发环境配置

Flink 是咱们实时流水线的“计算引擎”。环境配置其实不复杂，但有几个细节容易忽略。我刚开始学 Flink 时，光配置 IDEA 的运行环境就折腾了半天。

3.3.1 下载与安装

去 Flink 官网下载最新稳定版（我写这章时是 1.17.1）：

wget https://dlcdn.apache.org/flink/flink-1.17.1/flink-1.17.1-bin-scala_2.12.tgz
tar -xzf flink-1.17.1-bin-scala_2.12.tgz
mv flink-1.17.1 /opt/flink

配置环境变量：

echo 'export FLINK_HOME=/opt/flink' >> ~/.bashrc
echo 'export PATH=$PATH:$FLINK_HOME/bin' >> ~/.bashrc
source ~/.bashrc

验证安装：flink --version，看到版本号就对了。

3.3.2 启动本地集群

Flink 支持单机模式，直接运行：

# 启动
start-cluster.sh

# 查看进程
jps
# 应该看到：StandaloneSessionClusterEntrypoint 和 TaskManagerRunner

打开浏览器访问 http://localhost:8081，你会看到 Flink Web UI。这里可以查看作业状态、任务槽使用情况等。

# 修改 /opt/flink/conf/flink-conf.yaml
taskmanager.numberOfTaskSlots: 4
parallelism.default: 2

3.3.3 IDEA 开发环境配置

咱们用 Maven 项目来开发 Flink 作业。我个人习惯用 IntelliJ IDEA，配置如下：

pom.xml 核心依赖：

<properties>
    <flink.version>1.17.1</flink.version>
    <scala.binary.version>2.12</scala.binary.version>
</properties>

<dependencies>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-streaming-java</artifactId>
        <version>${flink.version}</version>
        <scope>provided</scope>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-clients</artifactId>
        <version>${flink.version}</version>
        <scope>provided</scope>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-connector-kafka</artifactId>
        <version>1.17.1</version>
    </dependency>
</dependencies>

第一个 Flink 作业：

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class WordCount {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.socketTextStream("localhost", 9999)
            .flatMap((String line, Collector<String> out) -> {
                for (String word : line.split(" ")) {
                    out.collect(word);
                }
            })
            .returns(Types.STRING)
            .map(word -> Tuple2.of(word, 1))
            .returns(Types.TUPLE(Types.STRING, Types.INT))
            .keyBy(value -> value.f0)
            .sum(1)
            .print();

        env.execute("Socket WordCount");
    }
}

运行前，先启动一个 socket 数据源：nc -lk 9999，然后在 IDEA 里运行 main 方法。在终端输入单词，IDEA 控制台就会输出统计结果。

3.4 本章知识体系总览

下面这张图把本章的核心组件和它们的关系梳理清楚了。你可以把它当作一个“导航图”，后续章节都会围绕这个架构展开。

3.5 环境验证清单

配置完所有环境后，建议按这个清单逐项检查。我在项目中吃过亏，少验证一项，后面排查问题要多花半天时间。