plumber实战：10个常用场景示例详解

plumber实战：10个常用场景示例详解

【免费下载链接】plumberA swiss army knife CLI tool for interacting with Kafka, RabbitMQ and other messaging systems.项目地址: https://gitcode.com/gh_mirrors/pl/plumber

plumber是一款功能强大的命令行工具，被誉为消息系统的"瑞士军刀"，能够与Kafka、RabbitMQ等多种消息系统无缝交互。本文将通过10个实用场景示例，带你快速掌握plumber的核心用法，轻松应对日常消息处理任务。

1. 快速读取Kafka消息

Kafka作为主流的分布式消息系统，经常需要查看主题中的消息内容。使用plumber可以轻松实现：

plumber read kafka --topics orders --address="broker1.domain.com:9092"

如果需要持续监听消息，添加--continuous参数即可：

plumber read kafka --topics orders --address="broker1.domain.com:9092" --continuous

2. 向Kafka写入消息

除了读取消息，plumber也能方便地向Kafka写入数据。支持直接输入字符串：

plumber write kafka --topics test --input "Hello Kafka"

或者从文件读取数据：

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

对于JSON数组格式的数据，使用--input-as-json-array参数可以批量写入：

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

3. 与RabbitMQ Streams交互

RabbitMQ Streams是RabbitMQ的流处理功能，plumber提供了完整的支持。首先读取流数据：

plumber read rabbit-streams --declare-stream --declare-stream-size 10mb --stream testing

然后写入数据：

plumber write rabbit-streams --stream testing --input-data "Hello Rabbit Streams"

4. AWS SQS消息处理

处理AWS SQS队列消息时，plumber可以指定最大消息数量：

plumber read aws-sqs --queue-name=orders --max-num-messages=10

添加--auto-delete参数可以在读取后自动删除消息：

plumber read aws-sqs --queue-name=orders --max-num-messages=10 --auto-delete

5. NATS消息系统操作

NATS是一个轻量级的消息系统，plumber对其提供了全面支持。读取消息：

plumber read nats --address="nats://user:pass@nats.test.io:4222" --subject "test-subject"

写入消息：

plumber write nats --subject testing --input-data "Hello NATS"

对于NATS JetStream，使用专用命令：

plumber read nats-jetstream --stream testing plumber write nats-jetstream --stream testing --input "Hello JetStream"

6. 消息格式转换与解码

plumber支持多种消息格式的解码，如Thrift格式：

plumber read kafka --topics thrifttest --decode-type thrift --pretty

先写入Thrift二进制数据：

plumber write kafka --topics thrifttest --input-file test-assets/thrift/test_message.bin

7. 数据库变更捕获(CDC)

plumber可以捕获MongoDB的变更数据：

plumber read mongo --database plumbertest --continuous

对于PostgreSQL，使用：

plumber read postgres --replication-slot-name plumber_slot --publisher-name streamdal_plumber --database postgres --address localhost:5432 --username postgres --password postgres

8. 消息中继(Relay)功能

plumber的中继功能可以在不同消息系统间转发消息。例如从RabbitMQ中继到Kafka：

plumber relay rabbit --to kafka --topics relay-test

从AWS SQS中继到Azure Service Bus：

plumber relay aws-sqs --queue-name=source-queue --to azure-service-bus --queue target-queue

9. MQTT协议支持

物联网场景中常用的MQTT协议，plumber也能轻松应对：

plumber read mqtt --address tcp://localhost:1883 --topic iotdata --qos-level at_least_once

对于SSL加密连接：

plumber read mqtt --address ssl://localhost:8883 --topic iotdata --qos-level at_least_once --tls-ca-cert=/path/to/ca_certificate.pem --tls-client-key=/path/to/client_key.pem --tls-client-cert=/path/to/client_certificate.pem

10. 内存数据库消息处理

对于Redis的Pub/Sub和Streams功能，plumber提供了专门支持：

# 读取Redis Pub/Sub消息 plumber read redis-pubsub --address="localhost:6379" --channels="new-orders" # 读取Redis Streams消息 plumber read redis-streams --address="localhost:6379" --streams="new-orders"

安装与使用

要开始使用plumber，首先克隆仓库：

git clone https://gitcode.com/gh_mirrors/pl/plumber cd plumber make build

查看所有可用命令：

plumber --help

每个子命令都有详细帮助：

plumber read --help plumber read kafka --help

总结

plumber作为消息系统的瑞士军刀，提供了丰富的功能和灵活的命令行接口。无论是开发调试、数据迁移还是系统集成，plumber都能大大简化工作流程。通过本文介绍的10个常用场景，相信你已经对plumber有了基本了解。更多高级用法和示例，请参考项目文档：docs/examples.md。

希望这篇实战指南能帮助你更好地利用plumber处理各种消息系统任务，提高工作效率！

【免费下载链接】plumberA swiss army knife CLI tool for interacting with Kafka, RabbitMQ and other messaging systems.项目地址: https://gitcode.com/gh_mirrors/pl/plumber