19.4.5 执行引擎与并发控制

执行引擎是自动化运维平台的核心运行时，负责将编排后的任务转化为可执行动作，并在资源受限、依赖复杂的环境中保证稳定性与可控性。目标是高吞吐、可观测、可回滚与可审计的执行能力。

原理草图（执行引擎与并发控制）：

执行架构常见组件：
- 任务分发：将编排图拆分为可执行单元，按依赖与优先级分发。
- 执行代理：在目标主机执行脚本/命令/API调用，支持幂等与重试。
- 状态汇聚：统一收集状态、日志与指标，提供实时可视化与追踪。
- 结果回传：将结果回写平台，触发后续节点或补偿逻辑。

并发控制策略：
- 全局并发上限：限制同时执行数量，防止资源耗尽。
- 目标维度并发：按主机/集群/应用限制并发。
- 分批与窗口：滚动执行，支持批大小与间隔。
- 优先级与队列：高优任务抢占资源。
- 资源配额：按租户/项目配额控制。

故障隔离与恢复机制：
- 超时与熔断：对长时间无响应任务超时处理。
- 重试与幂等：失败任务在可控次数内重试。
- 断点续跑：从失败节点继续执行。
- 状态机驱动：统一状态机管理任务生命周期。

可观测性指标建议：
- 任务耗时、失败率、重试次数、并发峰值。
- 关键阶段耗时拆分（分发、执行、回传）。
- 执行日志分级与脱敏。

示例：本地搭建轻量执行引擎（Celery + Redis）#

安装（Ubuntu）

sudo apt update
sudo apt install -y python3 python3-venv redis-server
python3 -m venv /opt/exec-engine-venv
source /opt/exec-engine-venv/bin/activate
pip install -U pip celery redis

配置：执行引擎与并发控制
- 文件：/opt/exec-engine/app.py

from celery import Celery

app = Celery(
    "exec_engine",
    broker="redis://127.0.0.1:6379/0",
    backend="redis://127.0.0.1:6379/1",
)

# 幂等示例：重复执行输出相同结果
@app.task(bind=True, max_retries=2, default_retry_delay=5)
def run_cmd(self, cmd):
    import subprocess, time
    start = time.time()
    p = subprocess.run(cmd, shell=True, capture_output=True, text=True)
    if p.returncode != 0:
        raise self.retry(exc=Exception(p.stderr[:200]))
    return {"cmd": cmd, "rc": p.returncode, "out": p.stdout[:200], "cost": time.time()-start}

启动执行引擎（并发=4，队列=ops）

source /opt/exec-engine-venv/bin/activate
celery -A app worker -Q ops -c 4 --loglevel=INFO

触发任务（模拟任务分发器）

source /opt/exec-engine-venv/bin/activate
python - <<'PY'
from app import run_cmd
tasks = [
    run_cmd.delay("hostname"),
    run_cmd.delay("uptime"),
    run_cmd.delay("echo OK"),
]
for t in tasks:
    print(t.id)
PY

查询结果

source /opt/exec-engine-venv/bin/activate
python - <<'PY'
from app import run_cmd
# 替换为上一步输出的task_id
r = run_cmd.AsyncResult("TASK_ID")
print(r.state, r.result)
PY

预期效果
- 控制台显示 3 个任务被并发执行，且失败任务会自动重试。
- r.result 输出执行结果与耗时。

示例：并发与分批控制（滚动窗口）#

分批脚本（示例：每批 2 台，间隔 5 秒）

#!/usr/bin/env bash
# 文件：/opt/exec-engine/rolling.sh
set -e
hosts=(10.0.0.11 10.0.0.12 10.0.0.13 10.0.0.14)
batch=2
interval=5

for ((i=0; i<${#hosts[@]}; i+=batch)); do
  slice=("${hosts[@]:i:batch}")
  echo "Batch: ${slice[*]}"
  for h in "${slice[@]}"; do
    ssh "$h" "systemctl status nginx >/dev/null" &
  done
  wait
  sleep "$interval"
done
echo "Done"

关键点说明
- batch 控制并发窗口大小；interval 控制批次间隔。
- wait 确保本批完成后进入下一批。

示例：执行状态汇聚（日志与指标）#

日志收集示例（本地汇聚）

# 文件：/opt/exec-engine/logs/exec.log
tail -n 50 /opt/exec-engine/logs/exec.log

Prometheus 指标示例（文本格式）

exec_task_total{status="success"} 120
exec_task_total{status="failed"} 3
exec_task_duration_seconds_bucket{le="1"} 20
exec_task_duration_seconds_bucket{le="3"} 80

排错指南#

任务一直 PENDING
- 检查 Redis：redis-cli ping 应返回 PONG
- 检查 Worker：ps -ef | grep celery
- 检查队列：任务发送到 ops 队列但 worker 没监听会卡住
任务执行失败且未重试
- 检查任务定义中的 max_retries 和 default_retry_delay
- 查看 worker 日志：/var/log/syslog 或 worker 启动日志
并发太高导致系统抖动
- 降低 worker 并发：celery -c 2
- 增加批次间隔或开启目标维度并发限制

练习#

将并发从 4 改为 2，观察任务完成时间变化。
模拟失败任务：run_cmd.delay("false")，确认重试机制生效。
将滚动脚本的 batch=1 与 batch=3 对比执行耗时与压力。
设计一个“超时+熔断”策略：超过 60 秒失败的任务直接终止并报警。

最佳实践建议：
- 关键操作先演练再执行，敏感任务采用低并发滚动。
- 为高风险动作配置审批与双人复核。
- 结合业务峰谷设置并发策略，避免高峰期放大影响。
- 为不同任务类型设置差异化超时与重试策略。

通过稳健的执行引擎与精细化并发控制，可显著提升自动化效率、降低变更风险，并为大规模运维场景提供可持续支撑。

19.4.5 执行引擎与并发控制

示例：本地搭建轻量执行引擎（Celery + Redis）#

示例：并发与分批控制（滚动窗口）#

示例：执行状态汇聚（日志与指标）#

排错指南#

练习#

本页内容