主题
Go 消息与工作流工具选型：先选需要哪一层，再检查确认和重试
要把“生成账单”放到后台，至少有三种做法：直接消费消息并执行，用任务队列管理重试和执行状态，或者把它放进一个能等待、补偿、跨重启恢复的工作流。三者都可能运行 Go 函数，但替应用承担的责任不同。
本文覆盖七个工具：Kafka 客户端 franz-go、segmentio/kafka-go，RabbitMQ 客户端 amqp091-go，NATS 客户端 nats.go，任务队列 asynq、river，以及工作流 Temporal。重点是常用入口、确认时点、失败以后谁负责恢复，以及哪类需求值得承担它的复杂度。
于 2026-10-13 从官方 Go 模块代理核对版本，并以校验和服务下载固定依赖。本次运行环境为 Go 1.27.2、Linux/amd64。七个库都参与实际编译；实验运行了离线 API、应用 handler、确认边界测试和 Temporal SDK 内存测试环境。没有连接 Kafka、RabbitMQ、NATS、Redis、数据库或 Temporal Server，不能把这些结果当成服务端集成验证。
三层工具不要放在一张“谁更快”的榜单上
正在绘制图示…
查看图源代码
flowchart TB
A["应用业务：账单、通知、订单处理"] --> C["消息客户端<br/>记录、路由、确认、消费位置"]
A --> T["任务队列<br/>领取、重试、定时、并发、任务状态"]
A --> W["工作流<br/>步骤顺序、等待、历史、恢复、补偿逻辑"]
C --> B["对应的 Broker"]
T --> S["Redis 或数据库等任务存储"]
W --> TS["Temporal Service 与事件历史"]
W --> AC["Activities 执行业务副作用"]
这是责任关系，不是说每个应用必须同时部署三层。
| 工具与固定版本 | 主要层次 | 常用入口 | 值得选择的条件 |
|---|---|---|---|
franz-go v1.22.2 | Kafka 客户端 | kgo.NewClient、Produce/ProduceSync、PollFetches/PollRecords、提交与事务 API | 需要细致控制 Kafka 消费、生产、事务与协议功能 |
kafka-go v0.4.51 | Kafka 客户端 | Reader、Writer、FetchMessage、CommitMessages | 希望用比较直接的 Reader/Writer 抽象，所需协议能力已验证 |
amqp091-go v1.15.0 | RabbitMQ AMQP 0-9-1 客户端 | DialConfig、Channel、声明路由、Consume、Ack、confirm | 系统需要 RabbitMQ 的交换机与队列模型 |
nats.go v1.54.0 | Core NATS / JetStream 客户端 | Connect、发布订阅、jetstream.New、consumer、ack | 已选 NATS，并明确区分实时与持久消息需求 |
asynq v0.26.0 | Redis 后台任务队列 | NewTask、Enqueue、Server、ServeMux | 有 Redis，想直接管理任务、重试、定时与并发 |
river v0.49.0 | 数据库后台任务队列 | JobArgs、Worker、Insert/InsertTx、Start | 想把任务入队与业务数据库事务放在同一提交边界 |
go.temporal.io/sdk v1.49.0 | 持久工作流 SDK | client、worker、workflow、activity、testsuite | 流程包含多步、长期等待、重试与跨部署恢复 |
这些是本文固定验证的版本，不是“永远使用最新版本”的建议。部署前还要核对服务端版本、认证、协议支持和升级说明。尤其不能因为一套客户端能连接 Kafka，就推断它实现了所有新协议。
franz-go：控制得细，也要理解消费生命周期
franz-go 的高层入口是 pkg/kgo。生产用 Produce 的逐条回调，或 ProduceSync(...).FirstErr() 检查结果；消费用 poll 取回 records，再按处理结果提交进度。需要 Kafka 内部读-处理-写事务时，它也有事务相关 API。使用之前先确定普通消费组还是 Share Group，因为两种进度模型不同。[1]
本版本默认开启幂等生产，并使用全部 ISR 确认。它会拒绝互相冲突的配置，而不是让代码带着更弱的保证悄悄运行：
go
c, err := kgo.NewClient(kgo.RequiredAcks(kgo.NoAck()))
if c != nil { c.Close() }
fmt.Println("franz invalid acks:", err)实际输出：
text
franz invalid acks: idempotency requires acks=all这次失败发生在客户端配置校验阶段，没有建立 Broker 连接。它证明固定版本确实拒绝这组配置，不证明任何集群写入已经持久化。若主动关闭幂等来接受弱确认，要明确接受重复和丢失风险，而不是为消掉错误机械地改选项。[1:1]
消费端有一个容易被异步代码破坏的前提。本版本普通组默认每 5 秒自动提交；默认自动提交使前一次 poll 的记录在下一次 poll 后进入可提交范围。若把记录交给后台 goroutine 就继续 poll，前一批业务可能还没完成，进度却已经可被提交。可以选 DisableAutoCommit 后显式提交已处理记录，或选 AutoCommitMarks 并在完成后标记；这两套选项不能混用。[1:2]
显式提交也要处理 Rebalance：分区被撤销后，旧 worker 的工作可能尚未结束。BlockRebalanceOnPoll 能让应用控制何时允许交接，但必须及时调用 AllowRebalance，不能无限阻塞。实际使用要把批量上限、取消、已完成前缀、提交错误和退出流程一起设计。Close 不是“所有业务已成功”的替代证明。
它适合需要这些控制点的 Kafka 应用，代价是选项和生命周期知识更多。若使用事务，必须保持正确的事务身份、消费组元数据、输入位置与输出记录关系；Kafka 事务仍不回滚外部数据库或邮件。
kafka-go：先分清 ReadMessage 与 FetchMessage
Reader 管读取与消费组，Writer 管生产与批量。名字贴近 Go 常见接口，入门较直观，但“读到”和“业务完成”之间仍有一条不能省略的边界。
在使用消费组时，ReadMessage 会自动提交返回消息对应的位置，而调用方此时尚未处理业务。需要先做业务再提交，就用 FetchMessage 与 CommitMessages。后者对同分区较大 offset 的提交会覆盖前面的进度，不能用于随意确认乱序完成的某一条。[2]
下面这个完整辅助函数既可接收真实 *kafka.Reader，也可用 fake 检验动作顺序：
go
func HandleKafka(ctx context.Context, r FetchCommit, work func([]byte) error) error {
m, err := r.FetchMessage(ctx)
if err != nil { return err }
if err := work(m.Value); err != nil { return err }
return r.CommitMessages(ctx, m)
}本文实际测试了两种故障：业务失败时 fake 的提交次数必须为 0；提交失败必须原样返回给调用方。测试均通过。这里测试的是应用辅助函数的控制流，真实网络读取与提交没有发生。
外层循环收到业务错误后，不能直接继续读下一条再提交更大的位置，否则依然会跨过失败记录。应停止该有序通道、重试或按明确策略隔离。默认 CommitInterval=0 使用同步提交；设为非零会改变提交的时间关系，不能仍把一次方法返回理解为同样的持久检查点。[2:1]
Writer 的零值确认级别值得专门检查
直接构造 kafka.Writer 时，本版本 RequiredAcks 默认是 RequireNone，也就是 0。它和旧 WriterConfig / NewWriter 路径的默认行为不同：
go
fmt.Println("kafka writer zero acks:", int((&kafka.Writer{}).RequiredAcks))text
kafka writer zero acks: 0需要等待同步副本确认，应显式设置 RequiredAcks: kafka.RequireAll，并配合 Broker 的最小 ISR 策略。同步 WriteMessages 要检查逐条写入错误；Async: true 时还要观察 Completion 回调，方法返回不代表发布完成。关闭前需处理在途请求和 Close 错误。[3]
这个库适合其 Reader/Writer 抽象已覆盖需求的项目。如果需求依赖 Kafka 新事务/消费协议或特定管理 API，应先查所选版本实现和集成测试，不能用“兼容较新 Broker”替代功能验收。本文没有做两个 Kafka 客户端的性能比较，因此不声称谁更快。
amqp091-go：生产确认与消费 Ack 是两个动作
amqp091-go 由 RabbitMQ 团队维护，针对 AMQP 0-9-1；AMQP 1.0 是另一套协议。常见生产流程是建立连接和 channel、声明 exchange/queue/binding、开启 Confirm、发布并等待确认，同时处理 mandatory 产生的 return。常见消费流程是 Qos 设置 prefetch、Consume 使用 autoAck=false、业务完成后 Delivery.Ack(false)。[4]
Ack(false) 中的 false 表示只确认当前 delivery tag；true 是累计确认范围，不能在一堆并行任务中随便使用。delivery tag 属于原 channel，也不应保存在重连后当作同一有效投递身份继续使用。
go
func HandleAMQP(d amqp.Delivery, work func([]byte) error) error {
if err := work(d.Body); err != nil { return err }
return d.Ack(false)
}我们用库公开的 Acknowledger 接口记录调用，让真实 Delivery.Ack 转发到测试替身。成功分支的实际输出：
text
amqp ack: 1 7 false表示调用一次、tag 为 7、没有累计确认。测试还验证业务失败不调用 ack，以及 ack 失败传回调用方。它没有验证 RabbitMQ 收到了确认；这个区别正是 mock 测试的边界。
失败分支这里只返回错误，外层消费者还必须决定重试、负确认或隔离。立即反复 Nack(..., requeue=true) 可能造成热循环；连接断开后重投又要求业务幂等。发布端 confirm 也不能与这个消费 ack 相互替代。
旧教程的两条说法需要按版本更新
第一，v1.15.0 已有实验性的自动连接和拓扑恢复，入口是 Config.Recovery；默认 nil 时关闭。启用后的恢复范围、状态通知和实验性 API 风险要看当前配置，不能再笼统说这个库“完全不支持自动恢复”。恢复连接仍不能确定网络断开前那次业务发布到底有没有成功，未决消息仍需重试与幂等策略。[5]
第二，函数名有 WithContext 不等于它会持续用 context 中断整个发送过程。v1.15.0 的 PublishWithContext 先检查调用开始时 context 是否取消，再调用 Publish；源码没有在整个发布期间持续监听该 context。确认等待、网络超时和连接恢复需要各自安排，不能只给它包一层 context.WithTimeout 就认为全部 I/O 都有同样的截止语义。[6]
优势是协议和路由模型表达直接；代价是应用需要认真处理 channel 生命周期、confirm/return、消费者确认和重连状态。
nats.go：Core Publish 与 JetStream Publish 要分开
nats.go 同时支持 Core NATS 和 JetStream。Core 的 Publish/Subscribe/QueueSubscribe/Request 适合对应的实时消息模式；需要持久接收确认、重放和 durable consumer 时，应使用 jetstream 包的 API，而不是仅把 subject 名字改成某个 stream 的匹配项。
常用入口是 jetstream.New(nc)，随后创建/取得 stream 与 consumer；生产用 js.Publish(ctx, ...) 等取得 publish ack，消费用 Fetch 或 Consume，完成后 Ack。需要等待服务端确认收到消费 ack，新 jetstream.Msg 使用 DoubleAck(ctx)；旧 API 里常见的名字是 AckSync()。不要把两代示例混在一个类型上。[7]
下面实际创建消息 header 和 consumer 配置，但没有把配置发给服务端：
go
m := nats.NewMsg("orders.created")
m.Header.Set(jetstream.MsgIDHeader, "e1")
jsCfg := jetstream.ConsumerConfig{
Durable: "billing", AckPolicy: jetstream.AckExplicitPolicy, MaxDeliver: 3,
}
fmt.Println("nats message/config:", m.Subject, m.Header.Get(jetstream.MsgIDHeader),
jsCfg.AckPolicy == jetstream.AckExplicitPolicy, jsCfg.MaxDeliver)text
nats message/config: orders.created e1 true 3稳定的 Nats-Msg-Id 供 JetStream 在配置去重窗口内识别重复发布；它不是永久业务去重数据库。明确设置 AckExplicitPolicy 可以把确认责任留给应用；MaxDeliver: 3 只是交付上限配置，不会自动生成完整 DLQ 处理流程。
还需要考虑 AckWait 与处理时长、暂时错误的 Nak/延迟重试、长期处理的进度通知、在途上限和 pull 批量。Consume 的回调外观可以由 SDK 后台持续拉取实现，不能据此推断服务端推拉模型。关闭时用与当前 API 匹配的停止/排空流程，避免在途业务尚未结束就释放消费职责。
选择它的理由通常是 NATS 生态已经贴合系统需求。不能因为 Core NATS 简单，就省略 JetStream 的存储、副本、容量与恢复测试。
Asynq：把 Redis 上的后台任务生命周期交给库
Asynq 用 Redis 保存任务状态，提供任务类型、handler、并发、队列优先级、延迟执行、重试和管理能力。生产端 NewTask 创建类型与 payload，再由 Client.Enqueue 持久入队；消费端把 handler 注册到 ServeMux，交给 Server.Run 执行。NewTask 本身只是创建本地对象，不代表 Redis 已接收任务。[8]
下面的 handler 把账单参数渲染成一行文本，写入测试缓冲区；它没有生成真实 PDF，也没有发邮件：
go
var out bytes.Buffer
h := InvoiceHandler{Out: &out}
task := asynq.NewTask("invoice:render", []byte(`{"order_id":"o1"}`), asynq.MaxRetry(3))
if err := h.ProcessTask(context.Background(), task); err != nil { panic(err) }
fmt.Printf("asynq result: %s", out.String())
err := h.ProcessTask(context.Background(), asynq.NewTask("invoice:render", []byte(`{}`)))
fmt.Println("asynq invalid skips retry:", errors.Is(err, asynq.SkipRetry))text
asynq result: invoice:o1
asynq invalid skips retry: true参数缺少订单号，重试也不会自己补出来，所以 handler 包装 asynq.SkipRetry。暂时性的存储错误则返回普通错误，让任务按策略重试。测试只直接运行 handler；是否入队、何时重试、归档状态怎样变化仍需要 Redis 集成测试。
v0.26.0 的默认最大重试次数是 25；MaxRetry(3) 指初次之后最多再重试 3 次。超时、deadline、ProcessIn/ProcessAt 和重试策略应按任务耗时明确设置。Unique 基于类型、payload、队列和有效期等条件减少重复入队，但锁有期限，完成后也有释放语义，不能替代业务执行幂等。[9]
Asynq 适合已经接受 Redis 依赖、希望少写任务管理代码的应用。它不解决“SQL 事务提交成功、Redis 入队失败”的双写裂缝，需要 Outbox 等方案。Redis 的持久化、故障切换、内存和淘汰策略也成为任务可靠性的一部分。
River：让任务入队参与已有的数据库事务
River 用类型化 JobArgs 描述参数，以 Kind() 指定任务类型，用 Worker[T].Work 执行业务。先注册 worker、准备数据库驱动与 migrations，再创建 client；生产端 Insert 或 InsertTx，执行端 Start 并按生命周期停止。当前 v0.49.0 支持 PostgreSQL，也支持 SQLite，不能沿用“River 只有 PostgreSQL”这一旧范围。[10][11]
它最有用的能力之一是 InsertTx：订单更新和任务插入复用同一数据库事务，提交后 worker 才能看到相应任务。这能封住业务数据库与任务入队的裂缝。它没有把任务后续的邮件、HTTP 或另一数据库更新一起变成事务。[12]
我们的 worker 使用同一个文本渲染函数，直接运行一次真实的 Work 方法：
go
w := &InvoiceWorker{Out: &out}
workers := river.NewWorkers()
river.AddWorker(workers, w)
if err := w.Work(context.Background(), &river.Job[InvoiceArgs]{
Args: InvoiceArgs{OrderID: "o2"},
}); err != nil { panic(err) }在完整程序中先清空缓冲区，实际输出：
text
river kind/result: invoice:render invoice:o2这验证了泛型参数、worker 注册和业务方法，不涉及数据库领取或提交。参数不合法时，示例返回 river.JobCancel(err)；暂时失败用普通 error，让 River 的任务策略处理。
v0.49.0 默认 MaxAttempts 为 25，它计的是总尝试次数，不能和 Asynq 的 MaxRetry=25 直接当作相同预算。需要配置并发、执行超时、最长重试窗口、失败保留，以及卡住任务的接管。任务唯一性可以避免特定条件下重复排队，却仍不是外部效果恰好一次。[13]
这里的优势是事务边界贴近业务，任务状态也便于数据库查询；代价是队列负载和业务数据库共享资源，必须管理索引、清理、连接池和 worker 并发。SQLite 与 PostgreSQL 的并发、连接和部署限制不一样，要按驱动文档评估，不能只替换连接字符串。v0.49.0 还调整了事务 helper 对已有事务/savepoint 的处理；调用者发生错误时应按新说明回滚事务，不能依赖旧的隐式局部回滚假设。[14]
Temporal：保存的是工作流历史，需要可重放的代码
如果任务是“生成一次缩略图”，任务队列可能已经够用。如果流程是“预留库存，等付款或超时，付款成功发货，失败释放库存，并能跨数次部署继续等待”，应用要保存的状态和恢复分支就多得多。Temporal 把这类流程写成 Workflow，由 Service 保存执行历史，Worker 执行 Workflow 和 Activity 任务。[15]
Workflow 负责步骤、等待和决策；Activity 负责可失败的外部操作。Worker 重启时可以根据历史重放 Workflow，已经记录的 Activity 结果用于恢复决策。Workflow 代码需要在相同历史下产生兼容的命令序列，因此不能随意直接读墙上时间、随机数、外部网络或用普通 goroutine 改变执行顺序。应使用 workflow.Now、workflow.Sleep、workflow.Go 等工作流 API，把数据库、HTTP 等 I/O 放进 Activity。[16]
正在绘制图示…
查看图源代码
sequenceDiagram
participant W as Workflow
participant S as Temporal 历史与调度
participant A as Activity Worker
W->>S: 调度 RenderInvoiceActivity
S->>A: 第一次尝试
A-->>S: 暂时失败
S->>S: 按重试策略等待
S->>A: 第二次尝试
A-->>S: 成功结果
S-->>W: 从历史返回结果
W->>S: 工作流完成
本文用官方 SDK 的内存测试环境运行这个场景。工作流给 Activity 设置 1 秒单次执行上限，以及最多 3 次总尝试：
go
ctx = workflow.WithActivityOptions(ctx, workflow.ActivityOptions{
StartToCloseTimeout: time.Second,
RetryPolicy: &temporal.RetryPolicy{
InitialInterval: time.Second, MaximumAttempts: 3,
},
})
var result string
err := workflow.ExecuteActivity(ctx, RenderInvoiceActivity, id).Get(ctx, &result)
return result, err测试安排 Activity 第一次返回临时错误，第二次返回成功；SDK 测试时钟跳过重试等待。实际日志中的关键结果是：
text
activity attempts=2 result=invoice:o1这是断言了两次 mock 调用后的测试日志，不是手写重试循环输出，也不是服务端故障演练。另一测试运行真实 Activity 函数，缺少订单 ID 时返回 non-retryable application error，工作流测试以错误结束。[17]
默认行为还要分对象：Activity 通常有默认重试策略，Workflow Execution 默认没有相同的自动重试策略；Workflow Task 的失败重试又是另一层。StartToCloseTimeout 限制一次 Activity 尝试，ScheduleToCloseTimeout 覆盖排队和重试的总执行窗口。长时间活动还要考虑 heartbeat 与取消。重试次数或最大时长最好按业务写出来，不依赖对“默认会重试”的模糊印象。[18]
工作流完成一次，不等于外部动作只发生一次。Activity 已经向外部服务成功请求、结果回报却丢失，它可能再次执行；外部调用仍需稳定幂等键。补偿也需要业务代码，例如发起释放库存，而非自动回滚历史世界里的所有动作。
升级 Workflow 代码还要考虑历史兼容性。直接调换两次 Activity 的顺序，旧历史重放可能不匹配；应采用 SDK 支持的版本化、补丁或 Worker 部署策略，并做 replay 测试。长流程的历史大小、Continue-As-New、payload 大小与敏感数据保留也是运维成本。[19]
真正决定选型的，是尚未交给工具的责任
| 需求 | 起点建议 | 仍需明确的责任 |
|---|---|---|
| 已有 Kafka，只需要发布与消费事件 | 先比较两个 Kafka 客户端是否覆盖所需协议 | 顺序、提交前缀、Rebalance、重复、认证 |
| 复杂 AMQP 路由和工作分发 | RabbitMQ + amqp091-go | confirm/return、ack、恢复、毒消息策略 |
| 已有 NATS，需要实时通信或可回放消息 | nats.go，明确 Core / JetStream | 离线补收、consumer 配置、存储与幂等 |
| 单步后台任务，已经用 Redis | 评估 Asynq | SQL 双写、Redis 持久化、重复副作用 |
| 任务要和业务数据库一起提交 | 评估 River 的相应驱动 | 数据库容量、worker 恢复、外部幂等 |
| 多步骤、长等待、跨部署恢复 | 评估 Temporal | Workflow 确定性、Activity 幂等、版本化与服务运维 |
本次没有同负载压测，也没有把七种工具放到真实服务端做故障对比，因此这里只给职责和 API 层面的选择依据。下一步集成验收应至少覆盖：发布结果未知、业务失败不提交、业务成功但确认失败、消费者恢复、毒消息、正常退出，以及需要时的数据库双写或工作流历史重放。工具选得对，能省去一部分状态机；剩下的责任仍需要测试说清楚。
依据与本次实际验证
所有库源码在固定版本模块下载后又用本地 go doc 和源文件核对；参考链接也指向固定 tag，滚动产品文档注明查阅日为 2026-10-13。
实际执行：go mod verify 返回 all modules verified；go run . 的 8 行输出由 Example 测试逐行核对；go test -v -count=1 ./...、go test -race -count=1 ./...、go vet ./... 均成功。两个 Kafka 确认测试、两个 AMQP 确认测试和两个 Temporal 测试都通过。日志中的预期 Activity error 是测试故障注入，不是测试套件失败。
这些测试没有服务器、消息网络或数据库事务。Asynq/River 的 worker 只向单线程内存缓冲区写文本，不是持久幂等实现，也不供多个 goroutine 共用该缓冲区。没有运行可靠性、性能、自动恢复、事务或 Redis/数据库故障切换的集成实验。
完整可复现代码与固定依赖
将以下六个文件放进同一目录：go.mod、go.sum、main.go、workflow.go、main_test.go、workflow_test.go。首次下载需要访问官方 Go 模块代理和校验服务；程序本身不连接任何业务服务：
sh
GOPROXY=https://proxy.golang.org GOSUMDB=sum.golang.org go mod download
go mod verify
go run .
go test -v -count=1 ./...
go test -race -count=1 ./...
go vet ./...go.mod
代码块 · 56 行 · 点击收起 / 展开
go.mod
module example.com/mq/tools
go 1.27.2
require (
github.com/hibiken/asynq v0.26.0
github.com/nats-io/nats.go v1.54.0
github.com/rabbitmq/amqp091-go v1.15.0
github.com/riverqueue/river v0.49.0
github.com/segmentio/kafka-go v0.4.51
github.com/stretchr/testify v1.12.1
github.com/twmb/franz-go v1.22.2
go.temporal.io/sdk v1.49.0
)
require (
github.com/cespare/xxhash/v2 v2.3.0 // indirect
github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
github.com/facebookgo/clock v0.0.0-20150410010913-600d898af40a // indirect
github.com/gogo/protobuf v1.3.2 // indirect
github.com/golang/mock v1.6.0 // indirect
github.com/google/uuid v1.6.0 // indirect
github.com/grpc-ecosystem/go-grpc-middleware/v2 v2.3.2 // indirect
github.com/grpc-ecosystem/grpc-gateway/v2 v2.22.0 // indirect
github.com/klauspost/compress v1.20.3 // indirect
github.com/nats-io/nkeys v0.4.16 // indirect
github.com/nats-io/nuid v1.0.1 // indirect
github.com/nexus-rpc/nexus-proto-annotations v0.1.0 // indirect
github.com/nexus-rpc/sdk-go v0.7.0 // indirect
github.com/pierrec/lz4/v4 v4.1.33 // indirect
github.com/redis/go-redis/v9 v9.14.1 // indirect
github.com/riverqueue/river/riverdriver v0.49.0 // indirect
github.com/riverqueue/river/rivershared v0.49.0 // indirect
github.com/riverqueue/river/rivertype v0.49.0 // indirect
github.com/robfig/cron v1.2.0 // indirect
github.com/robfig/cron/v3 v3.0.1 // indirect
github.com/spf13/cast v1.10.0 // indirect
github.com/stretchr/objx v0.5.3 // indirect
github.com/tidwall/gjson v1.19.0 // indirect
github.com/tidwall/match v1.2.0 // indirect
github.com/tidwall/pretty v1.2.1 // indirect
github.com/tidwall/sjson v1.2.5 // indirect
github.com/twmb/franz-go/pkg/kmsg v1.15.0 // indirect
go.temporal.io/api v1.63.5 // indirect
go.yaml.in/yaml/v3 v3.0.5 // indirect
golang.org/x/crypto v0.57.0 // indirect
golang.org/x/net v0.58.0 // indirect
golang.org/x/sync v0.23.0 // indirect
golang.org/x/sys v0.48.0 // indirect
golang.org/x/text v0.42.0 // indirect
golang.org/x/time v0.14.0 // indirect
google.golang.org/genproto/googleapis/api v0.0.0-20260526163538-3dc84a4a5aaa // indirect
google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa // indirect
google.golang.org/grpc v1.83.2 // indirect
google.golang.org/protobuf v1.36.11 // indirect
)go.sum
代码块 · 189 行 · 点击收起 / 展开
text
github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs=
github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c=
github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA=
github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0=
github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78=
github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc=
github.com/facebookgo/clock v0.0.0-20150410010913-600d898af40a h1:yDWHCSQ40h88yih2JAcL6Ls/kVkSE8GFACTGVnMPruw=
github.com/facebookgo/clock v0.0.0-20150410010913-600d898af40a/go.mod h1:7Ga40egUymuWXxAe151lTNnCv97MddSOVsjpPPkityA=
github.com/frankban/quicktest v1.14.6 h1:7Xjx+VpznH+oBnejlPUj8oUpdxnVs4f8XU8WnHkI4W8=
github.com/frankban/quicktest v1.14.6/go.mod h1:4ptaffx2x8+WTWXmUCuVU6aPUX1/Mz7zb5vbUoiM6w0=
github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI=
github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY=
github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag=
github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE=
github.com/gogo/protobuf v1.3.2 h1:Ov1cvc58UF3b5XjBnZv7+opcTcQFZebYjWzi34vdm4Q=
github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69NZV8Q=
github.com/golang/mock v1.6.0 h1:ErTB+efbowRARo13NNdxyJji2egdxLGQhRaY+DUumQc=
github.com/golang/mock v1.6.0/go.mod h1:p6yTPP+5HYm5mzsMV8JkE6ZKdX+/wYM6Hr+LicevLPs=
github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek=
github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps=
github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU=
github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
github.com/grpc-ecosystem/go-grpc-middleware/v2 v2.3.2 h1:sGm2vDRFUrQJO/Veii4h4zG2vvqG6uWNkBHSTqXOZk0=
github.com/grpc-ecosystem/go-grpc-middleware/v2 v2.3.2/go.mod h1:wd1YpapPLivG6nQgbf7ZkG1hhSOXDhhn4MLTknx2aAc=
github.com/grpc-ecosystem/grpc-gateway/v2 v2.22.0 h1:asbCHRVmodnJTuQ3qamDwqVOIjwqUPTYmYuemVOx+Ys=
github.com/grpc-ecosystem/grpc-gateway/v2 v2.22.0/go.mod h1:ggCgvZ2r7uOoQjOyu2Y1NhHmEPPzzuhWgcza5M1Ji1I=
github.com/hibiken/asynq v0.26.0 h1:1Zxr92MlDnb1Zt/QR5g2vSCqUS03i95lUfqx5X7/wrw=
github.com/hibiken/asynq v0.26.0/go.mod h1:Qk4e57bTnWDoyJ67VkchuV6VzSM9IQW2nPvAGuDyw58=
github.com/jackc/pgerrcode v0.0.0-20240316143900-6e2875d9b438 h1:Dj0L5fhJ9F82ZJyVOmBx6msDp/kfd1t9GRfny/mfJA0=
github.com/jackc/pgerrcode v0.0.0-20240316143900-6e2875d9b438/go.mod h1:a/s9Lp5W7n/DD0VrVoyJ00FbP2ytTPDVOivvn2bMlds=
github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM=
github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg=
github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo=
github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM=
github.com/jackc/pgx/v5 v5.11.0 h1:IzBBtyK9AHqf98cctWFifYSci2hgQR/cd56wB4p+ogg=
github.com/jackc/pgx/v5 v5.11.0/go.mod h1:mal1tBGAFfLHvZzaYh77YS/eC6IX9OWbRV1QIIM0Jn4=
github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo=
github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4=
github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8=
github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
github.com/klauspost/compress v1.20.3 h1:zWNoFqq/lSuYQRim3dUXlXeA0IqwuZGicdG5ZTfY5j4=
github.com/klauspost/compress v1.20.3/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI=
github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk=
github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
github.com/nats-io/nats.go v1.54.0 h1:vsXoOxjHp/GmPUN+EcI7uOf/uB+iAP+kEsAFNQN0yzA=
github.com/nats-io/nats.go v1.54.0/go.mod h1:y+DZoD1oBOYfZTU681eTUiUjI0vbqYGixNVFHcjHJ0k=
github.com/nats-io/nkeys v0.4.16 h1:rd5oAuLOb8mnAycB0xleuEBNS1pVVnN0fv/FF34Eypg=
github.com/nats-io/nkeys v0.4.16/go.mod h1:llLgWoI0o4z/Q57q2R1kHfmocyhGV6VG/U18Glg1Afs=
github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
github.com/nexus-rpc/nexus-proto-annotations v0.1.0 h1:2fELd+9sqUtNu6Fg//pw8YFsxOvp8vZ8hfP0nHhNI80=
github.com/nexus-rpc/nexus-proto-annotations v0.1.0/go.mod h1:n3UjF1bPCW8llR8tHvbxJ+27yPWrhpo8w/Yg1IOuY0Y=
github.com/nexus-rpc/sdk-go v0.7.0 h1:38NrfY5rLnZAiMMs2ZfCKI/CSDzdfJG+27iAgfA8bUI=
github.com/nexus-rpc/sdk-go v0.7.0/go.mod h1:FHdPfVQwRuJFZFTF0Y2GOAxCrbIBNrcPna9slkGKPYk=
github.com/pierrec/lz4/v4 v4.1.33 h1:GjG1TJ1V4IzKP8L96muuuDNpTwd7D+l2ccXrjAbe014=
github.com/pierrec/lz4/v4 v4.1.33/go.mod h1:7SE9MC2STkNtL4PIwGhjmyVwvILaGI9/COYQNBhKM/c=
github.com/rabbitmq/amqp091-go v1.15.0 h1:LEQL4/yp48/Wigt6A6XOu18RQRo8ZHtB5I/KZJn+gkw=
github.com/rabbitmq/amqp091-go v1.15.0/go.mod h1:Hy4jKW5kQART1u+JkDTF9YYOQUHXqMuhrgxOEeS7G4o=
github.com/redis/go-redis/v9 v9.14.1 h1:nDCrEiJmfOWhD76xlaw+HXT0c9hfNWeXgl0vIRYSDvQ=
github.com/redis/go-redis/v9 v9.14.1/go.mod h1:huWgSWd8mW6+m0VPhJjSSQ+d6Nh1VICQ6Q5lHuCH/Iw=
github.com/riverqueue/river v0.49.0 h1:JUCLFgregbX1Wu+bSTHCjF/WHGdrwGbtz/ddInHfeb0=
github.com/riverqueue/river v0.49.0/go.mod h1:USYb57gpMBXLQm2l7poskAalOPaf5uHjSZgp2AOoYDw=
github.com/riverqueue/river/riverdriver v0.49.0 h1:kSykNQJNB7AeG6kudjB0mThV29PvykoOBIFT8ipCHEc=
github.com/riverqueue/river/riverdriver v0.49.0/go.mod h1:rVUuX/fTF2kAiJjpPT+10Lft8n90kvLqjm6b+rFXaVE=
github.com/riverqueue/river/riverdriver/riverpgxv5 v0.49.0 h1:c7YA1plP/nrNS3SEHuKizf4wHx9OQ87Yxuhj3V0eQ50=
github.com/riverqueue/river/riverdriver/riverpgxv5 v0.49.0/go.mod h1:o7zkFstM+Fk+mnlOW2sx92mQW4JswSc6zvN191l8Isg=
github.com/riverqueue/river/rivershared v0.49.0 h1:wnCYVwftMiu85kT1JrUPKEgujKkBIWoRtSNZJTUtotY=
github.com/riverqueue/river/rivershared v0.49.0/go.mod h1:E8UzQAdDutFT8rVL1wZeNbuphmIS81TgngU7/1F4U5E=
github.com/riverqueue/river/rivertype v0.49.0 h1:3up3P2DtOqnM2yEYSC1xeK9AjrljOtIfl10yZlaF4aM=
github.com/riverqueue/river/rivertype v0.49.0/go.mod h1:XKkcRQR6zm8RR/JQa1Q2ywpj8uXQu21quPa4Lpw1Xhw=
github.com/robfig/cron v1.2.0 h1:ZjScXvvxeQ63Dbyxy76Fj3AT3Ut0aKsyd2/tl3DTMuQ=
github.com/robfig/cron v1.2.0/go.mod h1:JGuDeoQd7Z6yL4zQhZ3OPEVHB7fL6Ka6skscFHfmt2k=
github.com/robfig/cron/v3 v3.0.1 h1:WdRxkvbJztn8LMz/QEvLN5sBU+xKpSqwwUO1Pjr4qDs=
github.com/robfig/cron/v3 v3.0.1/go.mod h1:eQICP3HwyT7UooqI/z+Ov+PtYAWygg1TEWWzGIFLtro=
github.com/rogpeppe/go-internal v1.11.0 h1:cWPaGQEPrBb5/AsnsZesgZZ9yb1OQ+GOISoDNXVBh4M=
github.com/rogpeppe/go-internal v1.11.0/go.mod h1:ddIwULY96R17DhadqLgMfk9H9tvdUzkipdSkR5nkCZA=
github.com/segmentio/kafka-go v0.4.51 h1:JgDPPG75tC1rWIS2Me6MwcvXJ6f49UQ4HjAOef71Hno=
github.com/segmentio/kafka-go v0.4.51/go.mod h1:Y1gn60kzLEEaW28YshXyk2+VCUKbJ3Qr6DrnT3i4+9E=
github.com/spf13/cast v1.10.0 h1:h2x0u2shc1QuLHfxi+cTJvs30+ZAHOGRic8uyGTDWxY=
github.com/spf13/cast v1.10.0/go.mod h1:jNfB8QC9IA6ZuY2ZjDp0KtFO2LZZlg4S/7bzP6qqeHo=
github.com/stretchr/objx v0.5.3 h1:jmXUvGomnU1o3W/V5h2VEradbpJDwGrzugQQvL0POH4=
github.com/stretchr/objx v0.5.3/go.mod h1:rDQraq+vQZU7Fde9LOZLr8Tax6zZvy4kuNKF+QYS+U0=
github.com/stretchr/testify v1.12.1 h1:EuwCh5fleGS7H32xRwO3wRGT7DxrDhLAT6FF8MpWDWE=
github.com/stretchr/testify v1.12.1/go.mod h1:MDEgiDPPsNp5cuIrHPPCyornHKgEVbtFUmoNlxoYthg=
github.com/tidwall/gjson v1.14.2/go.mod h1:/wbyibRr2FHMks5tjHJ5F8dMZh3AcwJEMf5vlfC0lxk=
github.com/tidwall/gjson v1.19.0 h1:xwxm7n691Uf3u5OFjzngavjGTh55KX5q/9w9xHW88JU=
github.com/tidwall/gjson v1.19.0/go.mod h1:V37/opeE/JbLUOfH0QTXiNez2l0RUjYUhpT4szFQAfc=
github.com/tidwall/match v1.1.1/go.mod h1:eRSPERbgtNPcGhD8UCthc6PmLEQXEWd3PRB5JTxsfmM=
github.com/tidwall/match v1.2.0 h1:0pt8FlkOwjN2fPt4bIl4BoNxb98gGHN2ObFEDkrfZnM=
github.com/tidwall/match v1.2.0/go.mod h1:eRSPERbgtNPcGhD8UCthc6PmLEQXEWd3PRB5JTxsfmM=
github.com/tidwall/pretty v1.2.0/go.mod h1:ITEVvHYasfjBbM0u2Pg8T2nJnzm8xPwvNhhsoaGGjNU=
github.com/tidwall/pretty v1.2.1 h1:qjsOFOWWQl+N3RsoF5/ssm1pHmJJwhjlSbZ51I6wMl4=
github.com/tidwall/pretty v1.2.1/go.mod h1:ITEVvHYasfjBbM0u2Pg8T2nJnzm8xPwvNhhsoaGGjNU=
github.com/tidwall/sjson v1.2.5 h1:kLy8mja+1c9jlljvWTlSazM7cKDRfJuR/bOJhcY5NcY=
github.com/tidwall/sjson v1.2.5/go.mod h1:Fvgq9kS/6ociJEDnK0Fk1cpYF4FIW6ZF7LAe+6jwd28=
github.com/twmb/franz-go v1.22.2 h1:EdwIHGzD/4mgCpl/sNS0C74JYn6cL/r2KYS+8UC96Ww=
github.com/twmb/franz-go v1.22.2/go.mod h1:HzZMc00pBWNfe0hNcNK4NxnsDNyKOoXnlTBkB4SUTjI=
github.com/twmb/franz-go/pkg/kmsg v1.15.0 h1:mILdOQm4JkWojbSYVQ7KJNJ1qMdYNsLrdNOBUVKM6IA=
github.com/twmb/franz-go/pkg/kmsg v1.15.0/go.mod h1:sGkGxKexpRLBThFHJLjjywcyfkK09qB/g+X+wFWZQ/A=
github.com/xdg-go/pbkdf2 v1.0.0 h1:Su7DPu48wXMwC3bs7MCNG+z4FhcyEuz5dlvchbq0B0c=
github.com/xdg-go/pbkdf2 v1.0.0/go.mod h1:jrpuAogTd400dnrH08LKmI/xc1MbPOebTwRqcT5RDeI=
github.com/xdg-go/scram v1.1.2 h1:FHX5I5B4i4hKRVRBCFRxq1iQRej7WO3hhBuJf+UUySY=
github.com/xdg-go/scram v1.1.2/go.mod h1:RT/sEzTbU5y00aCK8UOx6R7YryM0iF1N2MOmC3kKLN4=
github.com/xdg-go/stringprep v1.0.4 h1:XLI/Ng3O1Atzq0oBs3TWm+5ZVgkq2aqdlvP9JtoZ6c8=
github.com/xdg-go/stringprep v1.0.4/go.mod h1:mPGuuIYwz7CmR2bT9j4GbQqutWS1zV24gijq1dTyGkM=
github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
github.com/yuin/goldmark v1.3.5/go.mod h1:mwnBkeHKe2W/ZEtQ+71ViKU8L12m81fl3OWwC1Zlc8k=
go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64=
go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y=
go.opentelemetry.io/otel v1.44.0 h1:JjwHmHpA4iZ3wBxluu2fbbE7j4kqlE8jXyAyPXH7HqU=
go.opentelemetry.io/otel v1.44.0/go.mod h1:BMgjTHL9WPRlRjL2oZCBTL4whCGtXch2H4BhOPIAyYc=
go.opentelemetry.io/otel/metric v1.44.0 h1:1w0gILTcHdr3YI+ixLyjemwrVnsMURbTZFrSYCdDdmc=
go.opentelemetry.io/otel/metric v1.44.0/go.mod h1:8O7hanEPBNgEMmybD3s2VBKcgWOCsA6tzHBPODAiquo=
go.opentelemetry.io/otel/sdk v1.44.0 h1:nHYwb9lK+fJPU/dnT6s7W7Z8itMWyqrnVfbheVYrZ58=
go.opentelemetry.io/otel/sdk v1.44.0/go.mod h1:Osuydd3Se74nqjAKxid74N5eC+jfEqfTegHRnq58oK0=
go.opentelemetry.io/otel/sdk/metric v1.44.0 h1:3LlKgI+VjbVsjNRFZJZAJ30WjXC5VkNRks6si09iEfI=
go.opentelemetry.io/otel/sdk/metric v1.44.0/go.mod h1:5B5pMARnXxKhltooO4xUuCBorl65a4EpnTalObqOigA=
go.opentelemetry.io/otel/trace v1.44.0 h1:jxF5CsGYCe74MCRx2X4g7WsY/VBKRqqpNvXlX/6gtIk=
go.opentelemetry.io/otel/trace v1.44.0/go.mod h1:oLl1jrMQAVo6v3GAggN+1VH9VIz9iUSvW53sW1Q8PIE=
go.temporal.io/api v1.63.5 h1:c11+kPYHkXXL3UiShPdbMD+xtvqGsbTibUA9ypmiCa4=
go.temporal.io/api v1.63.5/go.mod h1:SrlW2JMwVlDP4nRWSNznUFqnSHd+YeMDS1BkYo63HCQ=
go.temporal.io/sdk v1.49.0 h1:CtGI0BUe/SCo3eoqTwuWWtXKueii9GBVus7KrKKH1Vo=
go.temporal.io/sdk v1.49.0/go.mod h1:xP0FulN5JJSfisESUP60LlWsrKz2tLSStGjVdk8r5cc=
go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
go.yaml.in/yaml/v3 v3.0.5 h1:N6y/pJk8buWs9NY5ERU2HSMfm+IuD/OtfdAnq6kESPw=
go.yaml.in/yaml/v3 v3.0.5/go.mod h1:HVTZu1O7/Vkt2N+BFy8Zza+lnLsABggaTM2ZpNIGuKg=
golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
golang.org/x/crypto v0.57.0 h1:3ZVCjf8Ggz7zneR/EHRVx68Ctf+2pmIMP2UFhh9cC6M=
golang.org/x/crypto v0.57.0/go.mod h1:Fdz0i5U6CoizGwLda9DttjSk6qlZo25zYNtR+ycvuZA=
golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
golang.org/x/mod v0.4.2/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
golang.org/x/net v0.0.0-20200226121028-0de0cce0169b/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4/go.mod h1:p54w0d4576C0XHj96bSt6lcn1PtDYWL6XObtHCRCNQM=
golang.org/x/net v0.58.0 h1:ynWG7rqYi4ccpTEuPZ2QGWHktVEM9DMCj9yzDE0Q7To=
golang.org/x/net v0.58.0/go.mod h1:YwCddHnFlT7eLQqVprV19OnhLGtc5xOKgE0RyqgfWAU=
golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
golang.org/x/sync v0.23.0 h1:KameEIfc1IkluZyXWLn39Wd4tURc6GbCiISGiZm2bQk=
golang.org/x/sync v0.23.0/go.mod h1:sUUOizhqBxiL6pEWpqNLUiaJn1ShEbZ6BBqskPbjZm0=
golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20210330210617-4fbd30eecc44/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20210510120138-977fb7262007/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo=
golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og=
golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
golang.org/x/text v0.42.0 h1:JbOZXgfeCPU9gacVtYliJqOhD+zhrEqK4LfdpmlUZqI=
golang.org/x/text v0.42.0/go.mod h1:ojzP1Z+2QtioaF8DTtO8K5q7JWVVYwZKenzujK0Zd0E=
golang.org/x/time v0.14.0 h1:MRx4UaLrDotUKUdCIqzPC48t1Y9hANFKIRpNx+Te8PI=
golang.org/x/time v0.14.0/go.mod h1:eL/Oa2bBBK0TkX57Fyni+NgnyQQN4LitPmob2Hjnqw4=
golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
golang.org/x/tools v0.0.0-20200619180055-7c47624df98f/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE=
golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
golang.org/x/tools v0.1.1/go.mod h1:o0xws9oXOQQZyjljx8fwUC0k7L1pTE6eaCbjGeHmOkk=
golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
gonum.org/v1/gonum v0.17.0 h1:VbpOemQlsSMrYmn7T2OUvQ4dqxQXU+ouZFQsZOx50z4=
gonum.org/v1/gonum v0.17.0/go.mod h1:El3tOrEuMpv2UdMrbNlKEh9vd86bmQ6vqIcDwxEOc1E=
google.golang.org/genproto/googleapis/api v0.0.0-20260526163538-3dc84a4a5aaa h1:Kjn0N0tCrDgiAFW+lGO4JZ3ck44CehvJQMAwj9QF0G8=
google.golang.org/genproto/googleapis/api v0.0.0-20260526163538-3dc84a4a5aaa/go.mod h1:q4lMZS6kskjT5HvCPrnnypcDPVJqT/f4nfxmkE7gryY=
google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa h1:mZHHdPZl0dbGHCflZgAq/Q468DWVFcU2whhB2KAo8fk=
google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8=
google.golang.org/grpc v1.83.2 h1:EManeRomTObA0BU7I8vXgg/78uE5MJ9M8B39EX2WscU=
google.golang.org/grpc v1.83.2/go.mod h1:YPI1hK3kDked6iHvgX3tR0y+nX/qpMFKhPgFsokw1S8=
google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE=
google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco=main.go
代码块 · 149 行 · 点击收起 / 展开
go
package main
import (
"bytes"
"context"
"encoding/json"
"errors"
"fmt"
"io"
"github.com/hibiken/asynq"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
amqp "github.com/rabbitmq/amqp091-go"
"github.com/riverqueue/river"
"github.com/segmentio/kafka-go"
"github.com/twmb/franz-go/pkg/kgo"
)
type FetchCommit interface {
FetchMessage(context.Context) (kafka.Message, error)
CommitMessages(context.Context, ...kafka.Message) error
}
var _ FetchCommit = (*kafka.Reader)(nil)
// The caller must stop/retry on a work error, not skip to a later offset.
func HandleKafka(ctx context.Context, r FetchCommit, work func([]byte) error) error {
m, err := r.FetchMessage(ctx)
if err != nil {
return err
}
if err := work(m.Value); err != nil {
return err
}
return r.CommitMessages(ctx, m)
}
// Error classification / Nack policy belongs to the outer consumer loop.
func HandleAMQP(d amqp.Delivery, work func([]byte) error) error {
if err := work(d.Body); err != nil {
return err
}
return d.Ack(false)
}
type ackRecorder struct {
calls int
tag uint64
multiple bool
err error
}
func (a *ackRecorder) Ack(tag uint64, multiple bool) error {
a.calls++
a.tag = tag
a.multiple = multiple
return a.err
}
func (a *ackRecorder) Nack(uint64, bool, bool) error { return nil }
func (a *ackRecorder) Reject(uint64, bool) error { return nil }
type InvoiceArgs struct {
OrderID string `json:"order_id"`
}
func (InvoiceArgs) Kind() string { return "invoice:render" }
func render(id string) (string, error) {
if id == "" {
return "", errors.New("missing order_id")
}
return "invoice:" + id, nil
}
type InvoiceHandler struct{ Out io.Writer }
func (h InvoiceHandler) ProcessTask(ctx context.Context, t *asynq.Task) error {
if err := ctx.Err(); err != nil {
return err
}
var a InvoiceArgs
if err := json.Unmarshal(t.Payload(), &a); err != nil {
return fmt.Errorf("%w: %v", asynq.SkipRetry, err)
}
s, err := render(a.OrderID)
if err != nil {
return fmt.Errorf("%w: %v", asynq.SkipRetry, err)
}
_, err = fmt.Fprintln(h.Out, s)
return err
}
var _ asynq.Handler = InvoiceHandler{}
type InvoiceWorker struct {
river.WorkerDefaults[InvoiceArgs]
Out io.Writer
}
func (w *InvoiceWorker) Work(ctx context.Context, job *river.Job[InvoiceArgs]) error {
if err := ctx.Err(); err != nil {
return err
}
s, err := render(job.Args.OrderID)
if err != nil {
return river.JobCancel(err)
}
_, err = fmt.Fprintln(w.Out, s)
return err
}
func demo() {
c, err := kgo.NewClient(kgo.RequiredAcks(kgo.NoAck()))
if c != nil {
c.Close()
}
fmt.Println("franz invalid acks:", err)
cfg := kafka.ReaderConfig{Brokers: []string{"127.0.0.1:9092"}, Topic: "orders", GroupID: "billing"}
fmt.Println("kafka reader config:", cfg.Validate())
fmt.Println("kafka writer zero acks:", int((&kafka.Writer{}).RequiredAcks))
a := new(ackRecorder)
d := amqp.Delivery{Acknowledger: a, DeliveryTag: 7, Body: []byte("event")}
if err := HandleAMQP(d, func([]byte) error { return nil }); err != nil {
panic(err)
}
fmt.Println("amqp ack:", a.calls, a.tag, a.multiple)
m := nats.NewMsg("orders.created")
m.Header.Set(jetstream.MsgIDHeader, "e1")
jsCfg := jetstream.ConsumerConfig{Durable: "billing", AckPolicy: jetstream.AckExplicitPolicy, MaxDeliver: 3}
fmt.Println("nats message/config:", m.Subject, m.Header.Get(jetstream.MsgIDHeader), jsCfg.AckPolicy == jetstream.AckExplicitPolicy, jsCfg.MaxDeliver)
var out bytes.Buffer
h := InvoiceHandler{Out: &out}
task := asynq.NewTask("invoice:render", []byte(`{"order_id":"o1"}`), asynq.MaxRetry(3))
if err := h.ProcessTask(context.Background(), task); err != nil {
panic(err)
}
fmt.Printf("asynq result: %s", out.String())
err = h.ProcessTask(context.Background(), asynq.NewTask("invoice:render", []byte(`{}`)))
fmt.Println("asynq invalid skips retry:", errors.Is(err, asynq.SkipRetry))
out.Reset()
w := &InvoiceWorker{Out: &out}
workers := river.NewWorkers()
river.AddWorker(workers, w)
if err := w.Work(context.Background(), &river.Job[InvoiceArgs]{Args: InvoiceArgs{OrderID: "o2"}}); err != nil {
panic(err)
}
fmt.Printf("river kind/result: %s %s", (InvoiceArgs{}).Kind(), out.String())
}
func main() { demo() }workflow.go
go
package main
import (
"context"
"go.temporal.io/sdk/temporal"
"go.temporal.io/sdk/workflow"
"time"
)
func RenderInvoiceActivity(ctx context.Context, id string) (string, error) {
if err := ctx.Err(); err != nil {
return "", err
}
s, err := render(id)
if err != nil {
return "", temporal.NewNonRetryableApplicationError(err.Error(), "InvalidOrder", err)
}
return s, nil
}
func InvoiceWorkflow(ctx workflow.Context, id string) (string, error) {
ctx = workflow.WithActivityOptions(ctx, workflow.ActivityOptions{
StartToCloseTimeout: time.Second,
RetryPolicy: &temporal.RetryPolicy{InitialInterval: time.Second, MaximumAttempts: 3},
})
var result string
err := workflow.ExecuteActivity(ctx, RenderInvoiceActivity, id).Get(ctx, &result)
return result, err
}main_test.go
代码块 · 66 行 · 点击收起 / 展开
go
package main
import (
"context"
"errors"
amqp "github.com/rabbitmq/amqp091-go"
"github.com/segmentio/kafka-go"
"testing"
)
func Example() {
demo()
// Output:
// franz invalid acks: idempotency requires acks=all
// kafka reader config: <nil>
// kafka writer zero acks: 0
// amqp ack: 1 7 false
// nats message/config: orders.created e1 true 3
// asynq result: invoice:o1
// asynq invalid skips retry: true
// river kind/result: invoice:render invoice:o2
}
type fakeReader struct {
commits int
commitErr error
}
func (*fakeReader) FetchMessage(context.Context) (kafka.Message, error) {
return kafka.Message{Topic: "orders", Partition: 0, Offset: 10, Value: []byte("e1")}, nil
}
func (r *fakeReader) CommitMessages(context.Context, ...kafka.Message) error {
r.commits++
return r.commitErr
}
func TestKafkaNoCommitOnWorkFailure(t *testing.T) {
r := new(fakeReader)
err := HandleKafka(context.Background(), r, func([]byte) error { return errors.New("business failed") })
if err == nil || r.commits != 0 {
t.Fatalf("err=%v commits=%d", err, r.commits)
}
}
func TestKafkaCommitFailureIsReturned(t *testing.T) {
boom := errors.New("commit failed")
r := &fakeReader{commitErr: boom}
if err := HandleKafka(context.Background(), r, func([]byte) error { return nil }); !errors.Is(err, boom) {
t.Fatal(err)
}
}
func TestAMQPNoAckOnWorkFailure(t *testing.T) {
a := new(ackRecorder)
d := amqp.Delivery{Acknowledger: a, DeliveryTag: 7}
if err := HandleAMQP(d, func([]byte) error { return errors.New("business failed") }); err == nil {
t.Fatal("expected failure")
}
if a.calls != 0 {
t.Fatal("acked failed business")
}
}
func TestAMQPAckFailureIsReturned(t *testing.T) {
boom := errors.New("ack failed")
a := &ackRecorder{err: boom}
if err := HandleAMQP(amqp.Delivery{Acknowledger: a}, func([]byte) error { return nil }); !errors.Is(err, boom) {
t.Fatal(err)
}
}workflow_test.go
代码块 · 38 行 · 点击收起 / 展开
go
package main
import (
"errors"
"github.com/stretchr/testify/mock"
"go.temporal.io/sdk/testsuite"
"testing"
)
func TestWorkflowRetriesActivity(t *testing.T) {
var suite testsuite.WorkflowTestSuite
env := suite.NewTestWorkflowEnvironment()
env.RegisterActivity(RenderInvoiceActivity)
env.OnActivity(RenderInvoiceActivity, mock.Anything, "o1").Return("", errors.New("temporary")).Once()
env.OnActivity(RenderInvoiceActivity, mock.Anything, "o1").Return("invoice:o1", nil).Once()
env.ExecuteWorkflow(InvoiceWorkflow, "o1")
if !env.IsWorkflowCompleted() || env.GetWorkflowError() != nil {
t.Fatal(env.GetWorkflowError())
}
var result string
if err := env.GetWorkflowResult(&result); err != nil {
t.Fatal(err)
}
if result != "invoice:o1" {
t.Fatal(result)
}
env.AssertExpectations(t)
t.Log("activity attempts=2 result=invoice:o1")
}
func TestWorkflowRejectsInvalidOrder(t *testing.T) {
var suite testsuite.WorkflowTestSuite
env := suite.NewTestWorkflowEnvironment()
env.RegisterActivity(RenderInvoiceActivity)
env.ExecuteWorkflow(InvoiceWorkflow, "")
if env.GetWorkflowError() == nil {
t.Fatal("invalid order succeeded")
}
}franz-go v1.22.2 config.go：默认确认、幂等冲突、普通组自动提交、Share Group、Rebalance 选项；consumer_group.go：提交与消费生命周期。 ↩︎ ↩︎ ↩︎
kafka-go v0.4.51 reader.go：
ReadMessage、FetchMessage、CommitMessages和CommitInterval。 ↩︎ ↩︎kafka-go v0.4.51 writer.go：直接 Writer 与旧 WriterConfig 的默认差异、同步/异步生产与 Completion。 ↩︎
amqp091-go v1.15.0 API 与 delivery.go：协议、delivery tag、Ack。 ↩︎
amqp091-go v1.15.0 connection.go：
Config.Recovery为实验性、默认 nil、恢复范围。 ↩︎amqp091-go v1.15.0 channel.go，PublishWithContext：调用前检查 context 后转入 Publish 的实现。 ↩︎
nats.go v1.54.0 JetStream README、message.go、MIGRATION.md：新 API、消息 ID、确认与命名变化。 ↩︎
Asynq v0.26.0 README：任务架构与常用入口。 ↩︎
Asynq v0.26.0 client.go：MaxRetry 默认、调度、Unique；processor.go：handler 错误与 SkipRetry 分支。 ↩︎
River v0.49.0 API：JobArgs、Worker、client 生命周期。 ↩︎
River Using with SQLite：驱动、并发与部署条件。 ↩︎
River Transactional enqueueing：同事务入队。 ↩︎
River v0.49.0 client.go、rivertype/river_type.go、Job retries：默认、总尝试数与失败处理。 ↩︎
River v0.49.0 CHANGELOG：现有事务复用、savepoint 行为与升级注意事项。 ↩︎
Temporal Workflow Execution overview：持久历史与执行模型。 ↩︎
Temporal Go Workflow basics 与 Activity basics：确定性与外部操作边界。 ↩︎
Temporal Go Testing：内存测试环境、mock 与跳时钟；本地依赖固定为 SDK v1.49.0。 ↩︎
Temporal Retry policies 与 Go Activity timeouts：默认重试、尝试次数与 timeout 范围。 ↩︎
Temporal Go Workflow versioning：历史兼容与版本化。 ↩︎