Watermill

Watermill 是一个用于高效处理消息流的 Go 库。它旨在构建事件驱动的应用程序，支持事件溯源、基于消息的 RPC、saga 以及您能想到的其他功能。您可以使用传统的发布/订阅实现，如 Kafka 或 RabbitMQ，也可以根据需要使用 HTTP 或 MySQL binlog。

目标

易于理解。
通用 - 事件驱动架构、消息传递、流处理、CQRS - 根据您的需求使用。
快速（参见基准测试）。
通过中间件、插件和发布/订阅配置实现灵活性。
可靠 - 使用经过验证的技术并通过压力测试（参见稳定性）。

入门

选择您最喜欢的方式或按顺序进行：

遵循入门指南。
查看以下示例。
阅读完整文档：https://watermill.io/

我们的在线实践培训

示例

基础
- 您的第一个应用 - 从这里开始！
- 实时订阅
- 路由器
- 指标
- 使用 protobuf 的 CQRS
发布/订阅用法
- 这些示例是入门指南的一部分，展示了单个发布/订阅的使用。
真实世界示例
完整项目
- 带有实时代码重载的 NATS 示例
- RabbitMQ、webhook 和 Kafka 集成

背景

构建分布式和可扩展的服务并不像某些人所说的那么简单。编写这样的系统需要大量隐藏的知识。就像您不需要了解整个 TCP 堆栈就可以创建 HTTP REST 服务器一样，您也不应该需要学习所有这些知识就可以开始构建消息驱动的应用程序。

Watermill 的目标是使消息通信变得像使用 HTTP 路由器一样简单。它提供了开始使用事件驱动架构所需的工具，并允许您在使用过程中学习细节。

Watermill 的核心是一个简单的接口：

func(*Message) ([]*Message, error)

您的处理程序接收一条消息，并决定是发布新消息还是返回错误。接下来会发生什么取决于您选择的中间件。

您可以在我们的介绍 Watermill 博客文章中了解更多关于我们的动机。

发布/订阅

所有发布者和订阅者都必须实现一个接口：

type Publisher interface {
	Publish(topic string, messages ...*Message) error
	Close() error
}

type Subscriber interface {
	Subscribe(ctx context.Context, topic string) (<-chan *Message, error)
	Close() error
}

支持的发布/订阅系统：

AMQP 发布/订阅 (github.com/ThreeDotsLabs/watermill-amqp/v2)
Bolt 发布/订阅 (github.com/ThreeDotsLabs/watermill-bolt)
Firestore 发布/订阅 (github.com/ThreeDotsLabs/watermill-firestore)
Google Cloud 发布/订阅 (github.com/ThreeDotsLabs/watermill-googlecloud)
HTTP 发布/订阅 (github.com/ThreeDotsLabs/watermill-http)
io.Reader/io.Writer 发布/订阅 (github.com/ThreeDotsLabs/watermill-io)
Kafka 发布/订阅 (github.com/ThreeDotsLabs/watermill-kafka/v2)
NATS 发布/订阅 (github.com/ThreeDotsLabs/watermill-nats)
Redis Stream 发布/订阅 (github.com/ThreeDotsLabs/watermill-redisstream)
SQL 发布/订阅 (github.com/ThreeDotsLabs/watermill-sql/v2)

所有发布/订阅实现的文档可以在文档中找到。

非官方库

找不到你喜欢的发布/订阅或库集成？查看Awesome Watermill。

如果你知道其他库或者是其作者，请将其添加到列表中。

贡献

请查看我们的贡献指南。

稳定性

Watermill v1.0.0 已经发布并可用于生产环境。公共API是稳定的，不会在不改变主版本号的情况下进行更改。

为了确保所有发布/订阅系统都稳定且可安全地用于生产环境，我们创建了一套测试，每个实现在合并到主分支之前都需要通过这些测试。所有测试还以压力模式执行 - 这意味着我们将所有测试并行运行20次。

所有测试都在启用竞态条件检测器的情况下运行（测试中使用-race标志）。

有关调试测试的更多信息，你应该查看测试故障排除指南。

基准测试

用于对发布/订阅系统进行基准测试的初始工具可以在watermill-benchmark中找到。

所有基准测试都在单个16 CPU的虚拟机实例上进行，在Docker Compose中运行一个二进制文件和依赖项。

这些数字旨在粗略估计不同发布/订阅系统可以处理消息的速度。请记住，根据设置和配置的不同，结果可能会有很大差异（既可能低得多，也可能高得多）。

以下是消息大小为16字节时的简短版本。

发布/订阅	发布（消息/秒）	订阅（消息/秒）
GoChannel	331,882	118,943
Redis Streams	61,642	11,213
NATS Jetstream（16个订阅者）	49,255	33,009
Kafka（单节点）	44,090	108,285
SQL（MySQL）	5,599	167
SQL（PostgreSQL，批量大小=1）	3,834	455
Google Cloud 发布/订阅	3,689	30,229
AMQP	2,702	13,192