shell

文件监控与事件驱动流水线实战

讲解 inotifywait 与 fswatch 的事件模型、事件去抖与合并、触发式流水线的可靠性设计、watcher 数量与容器内限制，以及替代轮询的适用边界。

Leeting Yan 7 分钟阅读 3061 字

1. 事件驱动的动机

一句话总结： 轮询的问题不是慢，而是「延迟与开销不可兼得」；inotify 让内核在文件变化时主动通知，把延迟压到毫秒级且几乎零开销。

每分钟跑一次 find 找新文件，意味着最坏情况延迟一分钟，而绝大多数轮询都是空转。inotify 是 Linux 内核提供的文件系统事件接口，inotifywait 是它的命令行封装。

# 轮询：简单但有延迟与空转开销
while :; do
  new=$(find /incoming -type f -newer /tmp/last_run 2>/dev/null)
  [[ -n "$new" ]] && process "$new"
  touch /tmp/last_run
  sleep 60
done

# 事件驱动：内核主动通知
inotifywait -m -e close_write --format '%w%f' /incoming

1.1 事件类型速查

一句话总结： 写文件通常伴随多个事件（create、modify、close_write），监控「写完成」应选 close_write 而非 modify。

# 常用事件
# create      文件或目录被创建
# modify      内容被修改
# close_write 以可写方式打开的文件被关闭（写入完成）
# moved_to    文件被移入监控目录
# delete      文件被删除
# attrib      属性变化（权限、时间戳）

inotifywait -m -e create -e moved_to -e close_write --format '%e %w%f' /incoming

1.2 为什么 close_write 比 modify 可靠

一句话总结： 一次写入会触发多次 modify，用 modify 会在文件还没写完时就处理；close_write 保证写者已关闭描述符。

# 危险：modify 可能在文件只写了一半时触发
inotifywait -m -e modify --format '%w%f' /incoming

# 安全：close_write 表示写入方已完成写入并关闭
inotifywait -m -e close_write --format '%w%f' /incoming

2. inotifywait 的用法

一句话总结： -m 持续监控、-r 递归、--format 定制输出、-q 静默，组合起来才能嵌进流水线。

# 持续监控，只输出事件名与路径
inotifywait -m -q -r -e close_write,moved_to --format '%e|%w%f' /data

# 只等一次事件（不带 -m），适合脚本里的一次性等待
inotifywait -q -e close_write /incoming/file.csv

# 输出到 while 循环，用 | 分隔事件与路径
inotifywait -m -q -r -e close_write --format '%e|%w%f' /data \
  | while IFS='|' read -r ev path; do
      echo "事件 $ev 路径 $path"
    done

2.1 处理管道中的缓冲问题

一句话总结： inotifywait 输出进管道时会被块缓冲，while read 循环可能长时间收不到数据；用 stdbuf 或 --format 配合 -q 缓解。

# 用 stdbuf 关闭输出缓冲，保证事件即时到达循环
stdbuf -oL inotifywait -m -q -e close_write --format '%w%f' /data \
  | while IFS= read -r f; do handle "$f"; done

2.2 fswatch 的跨平台替代

一句话总结： macOS 没有 inotify，fswatch 用 FSEvents/kqueue 提供等价能力，-0 输出 NUL 分隔、-1 只等一次。

# macOS / BSD 上的等价写法
fswatch -0 -r /data | while IFS= read -r -d '' f; do handle "$f"; done

# 只等一次
fswatch -1 /incoming/file.csv

# 指定事件类型（部分平台支持）
fswatch -e '.*\.tmp$' -r /data

3. 事件去抖与合并

一句话总结： 一次保存操作会触发多个事件，一次批量拷贝会触发成百上千个事件，直接逐个处理会重复劳动甚至压垮下游。

去抖（debounce）是把短时间内的多个事件合并成一次处理；合并（coalesce）是把同一文件的重复事件去重。两者都是事件驱动流水线的必需品。

# 朴素做法：每个事件都处理，重复且低效
inotifywait -m -q -e close_write --format '%w%f' /data \
  | while read -r f; do heavy_process "$f"; done

3.1 用 sleep 窗口做去抖

一句话总结： 收到事件后不立即处理，而是等待一个静默窗口，窗口内再有事件就重置计时，窗口结束才真正执行。

# 去抖：收到事件后等待 2 秒，期间有新事件则重置
pending=""
timer_pid=""
flush() { [[ -n "$pending" ]] && { echo "处理: $pending"; pending=""; }; }

inotifywait -m -q -e close_write --format '%w%f' /data \
  | while IFS= read -r f; do
      pending="$f"
      kill "$timer_pid" 2>/dev/null
      ( sleep 2; flush ) & timer_pid=$!
    done

3.2 批量事件合并

一句话总结： 用「收集 + 定时清空」的模式，把窗口内所有变化路径汇总去重后一次性处理，天然适配增量构建。

# 收集窗口内所有事件，去重后一次处理
declare -A changed
flush_batch() {
  [[ ${#changed[@]} -eq 0 ]] && return
  printf '%s\n' "${!changed[@]}" | sort -u > /tmp/changed.txt
  echo "本轮变化 $(wc -l < /tmp/changed.txt) 个文件"
  changed=()
}

4. 触发式流水线的可靠性

一句话总结： 事件是「尽力而为」的通知，可能丢失、可能重复、可能乱序，流水线必须容忍这三种情况。

inotify 的队列有长度上限，溢出时会产生 IN_Q_OVERFLOW 事件而不是逐个通知。这意味着「只依赖事件」的流水线会漏文件，必须定期做一次全量扫描兜底。

# 检测溢出事件
inotifywait -m -q -e close_write -e overflow --format '%e|%w%f' /data \
  | while IFS='|' read -r ev path; do
      if [[ "$ev" == *OVERFLOW* ]]; then
        echo '事件队列溢出，触发全量扫描' >&2
        full_scan
      fi
    done

4.1 幂等处理

一句话总结： 处理函数必须对「同一文件被处理多次」无副作用，用状态文件或输出存在性判断来保证。

handle() {
  local f="$1" out="/processed/$(basename "$f").done"
  [[ -f "$out" ]] && { echo "已处理，跳过: $f"; return 0; }
  process_file "$f" && touch "$out"
}

4.2 处理期间的新事件

一句话总结： 处理耗时较长时，期间到达的事件不能丢；用「待处理队列 + 串行消费」把处理与接收解耦。

# 接收与处理分离：接收写入队列，消费者串行处理
inotifywait -m -q -e close_write --format '%w%f' /data >> /tmp/queue.txt &

while :; do
  if [[ -s /tmp/queue.txt ]]; then
    head -1 /tmp/queue.txt > /tmp/current.txt
    sed -i '1d' /tmp/queue.txt
    handle "$(cat /tmp/current.txt)"
  else
    sleep 1
  fi
done

5. 限制与资源

一句话总结： 每个被监控的目录占用一个 inotify watch，递归监控大目录树会耗尽 max_user_watches，必须显式评估与调高。

# 查看当前限制与使用量
cat /proc/sys/fs/inotify/max_user_watches
cat /proc/sys/fs/inotify/max_user_instances
cat /proc/sys/fs/inotify/max_queued_events

# 临时调高（需要 root）
sysctl fs.inotify.max_user_watches=524288

5.1 容器内的额外限制

一句话总结： 容器共享宿主内核参数，且容器内的 watch 计数独立；同时被监控的目录若是 bind mount，事件可能不传播。

# 容器内查看可用 watch 数量
find /proc/*/fd -lname 'anon_inode:inotify' 2>/dev/null | wc -l

# Docker 中调高限制（需要 --privileged 或对应 sysctl）
docker run --sysctl fs.inotify.max_user_watches=524288 myimage

# 注意：overlayfs 上监控上层目录可能收不到下层文件的变更

5.2 网络与虚拟文件系统的坑

一句话总结： NFS、SMB 等网络文件系统不产生本地 inotify 事件，必须回到轮询方案。

# 判断文件系统类型
df -T /data | awk 'NR==2 {print $2}'

# 网络文件系统上退化为轮询
if [[ "$(df -T /data | awk 'NR==2 {print $2}')" =~ ^(nfs|smb|cifs) ]]; then
  echo '网络文件系统，使用轮询模式'
  poll_mode
fi

6. 轮询与事件的混合策略

一句话总结： 事件负责低延迟触发，轮询负责兜底完整性与处理遗漏，两者结合才是生产级方案。

#!/usr/bin/env bash
set -euo pipefail

# 事件通道：低延迟
inotifywait -m -q -e close_write,moved_to --format '%w%f' /incoming \
  >> /var/queue.txt &

# 兜底通道：每 5 分钟全量扫描一次，补齐可能遗漏的文件
while :; do
  sleep 300
  find /incoming -type f -name '*.csv' -newer /var/last_scan -print \
    >> /var/queue.txt
  touch /var/last_scan
done &

6.1 消费队列

一句话总结： 两条通道写入同一队列，消费端去重后串行处理，既快又不会漏。

# 消费端：去重后逐个处理
while :; do
  if [[ -s /var/queue.txt ]]; then
    sort -u /var/queue.txt > /var/queue.dedup
    : > /var/queue.txt
    while IFS= read -r f; do
      [[ -f "$f" ]] && handle "$f"
    done < /var/queue.dedup
  fi
  sleep 2
done

6.2 监控 watcher 自身

一句话总结： watcher 进程本身会挂，用 systemd 的自动重启或 supervisor 保证它一直活着。

# systemd 单元片段：崩溃自动重启
# [Service]
# ExecStart=/usr/local/bin/watcher.sh
# Restart=always
# RestartSec=5

7. 实战：文件落地触发流水线

一句话总结： 把事件监控、去抖、幂等处理、失败重试与队列兜底组合成一条完整流水线，用于「文件落地即处理」的场景。

#!/usr/bin/env bash
set -euo pipefail

WATCH_DIR="${1:-/incoming}"
DONE_DIR=/processed
QUEUE=/var/spool/watch.queue
mkdir -p "$DONE_DIR"

# 处理函数：幂等 + 重试
handle() {
  local f="$1" base out
  base=$(basename "$f"); out="$DONE_DIR/$base.done"
  [[ -f "$out" ]] && return 0
  local i=0
  while (( i < 3 )); do
    if process_file "$f"; then touch "$out"; return 0; fi
    sleep $(( 2 ** i )); i=$(( i + 1 ))
  done
  echo "处理失败: $f" >&2
  return 1
}

7.1 事件接入与去抖

一句话总结： 用 NUL 分隔接收事件，进入去抖窗口，窗口结束后统一处理。

# 事件接入：写入队列（NUL 分隔）
inotifywait -m -q -e close_write -e moved_to --format '%w%f\0' "$WATCH_DIR" \
  >> "$QUEUE" &

# 去抖消费：每 3 秒检查一次队列，去重处理
while :; do
  sleep 3
  [[ -s "$QUEUE" ]] || continue
  tr '\0' '\n' < "$QUEUE" | sort -u > /tmp/batch.txt
  : > "$QUEUE"
  while IFS= read -r f; do
    [[ -f "$f" ]] && handle "$f" || true
  done < /tmp/batch.txt
done

7.2 启动自检与全量补偿

一句话总结： 流水线启动时先做一次全量扫描，补齐上次运行期间可能丢失的文件，然后才进入事件循环。

# 启动自检：处理所有未完成的文件
find "$WATCH_DIR" -type f ! -name '.*' | while IFS= read -r f; do
  handle "$f" || true
done

# 之后进入事件循环
echo '初始扫描完成，进入事件监听'

8. 总结

环节要点
动机轮询延迟与开销不可兼得，事件驱动压到毫秒级
事件选择用 close_write 而非 modify，避免处理半成品
工具Linux 用 inotifywait，macOS 用 fswatch
去抖静默窗口重置计时，批量事件合并去重
可靠性事件可能丢失重复乱序，处理必须幂等
溢出队列溢出只报 overflow，需全量扫描兜底
资源每目录一个 watch，注意 max_user_watches
文件系统网络文件系统不产生事件，退回轮询

事件驱动的核心矛盾是「通知是尽力而为的，而业务要求不丢不重」。解法不是迷信事件，而是让事件负责快、让轮询负责全、让处理逻辑负责幂等，三者叠加才构成可靠的触发式流水线。当流水线里的步骤越来越多、依赖关系越来越复杂时，就需要一个任务运行器来编排，这正是 Makefile 要解决的问题。

延伸阅读

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章 返回首页

「shell」更多文章

  1. 任务编排与 Makefile 实战
  2. 结构化数据清洗与报表生成实战
  3. 并发控制与文件锁实战