上一篇讲清了 DLQ 如何把逻辑上永久无法处理的 event 隔离到独立存储、等待补救管道重处理。这一篇进入另一个维度的隔离：一个 Logstash 进程里运行多条管道，管道之间互不干扰，又能通过 virtual address 串联成拓扑。

核心问题是三个：多管道的资源和生命周期如何隔离，pipeline-to-pipeline 通信怎么实现，以及什么时候该用多管道而不是在单管道里写 conditional。

单管道的局限

单管道（一个 logstash.conf）在简单场景下够用，但遇到两类需求时会暴露问题：

1
2
3
4
5
6
7
8
9
10
11
12
问题一：不同数据流的处理要求差异大
  - syslog 流：高吞吐、无状态 filter、8 个 worker
  - 慢查询日志：低频、复杂 Grok、2 个 worker
  - 同一管道里 worker 数只能统一设，无法按流分配资源

问题二：任一 output 挂掉，整条管道停摆
  - Logstash 默认在任一 output 不可用时阻塞整条管道
    （这是 at-least-once 投递保证的代价，不是缺陷）
  - 同时写 ES 和某个 HTTP 端点时，HTTP 端点抖动会连带停掉 ES 的写入
  - filter 与 output 由同一组 worker 线程执行，output 阻塞会把 filter 一起卡死
  - 单管道共享一个 queue，各数据流的 event 混在一起积压，无法区分优先级
  - 慢 filter（如重 geoip lookup）占满 worker slot 也是同一条链上的表现

Multiple Pipelines 的根本出发点是：不同数据流应该有各自独立的 queue、worker 池和生命周期，互不影响。第二类问题在官方文档里的措辞是 “Logstash, by default, is blocked when any single output is down”——这句话是多管道全部工程价值里最硬的一条动机。

Multiple Pipelines 数据流

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
                  Logstash 进程（单 JVM）
┌─────────────────────────────────────────────────────┐
│                                                     │
│  Pipeline A                                         │
│  ┌──────────┐  ┌──────┐  ┌────────────┐            │
│  │ input(s) │→ │queue │→ │filter+out  │→ ES-A      │
│  └──────────┘  └──────┘  └────────────┘            │
│  workers: 8   PQ: 512mb                             │
│                                                     │
│  Pipeline B                                         │
│  ┌──────────┐  ┌──────┐  ┌────────────┐            │
│  │ input(s) │→ │queue │→ │filter+out  │→ ES-B      │
│  └──────────┘  └──────┘  └────────────┘            │
│  workers: 2   memory queue                          │
│                                                     │
└─────────────────────────────────────────────────────┘
      共享：JVM heap、direct memory、OS 文件句柄、CPU 调度
      隔离：queue、worker 线程池、plugin 实例

每条 pipeline 有独立的 queue（各自的 queue.type，PQ 还带各自的 queue.max_bytes）、独立的 worker 线程池（pipeline.workers）、独立的插件实例生命周期。PQ 和 DLQ 的存储位置按 pipeline.id 分命名空间，天然隔离。一条管道崩溃或重载不影响其他管道。

pipelines.yml 配置

Multiple Pipelines 通过 config/pipelines.yml 定义，每个列表项对应一条 pipeline：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
# config/pipelines.yml
- pipeline.id: syslog
  path.config: "/etc/logstash/conf.d/syslog.conf"
  pipeline.workers: 8
  queue.type: persisted
  queue.max_bytes: 512mb

- pipeline.id: slow-query
  path.config: "/etc/logstash/conf.d/slow_query.conf"
  pipeline.workers: 2
  queue.type: memory

- pipeline.id: dlq-replay
  path.config: "/etc/logstash/conf.d/dlq_replay.conf"
  pipeline.workers: 1
  queue.type: memory

pipeline.id 是 pipeline 的唯一标识，在 Node Stats API、DLQ 路径、pipeline-to-pipeline 地址里都会用到。path.config 可以指向单个文件，也可以用 glob 表达式指向一个目录（目录下所有 .conf 文件合并成一条 pipeline 的配置）。

优先级的方向容易记反：不带参数启动时，Logstash 读 pipelines.yml 并实例化其中所有 pipeline；而一旦命令行带了 -e 或 -f，Logstash 就整个忽略 pipelines.yml，只在日志里留一条 warning。赢的一方是命令行参数。

两者也不是互斥关系。pipelines.yml 里某条 pipeline 没有显式写出的设置项，会回落到 logstash.yml 里的对应值——继承而非覆盖全部。上面例子里 dlq-replay 没写 pipeline.batch.size，它拿到的就是 logstash.yml（或内置默认）的 125。

pipeline-to-pipeline 通信

Multiple Pipelines 解决了隔离问题，但有时需要把多条 pipeline 串成拓扑：一条 pipeline 的 output 把 event 传给另一条 pipeline 的 input，整个过程在同一进程内完成，不经过 Kafka 或其他外部中间件。

这套机制叫 pipeline-to-pipeline，通过 virtual address 实现：

1
2
3
4
5
6
7
8
9
10
11
12
13
# 上游 pipeline 的 output 段
output {
  pipeline {
    send_to => ["downstream-addr"]
  }
}

# 下游 pipeline 的 input 段
input {
  pipeline {
    address => "downstream-addr"
  }
}

send_to 是一个字符串列表，对应下游 pipeline input 里声明的 address。virtual address 只在进程内有效，不是网络地址，不需要端口。

pipeline output 把 event 推入下游 pipeline 的 input queue，下游的 queue 类型（memory / PQ）由下游 pipeline 自己的配置决定。

四种拓扑模式

官方给出的是四种，两两成对：distributor 与 collector 是数据的收敛方向相反，output isolator 与 forked path 是同一份数据扇出后下游做不做额外处理的区别。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
拓扑一：distributor（分发）
  一条 input pipeline 接收所有数据，按条件分发到多条专门 pipeline

  input-router pipeline:
    input { beats { port => 5044 } }
    output {
      if [type] == "syslog"      { pipeline { send_to => ["syslog-proc"]   } }
      else if [type] == "nginx"  { pipeline { send_to => ["nginx-proc"]    } }
      else                       { pipeline { send_to => ["default-proc"]  } }
    }

拓扑二：output isolator（输出隔离）
  同一份数据扇出给多条只挂单个 output 的 pipeline，每条各带 PQ
  任一下游挂掉只堵自己的队列，不牵连其他 output

拓扑三：forked path（分叉处理）
  同一份数据扇出后，各下游 pipeline 再跑各自不同的 filter
  与 output isolator 的区别就在"下游有没有额外规则"

拓扑四：collector（汇聚）
  多条数据源 pipeline 各自采集，汇聚到一条公共处理 pipeline

  source-a pipeline: input → ... → output { pipeline { send_to => ["common"] } }
  source-b pipeline: input → ... → output { pipeline { send_to => ["common"] } }
  common pipeline:   input { pipeline { address => "common" } } → filter → output

distributor 是最常见的模式：一个统一入口承接所有 Beats/syslog 连接，避免每条子 pipeline 都开监听端口；再按 [type] 或其他元字段路由到对应的专门处理 pipeline。

collector 是 distributor 的反向，代价也相反：配置被简化到一处，但所有数据挤回同一条 pipeline，隔离度随之消失。

forked path 的位置值得单独交代。在 pipeline input/output 出现之前，"一份 event 走两套处理规则"只能靠 clone filter 复制再用 if/else 分流，配置迅速变得难读；现在这种需求的标准写法就是扇出到多条下游 pipeline，各自写自己的 filter 段。

output isolator：多管道最硬的那个用例

output isolator 值得给完整配置，因为它是"问题二"的正面解法。一台机器同时把日志写 ES 和某个 HTTP 端点，HTTP 端点因为例行维护频繁不可用——单管道下这段时间 ES 也一起写不进去。

1
2
3
4
5
6
7
8
9
10
11
# config/pipelines.yml
- pipeline.id: intake
  path.config: "/etc/logstash/conf.d/intake.conf"

- pipeline.id: buffered-es
  path.config: "/etc/logstash/conf.d/out_es.conf"
  queue.type: persisted

- pipeline.id: buffered-http
  path.config: "/etc/logstash/conf.d/out_http.conf"
  queue.type: persisted
1
2
3
4
5
6
7
8
9
10
11
# intake.conf
input { beats { port => 5044 } }
output { pipeline { send_to => ["es", "http"] } }

# out_es.conf
input  { pipeline { address => "es" } }
output { elasticsearch { hosts => ["http://es:9200"] } }

# out_http.conf
input  { pipeline { address => "http" } }
output { http { url => "https://partner.example.com/ingest" format => "json" } }

HTTP 端点不可用期间，event 堆在 buffered-http 自己的 PQ 里，buffered-es 照常写入。代价有三笔明账：磁盘占用最多翻倍（同一份数据进了两条 PQ）、序列化与反序列化成本约为单管道的三倍，以及一条兜底规则——任一下游 PQ 写满之后，两个 output 还是会一起停。PQ 买到的是缓冲时长，不是无限期解耦。

最小实验：Distributor 拓扑

1
2
3
4
5
6
7
8
9
10
11
12
13
14
# pipelines.yml
- pipeline.id: router
  path.config: "/etc/logstash/conf.d/router.conf"
  pipeline.workers: 4

- pipeline.id: syslog-proc
  path.config: "/etc/logstash/conf.d/syslog_proc.conf"
  pipeline.workers: 6
  queue.type: persisted
  queue.max_bytes: 256mb

- pipeline.id: nginx-proc
  path.config: "/etc/logstash/conf.d/nginx_proc.conf"
  pipeline.workers: 2
1
2
3
4
5
6
7
8
9
10
11
12
# router.conf
input {
  beats { port => 5044 }
}
filter {}
output {
  if [@metadata][beat] == "filebeat" and [fields][log_type] == "syslog" {
    pipeline { send_to => ["syslog-proc"] }
  } else {
    pipeline { send_to => ["nginx-proc"] }
  }
}
1
2
3
4
5
6
7
8
9
10
11
# syslog_proc.conf
input {
  pipeline { address => "syslog-proc" }
}
filter {
  grok { match => { "message" => "%{SYSLOGLINE}" } }
  date { match => ["timestamp", "MMM  d HH:mm:ss", "MMM dd HH:mm:ss"] }
}
output {
  elasticsearch { hosts => ["http://localhost:9200"] index => "syslog-%{+YYYY.MM.dd}" }
}

router pipeline 开一个 Beats 监听端口，按 log_type 字段把 event 分发给 syslog-proc 或 nginx-proc。两条处理 pipeline 各自有独立的 worker 数和 queue 配置，互不干扰。

投递语义与 ensure_delivery

串联点上到底什么时候阻塞、什么时候丢，由 pipeline output 的 ensure_delivery 参数和下游的状态共同决定。标准配置下 pipeline input/output 是 at-least-once，ensure_delivery 默认 true。

关键在于下游有两种"不通"，官方对它们的处理完全不同。unavailable 指下游 pipeline 正在启动或正在 reload，只有这一种状态受 ensure_delivery 影响。blocked 指下游 pipeline 存在且在跑，但它内部某个插件卡住了，比如 output 正在等 ES 响应。

1
2
3
4
5
ensure_delivery   下游 unavailable        下游 blocked
                  （启动中 / reload 中）   （插件卡住）
─────────────────────────────────────────────────────────
true（默认）       阻塞上游                阻塞上游
false             丢弃该消息              阻塞上游

这张表右下角那格是最容易踩的坑：把 ensure_delivery 设成 false 并不能让上游"不受下游影响"。它只覆盖 unavailable 一种情形，对 blocked 的下游无论取何值都照样阻塞上游。ensure_delivery => false 的真实用途是官方写明的那一个：想临时停掉某条下游 pipeline 而不牵连所有上游。

另外两条约束容易在 reload 时撞上。一条是避免成环：串联时要让数据保持单向流动，Logstash 关停时会等每条 pipeline 各自的工作完成，环形拓扑会让它永远等不到那一刻。另一条是 reload 的即时性：它会立刻按请求生效，包括删掉一条正在被上游写入的下游 pipeline。这会让上游阻塞，而且必须把下游恢复回来才能干净关停 Logstash。此时可以强制 kill，但在途 event 会丢，除非该 pipeline 开了 PQ。

对应回内部对象

virtual address 在 Logstash 内部是一个进程内的有界阻塞队列，实现在 PipelineBus / AbstractPipelineBus 两个 Java 类里。"有界"和"阻塞"这两点正是上面那张真值表的机制来源：队列满了写不进去，写入方就停在那里。

pipeline.workers 在 pipelines.yml 里针对每条 pipeline 单独配置，覆盖 logstash.yml 里的全局默认值。这是 Multiple Pipelines 最直接的资源隔离手段：高吞吐 pipeline 分配更多 worker，低频 pipeline 节省线程开销。

queue.max_bytes 是持久队列专属的：官方定义写的是"每条队列的总容量，以字节计"，前提是 queue.type: persisted。内存队列没有字节记账这回事，它的在途上界由 pipeline.workers × pipeline.batch.size 决定，不能独立配置。

多管道的内存账不止 heap 一项。官方给了可以直接代入的整机估算：

1
2
pipelines number * (pipeline threads * stack size + 2 * PQ page size)
  + direct memory + Java heap
1
2
3
4
5
6
7
8
9
10
11
12
三块内存的性质
──────────────────────────────────────────────────────────
Native memory         不受 -Xmx 控制，也无法设上界。PQ 的 page
（PQ page + 线程栈）   是 mmap 的，每条 PQ pipeline 至少要 head 和
                      tail 两个 page 常驻可访问，默认 64mb × 2
                      ≈ 128MB 起步

JVM direct memory     默认大小等于 heap，Beats/TCP/HTTP input
                      会用到。可用 -XX:MaxDirectMemorySize 限制，
                      官方建议考虑设成 heap 的一半

Java heap             -Xmx，全进程共享，不能按 pipeline 分配上限

对"多管道 + PQ"这个本篇主推的组合，第一块最容易被忽略：10 条 PQ pipeline 在 heap 之外还要多备出 1.4GB 上下的原生内存，而这部分没有任何 JVM 参数能约束。

模式提炼

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
模式：进程内多租户管道 + virtual address 串联

隔离维度（每条 pipeline 独立）：
  - queue（类型、容量、落盘路径）
  - worker 线程池（数量、batch size）
  - plugin 实例（插件状态不跨 pipeline 共享）
  - 生命周期（reload、崩溃不传染；但 p2p 串联点上，
    下游 reload 期间会阻塞上游）

共享维度（全进程共享）：
  - JVM heap（-Xmx 全局上限）
  - direct memory（默认等于 heap 大小）
  - OS 资源（文件句柄、网络端口）
  - CPU 调度（线程调度由 OS 决定，非绝对隔离）

按 pipeline 线性累加（容量规划时最易漏）：
  - PQ 的 mmap page，每条 PQ pipeline 约 128MB 起步

串联原则：
  - 下游 unavailable 才受 ensure_delivery 管，下游 blocked 一律阻塞上游
  - 数据保持单向流动，不成环，否则关停时无法干净退出
  - 默认值是按单条 pipeline 调的：每条 pipeline 默认每核 1 个 worker，
    多管道时要主动下调各自的 pipeline.workers，并把 pipeline 间的
    资源竞争算进容量规划

工程迁移表

Logstash 概念 Kafka 生态对应 Flink 对应 通用 ETL 对应
pipelines.yml（多管道定义） consumer group 配置 / Kafka Streams topology JobGraph 多算子链 多作业定义文件
pipeline.id（管道标识） consumer group id JobVertex id 作业名 / DAG 节点名
pipeline.workers（每管道并发） 消费者线程数 / partition 数 算子并行度 并发任务数
pipeline-to-pipeline virtual address 内部 topic / in-memory channel 算子间 network buffer 内存队列 / channel
distributor 拓扑 Router / filter by header KeyedBroadcastProcessFunction 分流 ETL 路由层
collector 拓扑 多 topic 同一 consumer group union / connect 算子 多源合并作业
output isolator 拓扑 每 sink 一个 topic + 独立 consumer 每 sink 独立分支 + 各自 buffer 每目标一条独立加载作业
共享 JVM heap 共享 broker / 共享 TaskManager 共享 TaskManager heap 共享执行引擎进程

Kafka Streams 的多拓扑（multiple topologies in one application）和 Logstash Multiple Pipelines 在设计意图上接近：在一个进程里并行运行多个处理逻辑，通过内部 channel 串联，避免跨网络的序列化开销。Flink 的算子链（operator chaining）在同一 TaskManager 线程内把多个算子合并成一个处理链，与 pipeline-to-pipeline 的"进程内直传"思路一致。核心取舍都是：进程内传递比网络传递快、但隔离性比独立进程弱，故障域扩展到整个进程。

常见误解

误解一：“Multiple Pipelines 等于多个 Logstash 进程”。Multiple Pipelines 是同一个 JVM 进程里的多条逻辑管道，共享 heap 和 OS 资源，不是多进程部署。真正的进程级隔离需要运行多个 Logstash 实例，代价是每个实例都有独立的 JVM overhead（内存、启动时间）。

误解二：“pipeline-to-pipeline 和写 Kafka 再读 Kafka 等价”。两者功能相似，但保证不同。pipeline-to-pipeline 的 virtual address 是进程内传递，进程崩溃时 in-process queue 里的 event 会丢（除非下游 pipeline 配置了 PQ）；写 Kafka 再消费是跨进程的持久化传递，进程崩溃后 Kafka 里的数据不丢。需要跨进程可靠传递时，Kafka 是更合适的选择；纯粹的进程内路由和资源隔离时，pipeline-to-pipeline 更轻量。

误解三：“pipelines.yml 存在就会盖掉命令行参数”。方向相反。不带参数启动时 Logstash 读 pipelines.yml；一旦带上 -e 或 -f，pipelines.yml 整个被忽略，只在日志里留一条 warning。用 -f 调试单条配置时以为多管道还在跑，就是这么来的。同时 pipelines.yml 与 logstash.yml 也不互斥：前者没写的设置项会回落到后者。

误解四：“ensure_delivery => false 能让上游不受下游影响”。它只覆盖 unavailable（下游启动中或 reload 中）一种情形。下游 pipeline 已经跑起来、只是内部插件卡住时属于 blocked，此时无论 ensure_delivery 取什么值都会阻塞上游。想真正把故障域切开，靠的是给下游各自配 PQ（output isolator 拓扑），而 PQ 也只是把阻塞推迟到队列写满。

误解五：“多 pipeline 一定比单 pipeline 快”。Multiple Pipelines 的核心收益是隔离和可管理性，不是绝对性能提升。如果原本的单管道已经充分利用了 CPU 和 I/O，把它拆成多个 pipeline 只是把同样的线程换了个组织方式，吞吐不会提升，反而增加了 pipeline 间协调和调度的开销。选择多管道的理由应该是"这些数据流的处理要求差异大，需要独立的 worker/queue 配置"，而不是"多管道=更快"。

练习

  1. 在本地用 pipelines.yml 定义两条 pipeline：pipeline A 用内存 queue 跑 generator input，pipeline B 用 pipeline { address => "output-b" } input 接收来自 A 的 event，最后用 stdout { codec => rubydebug } 输出。观察 pipeline A 的 output 发出的 event 是否在 pipeline B 的 stdout 里出现，确认 virtual address 串联生效。

  2. 修改上面的实验，把 pipeline B 的 queue.type 改成 persisted，然后在 pipeline B 的 filter 里加一个 sleep 操作（用 ruby { code => "sleep 0.5" }）模拟处理慢。观察 pipeline A 的 generator 是否会被背压。接着把 pipeline A 的 output 改成 pipeline { send_to => ["output-b"] ensure_delivery => false } 重跑一次，确认背压依然存在——此时 B 属于 blocked 而非 unavailable。最后把 B 从 pipelines.yml 里删掉触发 reload，观察 A 是否阻塞，以此把真值表的两列各验一遍。

  3. 思考题：给定一个场景——需要采集三个来源（syslog、nginx access log、应用 JSON log），分别做不同的 Grok 解析，但最终都写到同一个 ES 索引。用 Distributor 拓扑（一个 router pipeline + 三个 proc pipeline）和单管道（用 if/else if 分支）各自实现，列出两种方案在 worker 资源分配、queue 独立性、配置可维护性三个维度上的差异，决定哪种方案更适合这个场景。

系列导航

序号 主题
00 导读：核心对象是 event，骨架是三段管道
01 架构：JRuby、JVM 与 pipeline 的运行形态
02 event 模型：@timestamp、@metadata 与字段引用
03 codec：字节流与 event 的边界转换
04 input 插件：拉取、监听与 Beats 接入
05 Grok 的本质：命名正则加预定义 pattern
06 dissect 与结构化 filter：放弃回溯换吞吐
07 常用 filter 组合：mutate、date、geoip 与条件
08 output 插件：Elasticsearch output 与批量写入
09 pipeline 执行模型：worker、batch 与背压
10 内存队列 vs 持久队列：可靠性的分界线
11 死信队列（DLQ）：无法处理的 event 去哪
12 Multiple Pipelines 与 pipeline-to-pipeline（本篇）
13 监控：Node Stats API、hot threads 与瓶颈定位
14 性能调优：JVM heap、批处理与持久队列磁盘
15 Logstash vs Beats vs Ingest Pipeline：该用谁
16 Logstash vs Fluentd vs Vector：日志管道的三种取舍
17 Logstash 的演进与 Elastic Agent 的冲击

参考资料