一次性交货的 Nats Jetstream 流

积累知识，胜过积蓄金银！毕竟在Golang开发的过程中，会遇到各种各样的问题，往往都是一些细节知识点还没有掌握好而导致的，因此基础知识点的积累是很重要的。下面本文《一次性交货的 Nats Jetstream 流 》，就带大家讲解一下知识点，若是你对本文感兴趣，或者是想搞懂其中某个知识点，就请你继续往下看吧~

我想使用 Nats Jetstream 实现一个恰好一次的交付系统。文档称 Jetstream 有此选项，但没有有关其工作原理以及客户如何实现此选项的示例或详细信息。我知道在发布者端我们可以在创建Stream时设置MsgId并指定复制窗口，但是消费者端呢？

正确答案

这里是 exactly-once delivery 的文档。这有点用词不当，因为实际需要的（以及此功能提供的）是一次性处理。

正如您所指出的，它是服务器在接收已发布消息时进行重复数据删除以及已接收消息的订阅进行的双重确认调用的组合（加上重试，如果必要）。

这是一个示例（为简洁起见，省略了过多的错误处理）。启动启用 jetstream 的服务器：nats-server --js，然后运行此代码（假设 nats.go v1.16+）。

package main

import (
    "log"
    "time"

    "github.com/nats-io/nats.go"
)

func failOnErr(err error) {
    if err != nil {
        log.Fatal(err)
    }
}

func main() {
    // Connect and get the JetStream context.
    nc, _ := nats.Connect(nats.DefaultURL)
    js, _ := nc.JetStream()

    // Create a test stream.
    _, err := js.AddStream(&nats.StreamConfig{
        Name:       "test",
        Storage:    nats.MemoryStorage,
        Subjects:   []string{"test.>"},
        Duplicates: time.Minute,
    })
    failOnErr(err)

    defer js.DeleteStream("test")

    // Publish some messages with duplicates.
    js.Publish("test.1", []byte("hello"), nats.MsgId("1"))
    js.Publish("test.2", []byte("world"), nats.MsgId("2"))
    js.Publish("test.1", []byte("hello"), nats.MsgId("1"))
    js.Publish("test.1", []byte("hello"), nats.MsgId("1"))
    js.Publish("test.2", []byte("world"), nats.MsgId("2"))
    js.Publish("test.2", []byte("world"), nats.MsgId("2"))

    // Create an explicit pull consumer on the stream.
    _, err = js.AddConsumer("test", &nats.ConsumerConfig{
        Durable:       "test",
        AckPolicy:     nats.AckExplicitPolicy,
        DeliverPolicy: nats.DeliverAllPolicy,
    })
    failOnErr(err)
    defer js.DeleteConsumer("test", "test")

    // Create a subscription on the pull consumer.
    // Subject can be empty since it defaults to all subjects bound to the stream.
    sub, err := js.PullSubscribe("", "test", nats.BindStream("test"))
    failOnErr(err)

    // Only two should be delivered.
    batch, _ := sub.Fetch(10)
    log.Printf("%d messages", len(batch))

    // AckSync both to ensure the server received the ack.
    batch[0].AckSync()
    batch[1].AckSync()

    // Should be zero.
    batch, _ = sub.Fetch(10, nats.MaxWait(time.Second))
    log.Printf("%d messages", len(batch))
}

值得注意的是，如果 acksync 确实失败（可以从中返回错误），则此代码将再次重试 ack，直到收到响应。来自客户端的冗余确认是无操作的。

以上就是《一次性交货的 Nats Jetstream 流 》的详细内容，更多关于的资料请关注golang学习网公众号！