batchcorp/plumber

plumber：消息系统的瑞士军刀

!https://raw.githubusercontent.com/batchcorp/plumber/master/assets/plumber_logo_full.png

https://github.com/batchcorp/plumber/workflows/master/badge.svg](https://github.com/batchcorp/plumber/actions/workflows/master-test.yaml) https://goreportcard.com/badge/github.com/batchcorp/plumber](https://goreportcard.com/report/github.com/batchcorp/plumber)

镜像概述和主要用途

plumber是一款CLI开发工具，用于检查、管道传输、消息传递和重定向Kafka、RabbitMQ、GCP PubSub等多种消息系统中的数据。[1]
_{[1] 它就像消息系统的curl。}

为什么需要它？

消息系统通常像黑盒——要了解其中传输的数据，需要编写脆弱的消费者代码，最终却会被丢弃。plumber让您无需再浪费时间编写一次性代码，可用于查看队列和数据流、连接不同系统或调试事件驱动系统。

演示

!https://raw.githubusercontent.com/batchcorp/plumber/master/assets/demo.gif

核心功能和特性

• 安全查看数据流内容
• 向任何系统写入纯文本或编码数据
• 在不同系统间路由数据
• 实时解码protobuf/avro/thrift/JSON数据
  • 支持深度和浅层protobuf信封类型
  • 支持google.protobuf.Any字段
• 将数据中继到Batch平台
• 将变更数据捕获事件发送到Batch平台
• 在本地网络中将事件重放至消息系统
• 更多功能（完整列表：plumber -h）

技术特性

• 多格式编解码支持
  • Protobuf（深度和浅层信封）
  • Avro、Thrift、Flatbuffer、GZip、JSON、JSONPB（protobuf序列化为JSON）、Base64
• --continuous持续读取支持（类似tail -f）
• 支持大多数消息系统
• 支持通过字符串、文件或管道写入
• 观察、中继和归档消息数据
• 单二进制文件，零配置，易于安装

使用场景和适用范围

• 开发调试：查看消息系统中的实时数据，验证数据格式和内容
• 数据迁移：在不同消息系统间路由数据
• 事件重放：将历史事件重放至本地消息系统进行测试
• 数据格式转换：实时解码/编码不同格式的数据（如protobuf转JSON）
• 系统集成：连接分散的消息系统，实现数据互通

安装方法

通过brew

bash
$ brew tap batchcorp/public
$ brew install plumber

手动安装

• https://github.com/batchcorp/plumber/releases/latest/download/plumber-darwin
• https://github.com/batchcorp/plumber/releases/latest/download/plumber-linux
• https://github.com/batchcorp/plumber/releases/latest/download/plumber-windows.exe

plumber是单二进制文件，安装只需下载、赋予执行权限并从shell调用。示例命令：

bash
$ curl -L -o plumber https://github.com/batchcorp/plumber/releases/latest/download/plumber-darwin
$ chmod +x plumber
$ mv plumber /usr/local/bin/plumber

使用方法和配置说明

写入消息

bash
❯ plumber write kafka --topics test --input foo
INFO[0000] Successfully wrote message to topic 'test'    backend=kafka
INFO[0000] Successfully wrote '1' message(s)             pkg=plumber

读取消息

bash
❯ plumber read kafka --topics test
INFO[0000] Initializing (could take a minute or two) ...  backend=kafka

------------- [Count: 1 Received at: 2021-11-30T12:51:32-08:00] -------------------

+----------------------+------------------------------------------+
| Key                  |                                     NONE |
| topic                |                                     test |
| Offset               |                                        8 |
| Partition            |                                        0 |
| Header(s)            |                                     NONE |
+----------------------+------------------------------------------+

foo

注：添加-f参数可进行持续读取（类似tail -f）

通过管道写入消息

写入多条消息

注：多条消息以换行符分隔。

bash
$ cat mydata.txt
line1
line2
line3

$ cat mydata.txt | plumber write kafka --topics foo

INFO[0000] Successfully wrote message to topic 'foo'  pkg=kafka/write.go
INFO[0000] Successfully wrote message to topic 'foo'  pkg=kafka/write.go
INFO[0000] Successfully wrote message to topic 'foo'  pkg=kafka/write.go

将JSON数组的每个元素作为消息写入

bash
$ cat mydata.json
[{"key": "value1"},{"key": "value2"}]

$ cat mydata.json | plumber write kafka --topics foo --json-array

INFO[0000] Successfully wrote message to topic 'foo'  pkg=kafka/write.go
INFO[0000] Successfully wrote message to topic 'foo'  pkg=kafka/write.go

获取帮助

使用--help标志可查看不同命令部分的可用标志：

bash
$ plumber --help
$ plumber read --help
$ plumber read kafka --help

支持的消息系统

• Kafka
• RabbitMQ
• RabbitMQ Streams
• Google Cloud Platform PubSub
• MQTT
• Amazon Kinesis Streams (新增)
• Amazon SQS
• Amazon SNS（发布）
• ActiveMQ（STOMP协议）
• Azure Service Bus
• Azure Event Hub
• NATS
• NATS Streaming (Jetstream)
• Redis-PubSub
• Redis-Streams
• Postgres CDC（变更数据捕获）
• MongoDB CDC（变更数据捕获）
• Apache Pulsar
• NSQ
• KubeMQ

注：如果您的消息系统未被支持，请提交issue，我们将尽力支持！

Kafka注意事项

确保所有plumber实例使用相同的消费者组。

RabbitMQ注意事项

确保所有plumber实例指向同一个队列。

布尔标志说明

要将布尔标志设为false，在标志前添加--no。
例如：--queue-declare默认值为true，设为false时使用--no-queue-declare。

隧道模式

plumber现在可作为重放目标（隧道）。隧道模式允许在本地网络运行plumber实例，该实例将在Batch平台中作为_重放目标_可用。
这避免了为将Batch集合中的消息重放至消息总线而修改防火墙的需求。
完整文档见：[***]

高性能与高可用性

plumber提供“服务器模式”，使其作为高可用集群运行。
有关“服务器模式”的更多信息：文档
服务器模式示例：https://github.com/batchcorp/plumber/blob/master/docs/server.md

扩展文档

• https://github.com/batchcorp/plumber/blob/master/docs/examples.md%EF%BC%9A%E6%9B%B4%E5%A4%9A%E4%BD%BF%E7%94%A8%E7%A4%BA%E4%BE%8B
• https://github.com/batchcorp/plumber/blob/master/docs/env.md%EF%BC%9A%E6%94%AF%E6%8C%81%E7%9A%84%E7%8E%AF%E5%A2%83%E5%8F%98%E9%87%8F%E5%88%97%E8%A1%A8
• https://github.com/batchcorp/plumber/blob/master/docs/metrics.md%EF%BC%9Aplumber%E5%85%AC%E5%BC%80%E7%9A%84%E6%8C%87%E6%A0%87%E4%BF%A1%E6%81%AF
• https://github.com/batchcorp/plumber/blob/master/docs/server.md%EF%BC%9A%E6%9C%8D%E5%8A%A1%E5%99%A8%E6%A8%A1%E5%BC%8F%E8%BF%90%E8%A1%8C%E7%A4%BA%E4%BE%8B

关于Batch

我们是分布式系统爱好者，创立了Batch公司。公司专注于为复杂系统和工作流解决数据流可观测性问题，目标是让_每个人_都能构建异步系统，而不必担心引入过多复杂性。在工作中，我们开发了用于读写消息系统消息的工具，发现该领域严重缺乏工具，因此创建了plumber。

名称由来

我们自认为是“互联网水管工”，因此这个名字很合适 :)

致谢

特别感谢https://github.com/jhump%E5%8F%8A%E5%85%B6%E5%87%BA%E8%89%B2%E7%9A%84https://github.com/jhump/protoreflect%E5%BA%93%EF%BC%8C%E6%B2%A1%E6%9C%89%E5%AE%83%EF%BC%8Cplumber%E7%9A%84%E5%AE%9E%E7%8E%B0%E4%B8%8D%E4%BC%9A%E5%A6%82%E6%AD%A4%E7%AE%80%E5%8D%95%E3%80%82_%E8%B0%A2%E8%B0%A2%EF%BC%81_

发布流程

推送新plumber版本的步骤：

1. git tag v0.18.0 master
2. git push origin v0.18.0
3. 观察github action
4. 新版本将自动创建于：https://github.com/batchcorp/plumber/releases/
5. 更新发布说明，包含相关信息
6. 更新https://github.com/batchcorp/homebrew-public/blob/master/plumber.rb%E7%9A%84SHA%E5%92%8C%E7%89%88%E6%9C%AC%E5%BC%95%E7%94%A8

贡献

我们欢迎贡献！在提交PR前，请先打开issue讨论您计划的工作。准备好PR时，请添加良好的测试，我们将尽快合并！