Apache Flink 单节点伪集群（多TaskManager）指南

目标：在一台 Ubuntu 22.04 单机上，用 1 个 JobManager + 3 个 TaskManager 构建 Flink 1.19.1 的 Standalone 伪集群。伪集群不作真实网络隔离，而是通过「多份 conf + 不同端口 + 多个 JVM 进程」在同一主机内模拟一个 4 节点的分布式集群，用于学习 Flink 的 JobManager / TaskManager 协同、内存模型与资源调度，避免占用多台机器。本文将完整复现 2026-09-29 在单台 Ubuntu 22.04 服务器（主机名 flink-pc）上的从 0 到 1 实际安装与排错过程。

一、环境信息

项	值
操作系统	Ubuntu 22.04（单台真实服务器，主机名 flink-pc）
JDK	OpenJDK 17.0.20.1（`java -version` 正常；完整 JDK 在 `/usr/lib/jvm/java-17-openjdk-amd64`，`JAVA_HOME=/usr/lib/jvm/java-17-openjdk-amd64`）
Flink	1.19.1（Standalone 会话集群）
角色数	1 JobManager + 3 TaskManager（同主机 4 个 JVM 进程）
安装根目录	`/opt/data/flink/`
程序目录	`/opt/data/flink/flink/`
日志目录	`/opt/data/flink/flink/log/`
内存预算	约 2GB（JM 512m + 3×TM 512m）

说明：与「分布式部署」不同——单节点上同时跑多个 Flink 进程，端口必须互不冲突，内存也必须在服务器预算内。这正是伪集群最有技巧的地方（后续「端口/目录规划」与「内存模型」两节重点展开）。

二、伪集群设计

Flink Standalone 集群的关键角色：JobManager（JM） 接收作业、调度 Task、协调 Checkpoint 并负责 Web UI；TaskManager（TM） 向 JM 注册，把资源切成 task slot，真正执行并行的 operator。多 TaskManager 意味着总并行度 = Σ(各 TM 的 slot 数)。

单机伪集群的难点在于：同一主机内 JM 与 3 个 TM、以及 TM 之间是平级进程，天然会争用端口与内存。设计要点：

JM 单份 conf：conf/flink-conf.yaml 承载全部全局配置（REST/RPC 地址、内存模型、心跳超时等）。
TM 用「共享 conf + 动态 -D 参数」区分端口：Flink 的 taskmanager.sh start [-Dkey=value ...] 支持命令行动态覆盖配置项。因此不必复制 3 份 conf，只需在启动每个 TM 时通过 -Dtaskmanager.rpc.port=X -Dtaskmanager.data.port=Y 指定其专属端口即可（任务描述中「复制三份 conf」的思路与这里等价，本方案用 -D 更省）。
JM 端口固定：REST 8081、RPC 6123（jobmanager.rpc.port）。TM 的 rpc port 与 data port 需避开，见下表。
内存按 host 预算压缩：把各进程 process.size 压到 512m，并显式固定各内存子项以满足 Flink 内存模型校验（否则启动即 IllegalConfigurationException，详见「内存模型配置」与「问题排查」）。

2.1 端口 / 目录规划

进程	RPC 端口	REST 端口	Data 端口（Netty，`taskmanager.data.port`）	说明
JobManager	6123	8081	—	`jobmanager.rpc.port`
TaskManager #1	6125	—	9681	`-Dtaskmanager.rpc.port=6125 -Dtaskmanager.data.port=9681`
TaskManager #2	6126	—	9682	同上，端口 +1
TaskManager #3	6127	—	9683	同上，端口 +2

环境变量 FLINK_CONF_DIR=/opt/data/flink/flink/conf：让 JM 与所有 TM 读取共享 conf。
日志：systemd 前台托管时日志进 journald（journalctl -u flink-jm / flink-tm1/2/3）；JM 也会写 log/flink-root-standalonesession-*.log，TM 写 log/flink-root-taskexecutor-*.log/.out。
数据/临时目录：Flink 默认使用 /tmp/jm_*、/tmp/flink-* 等，无需特别配置。

三、基础环境与安装（0 → 1）

Flink 是纯 JVM 程序，前置只需一个完整 JDK。以下是单台 Ubuntu 22.04 上从零开始的完整落地流程（数据根目录 /opt/data/flink，主机名 flink-pc）。

3.1 基础环境

apt-get update && apt-get install -y openjdk-17-jdk wget python3
java -version        # 应显示 OpenJDK 17.x（完整 JDK，自带 java.security 配置）

3.2 下载 Flink 发行版

从 Apache 官方镜像站下载 Flink 1.19.1（Scala 2.12 二进制包）：

mkdir -p /opt/data/flink && cd /opt/data/flink
wget https://dlcdn.apache.org/flink/flink-1.19.1/flink-1.19.1-bin-scala_2.12.tgz

3.3 解压并重命名

tar -xzf flink-1.19.1-bin-scala_2.12.tgz
mv flink-1.19.1 /opt/data/flink/flink

3.4 导出环境变量

export JAVA_HOME=/usr/lib/jvm/java-17-openjdk-amd64
export FLINK_HOME=/opt/data/flink/flink
export PATH=$FLINK_HOME/bin:$JAVA_HOME/bin:$PATH

可将上述 export 追加到 ~/.bashrc，以便登录自动加载。

3.5 写集群主机名（拓扑）

cd /opt/data/flink/flink
echo "flink-pc" > conf/masters    # 单机单角色，只保留本机（避免 start-cluster.sh 误读多节点）
echo "flink-pc" > conf/workers

💡 基础环境已通过 apt 完整安装 openjdk-17-jdk，conf/security/java.security 等配置随 JDK 包自带，理论上不会出现 Error loading java.security；若仍报错，见「问题排查」第 1 条。

四、内存模型配置

Flink 内存模型把单个进程（taskmanager.memory.process.size / jobmanager.memory.process.size）拆成 N 个子项，并有严格校验：process.size = jvm-overhead + jvm-metaspace + (framework.heap + framework.off-heap + task.heap + task.off-heap + managed + network)。所有子项要么显式给出、要么由 fraction 自动推导；本方案为精确控制与可预测性，全部显式固定，并令总和恰好等于 512m（否则多出的「剩余内存」会被算进 derived JVM Overhead，触发越界，见「问题排查」第 2、3 条）。

本集群共用一份 conf/flink-conf.yaml，关键配置如下：

# ---- JobManager ----
jobmanager.rpc.address: flink-pc
jobmanager.rpc.port: 6123
rest.address: flink-pc
rest.port: 8081
jobmanager.memory.process.size: 512m
jobmanager.memory.jvm-metaspace.size: 96m
jobmanager.memory.jvm-overhead.min: 64m
jobmanager.memory.jvm-overhead.max: 64m
# 不要显式设置 jobmanager.memory.heap.size（legacy），否则剩余内存全算 overhead 而越界

# ---- TaskManager（共享模板，3 个 TM 用 -D 覆盖端口）----
taskmanager.memory.process.size: 512m
taskmanager.memory.jvm-metaspace.size: 96m
taskmanager.memory.jvm-overhead.min: 64m
taskmanager.memory.jvm-overhead.max: 64m
taskmanager.memory.framework.heap.size: 32m
taskmanager.memory.framework.off-heap.size: 64m   # 必须 ≥ network 所需直接内存，否则 Netty DirectBufferMemory OOM
taskmanager.memory.task.off-heap.size: 0m
taskmanager.memory.task.heap.size: 192m
taskmanager.memory.managed.size: 32m
taskmanager.memory.network.min: 32m
taskmanager.memory.network.max: 32m

# ---- 资源 ----
taskmanager.numberOfTaskSlots: 2
parallelism.default: 2

# ---- 容错 / heartbeat ----
taskmanager.registration.timeout: 120000
heartbeat.timeout: 60000

# ---- JDK 17 模块强封装：Flink 内部反射需要 opens ----
env.java.opts: --add-opens=java.base/java.lang=ALL-UNNAMED --add-opens=java.base/java.util=ALL-UNNAMED --add-opens=java.base/java.io=ALL-UNNAMED --add-opens=java.base/java.util.concurrent=ALL-UNNAMED --add-opens=java.base/java.util.concurrent.atomic=ALL-UNNAMED --add-opens=java.base/java.math=ALL-UNNAMED

内存分配核算：

JobManager：96（metaspace）+ 64（overhead）= 160m，其余 512−160=352m 由 Flink 自动分给 heap/off-heap。关键：不要写死 jobmanager.memory.heap.size（legacy 选项），否则 Flink 走 interpretLegacyHeap 分支把剩余全部算成 overhead。
TaskManager（每个）：96（meta）+ 64（overhead）+ 32（fw.heap）+ 64（fw.off-heap）+ 192（task.heap）+ 32（managed）+ 32（network）= 512m，恰好填满。

# 用 Flink 自带的 JVM 参数查询脚本可快速核对是否合法：
/bin/bash /opt/data/flink/flink/bin/bash-java-utils.sh queryJmResourceParams queryTmMemoryParams

五、命令与 systemd

伪集群生产化采用 systemd 前台托管（Type=simple），日志直接进 journald，配置如下 4 个单元。

5.1 JobManager：`/etc/systemd/system/flink-jm.service`

[Unit]
Description=Apache Flink 1.19.1 JobManager (flink-pc pseudo cluster)
After=network.target
Wants=network.target
[Service]
Type=simple
User=root
WorkingDirectory=/opt/data/flink/flink
Environment=FLINK_CONF_DIR=/opt/data/flink/flink/conf
Environment=JAVA_HOME=/usr/lib/jvm/java-17-openjdk-amd64
ExecStart=/opt/data/flink/flink/bin/jobmanager.sh start-foreground
Restart=on-failure
RestartSec=5
[Install]
WantedBy=multi-user.target

5.2 TaskManager：`/etc/systemd/system/flink-tm1.service`（tm2/tm3 仅换 `-D` 端口 6126/6127、9682/9683）

[Unit]
Description=Apache Flink 1.19.1 TaskManager #1 (flink-pc pseudo cluster)
After=network.target flink-jm.service
[Service]
Type=simple
User=root
WorkingDirectory=/opt/data/flink/flink
Environment=FLINK_CONF_DIR=/opt/data/flink/flink/conf
Environment=JAVA_HOME=/usr/lib/jvm/java-17-openjdk-amd64
ExecStart=/opt/data/flink/flink/bin/taskmanager.sh start-foreground -Dtaskmanager.rpc.port=6125 -Dtaskmanager.data.port=9681
Restart=on-failure
RestartSec=5
[Install]
WantedBy=multi-user.target

5.3 启用与启动

本集群的 start/stop 均逐实例单独执行，不写批量脚本循环——每条命令只操作一个 unit，便于看清每个进程的启动顺序，也便于单独排错。

systemctl daemon-reload

# ---- 1. 逐个 enable（开机自启），每个 unit 一行 ----
systemctl enable flink-jm     # 启用 JobManager
systemctl enable flink-tm1    # 启用 TaskManager #1
systemctl enable flink-tm2    # 启用 TaskManager #2
systemctl enable flink-tm3    # 启用 TaskManager #3

# ---- 2. 逐个 start（先 JM，再逐个 TM），每个 unit 一行 ----
systemctl start flink-jm      # 先启动 JobManager
systemctl start flink-tm1     # 再启动 TaskManager #1
systemctl start flink-tm2     # 再启动 TaskManager #2
systemctl start flink-tm3     # 再启动 TaskManager #3

# ---- 3. 逐个 is-active 确认，应全部输出 active ----
systemctl is-active flink-jm
systemctl is-active flink-tm1
systemctl is-active flink-tm2
systemctl is-active flink-tm3

可选：手动命令演示 flink-daemon.sh 机制（与 systemd 等价，但直接调用 Flink 脚本并 fork 到后台）。同样逐一实例、一条命令只起一个进程：

# JobManager：一条命令、一个进程
FLINK_CONF_DIR=/opt/data/flink/flink/conf \
  /opt/data/flink/flink/bin/jobmanager.sh start-foreground &

# TaskManager #1：一条命令，端口 rpc 6125 / data 9681
FLINK_CONF_DIR=/opt/data/flink/flink/conf \
  /opt/data/flink/flink/bin/taskmanager.sh start-foreground -Dtaskmanager.rpc.port=6125 -Dtaskmanager.data.port=9681 &

# TaskManager #2：一条命令，端口 rpc 6126 / data 9682
FLINK_CONF_DIR=/opt/data/flink/flink/conf \
  /opt/data/flink/flink/bin/taskmanager.sh start-foreground -Dtaskmanager.rpc.port=6126 -Dtaskmanager.data.port=9682 &

# TaskManager #3：一条命令，端口 rpc 6127 / data 9683
FLINK_CONF_DIR=/opt/data/flink/flink/conf \
  /opt/data/flink/flink/bin/taskmanager.sh start-foreground -Dtaskmanager.rpc.port=6127 -Dtaskmanager.data.port=9683 &

停止时同理一对一停下（先停 TM，再停 JM）：

# 逐个停止 TaskManager，再停 JobManager
systemctl stop flink-tm3     # 停 TaskManager #3
systemctl stop flink-tm2     # 停 TaskManager #2
systemctl stop flink-tm1     # 停 TaskManager #1
systemctl stop flink-jm      # 最后停 JobManager

💡 为什么不直接 bin/start-cluster.sh？它按 conf/workers 里的主机名逐台 ssh 拉起集群，只适合真多机或假分发；单机多进程用 jobmanager.sh / taskmanager.sh 逐个拉起并各自指定端口更可控。

六、验证回显

6.1 4 个 JVM 进程

ps -eo pid,rss,args | grep -E "TaskManagerRunner|StandaloneSessionClusterEntrypoint" | grep -v grep
# 约 4 行：1 行 StandaloneSessionClusterEntrypoint（JM）+ 3 行 TaskManagerRunner（TM）
# 各 TM Xmx≈301MB（task heap 192+framework heap 32 等），符合 512m process.size

6.2 3 个 TM 注册到 JM（Web UI 显示 Slots ≥3）

curl -s http://flink-pc:8081/taskmanagers
# {"taskmanagers":[{"id":"10.0.0.8:6125-...","dataPort":9681,"slotsNumber":2,...
#                     ,{"dataPort":9682,...},{"dataPort":9683,...}]}   # 3 个，各 2 slots

整个 /taskmanagers 里每个 TM 都带 "totalProcessMemory":536870912（=512MB），可见内存模型生效。

6.3 提交 WordCount 作业（`-p 6` 占满 3 TM×2 slot）

cd /opt/data/flink/flink
./bin/flink run -m flink-pc:8081 -d -p 6 examples/streaming/WordCount.jar
# Job has been submitted with JobID 31421550b1afffde26ec31db7df75f3d

curl -s http://flink-pc:8081/jobs/overview
# {"jobs":[{"name":"WordCount","state":"FINISHED","tasks":{"total":13,"finished":13,"failed":0}}]}

6.4 常驻无界作业（演示 EXECUTING/RUNNING 与跨 TM 调度）

用 python3 起一个 9999 端口 TCP 单词流（本机未装 nc，用 python3 更通用），再提交 SocketWindowWordCount：

nohup python3 -c '
import socket,time
s=socket.socket(); s.setsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR,1)
s.bind(("0.0.0.0",9999)); s.listen(1); c,a=s.accept()
ws=["hello","world","flink","pseudo","cluster"]; i=0
while True:
    c.sendall((ws[i%len(ws)]+"\n").encode()); i+=1; time.sleep(0.1)' >/tmp/sockfeed.log 2>&1 &
# 说明：上面这个 while True 是"持续发送单词流"的演示程序（数据源程序逻辑），并非集群启停脚本，
#       它只负责往 9999 端口喂数据供 SocketWindowWordCount 消费，与各实例的启停无关，故保留。

./bin/flink run -m flink-pc:8081 -d -p 6 \
    examples/streaming/SocketWindowWordCount.jar --hostname flink-pc --port 9999
# Job has been submitted with JobID e38199e0780a645814b51c2e274386f6

./bin/flink list -m flink-pc:8081
# ------------------ Running/Restarting Jobs -------------------
# 29.09.2026 14:23:23 : e38199e0780a645814b51c2e274386f6 : Socket Window WordCount (RUNNING)

curl -s http://flink-pc:8081/taskmanagers \
  | python3 -c 'import sys,json;[print("dataPort",t["dataPort"],"freeSlots",t["freeSlots"]) for t in json.load(sys.stdin)["taskmanagers"]]'
# 说明：上面只是"解析 REST 接口 /taskmanagers 返回、打印每个 TM 端口"的单行显示命令，
#       遍历的对象是返回的 JSON 数据而非执行实例命令，不涉及任何实例启停循环，故保留。
# dataPort 9683 freeSlots 0      <- 3 个 TM 的 slot 全被占满，job 并行度 6 已分发到 3 个 TM
# dataPort 9682 freeSlots 0
# dataPort 9681 freeSlots 0

./bin/flink cancel -m flink-pc:8081 e38199e0780a645814b51c2e274386f6   # 验证后取消

输出：WordCount FINISHED（13 个 task 全部 finished、failed=0）、Socket Window WordCount RUNNING 且 3 个 TM 各 2 slots 被占满 → 并发度 6 已被拆分到 3 个 TaskManager 上执行，伪集群分布式语义成立。

七、内存调优要点

优先用 taskmanager.memory.process.size + 显式子项：单机内存吃紧时，把 512m 里的 kernel（metaspace/overhead/off-heap）压到最小，把预算留给 task.heap（真正跑算子的堆）。本方案 task.heap=192m、slots=2，单个 slot 约 96m 堆，可跑 WordCount/Socket 这类轻量 job。
framework.off-heap.size 不能设为 0：Netty 的 network buffer 走直接内存（direct buffer），预算来源是 framework.off-heap + task.off-heap（对应 JVM -XX:MaxDirectMemorySize）。off-heap=0 时 NetworkBufferPool 需 32m 直接内存却申请不到 → OutOfMemoryError: Direct buffer memory。
totalProcessMemory 校验：可在 Web UI 或 /taskmanagers JSON 的 memoryConfiguration 字段核对各子项是否符合预期。
服务器内存紧张时，可进一步把 JM 或 TM 的 process.size 收到 384m，并将 network 缩到 network.min/max=16m、metaspace 缩到 64m，腾出预算给 task.heap。

八、问题排查

8.1 `Error loading java.security file`（`InternalError`）

现象：任何 ./bin/flink ...、jobmanager.sh、taskmanager.sh 启动即抛 java.lang.InternalError: Error loading java.security file，堆栈停在 BashJavaUtils 计算内存参数处。
起因：此报错通常是 JDK 未完整安装所致——/usr/lib/jvm/java-17-openjdk-amd64/conf/security/java.security（软链）指向 /etc/java-17-openjdk/security/java.security，若该目录缺失或损坏则 java.home 下找不到实时配置文件。基础环境用 apt 完整安装 openjdk-17-jdk 后本应自带配置，该问题一般来自对 JDK 的手工复制/裁剪。
解决：先确认 readlink -f /usr/lib/jvm/java-17-openjdk-amd64/conf/security/java.security 能解析到真实文件；若解析失败或 /etc/java-17-openjdk/security/java.security 缺失，用 apt --reinstall openjdk-17-jre（或 openjdk-17-jdk）修复，必要时同步确认 /etc/java-17-openjdk 与 /etc/ssl/certs/java 目录齐全。

8.2 `IllegalConfigurationException: Derived JVM Overhead ... is not in configured range`

现象：JM 启动报 Derived JVM Overhead size (256mb) is not in configured range [64,128]。
起因：显式设置了 jobmanager.memory.heap.size（legacy 选项），Flink 走 interpretLegacyHeap 把「剩余内存」全算作 JVM Overhead（512−128 heap−32 offheap−96 metaspace=256），越过了上限。
解决：删掉 jobmanager.memory.heap.size（和 off-heap.size）显式项，只固定 metaspace 与 overhead，让 heap 由剩余量自动推导到合法区间。

8.3 `Derived JVM Overhead size (96mb) not in range [64,64]`（TM）

现象：TM 启动时 overhead 被推导为 96mb，超出我设的 [64,64]。
起因：显式各组件总和+metaspace < process−overhead 时，「剩余内存」会继续计入 JVM Overhead。
解决：令各显式组件之和精确等于 512−96−64=352m（即 task.heap + framework.heap/off-heap + managed + network = 352m），使总分配恰好填满 process.size，overhead 恰好落在固定值 64m。

8.4 `InaccessibleObjectException: ... does not "opens java.lang"`

现象：TM 日志出现 Unable to make field private java.lang.String Throwable.detailMessage accessible: module java.base does not "opens java.lang" to unnamed module，随后 Terminating TaskManagerRunner with exit code 1。
起因：JDK 17+ 模块强封装，Flink 内部反射（如 ExceptionUtils 富化 OOM 信息）需要目标包对未命名模块开放。
解决：在 conf/flink-conf.yaml 增加 env.java.opts 携带若干 --add-opens=java.base/java.lang=ALL-UNNAMED 等参数（见上文配置模板）。

8.5 TM `OutOfMemoryError: Direct buffer memory`（NetworkBufferPool）

现象：TM 启动时 Cannot allocate direct memory segment ... required 32MB, allocated 31MB, missing 1MB。
起因：framework.off-heap.size 与 task.off-heap.size 都被设成 0，MaxDirectMemorySize 侧无预算，Netty 的 network buffer 申请不到直接内存。
解决：给 framework.off-heap.size=64m（并同步下调 task.heap 保持总和 352m），足以覆盖 32m network + Netty 开销。

8.6 服务反复 `activating` / 状态一直转

systemd 单元 Restart=on-failure 会在启动失败后按 RestartSec 反复重试，期间显示 activating。先用 journalctl -u <单元> --no-pager -n 60 定位内存/反射类报错，修完配置后 systemctl restart 即可。排查顺序建议：内存模型 → JDK add-opens → off-heap 预算 → 端口冲突。

Apache Flink 单节点伪集群（多TaskManager）指南

Apache Flink 单节点伪集群（多TaskManager）指南

一、环境信息

二、伪集群设计

2.1 端口 / 目录规划

三、基础环境与安装（0 → 1）

3.1 基础环境

3.2 下载 Flink 发行版

3.3 解压并重命名

3.4 导出环境变量

3.5 写集群主机名（拓扑）

四、内存模型配置

五、命令与 systemd

5.1 JobManager：/etc/systemd/system/flink-jm.service

5.2 TaskManager：/etc/systemd/system/flink-tm1.service（tm2/tm3 仅换 -D 端口 6126/6127、9682/9683）

5.3 启用与启动

六、验证回显

6.1 4 个 JVM 进程

6.2 3 个 TM 注册到 JM（Web UI 显示 Slots ≥3）

6.3 提交 WordCount 作业（-p 6 占满 3 TM×2 slot）

6.4 常驻无界作业（演示 EXECUTING/RUNNING 与跨 TM 调度）

七、内存调优要点

八、问题排查

8.1 Error loading java.security file（InternalError）

8.2 IllegalConfigurationException: Derived JVM Overhead ... is not in configured range

8.3 Derived JVM Overhead size (96mb) not in range [64,64]（TM）

8.4 InaccessibleObjectException: ... does not "opens java.lang"

8.5 TM OutOfMemoryError: Direct buffer memory（NetworkBufferPool）

8.6 服务反复 activating / 状态一直转

5.1 JobManager：`/etc/systemd/system/flink-jm.service`

5.2 TaskManager：`/etc/systemd/system/flink-tm1.service`（tm2/tm3 仅换 `-D` 端口 6126/6127、9682/9683）

6.3 提交 WordCount 作业（`-p 6` 占满 3 TM×2 slot）

8.1 `Error loading java.security file`（`InternalError`）

8.2 `IllegalConfigurationException: Derived JVM Overhead ... is not in configured range`

8.3 `Derived JVM Overhead size (96mb) not in range [64,64]`（TM）

8.4 `InaccessibleObjectException: ... does not "opens java.lang"`

8.5 TM `OutOfMemoryError: Direct buffer memory`（NetworkBufferPool）

8.6 服务反复 `activating` / 状态一直转