nats-io/nats.go
GitHub: nats-io/nats.go
NATS 云原生消息系统的官方 Go 客户端，提供发布订阅、JetStream 持久化、集群连接和多种认证机制等完整的消息通信能力。
Stars: 6693 | Forks: 813
# NATS - Go Client
一个用于 [NATS messaging system](https://nats.io) 的 [Go](http://golang.org) 客户端。
[][License-Url] [][ReportCard-Url] [][Build-Status-Url] [][GoDoc-Url] [][Coverage-Url]
**查看 [NATS by example](https://natsbyexample.com) - 不断完善的 NATS 可运行、跨客户端参考示例集合。**
## 安装
```
# 获取最新发布的 Go client：
go get github.com/nats-io/nats.go@latest
# 获取特定版本：
go get github.com/nats-io/nats.go@v1.52.0
# 请注意，NATS Server 的最新主要版本是 v2：
go get github.com/nats-io/nats-server/v2@latest
```
## 基本用法
```
import "github.com/nats-io/nats.go"
// Connect to a server
nc, _ := nats.Connect(nats.DefaultURL)
// Simple Publisher
nc.Publish("foo", []byte("Hello World"))
// Simple Async Subscriber
nc.Subscribe("foo", func(m *nats.Msg) {
fmt.Printf("Received a message: %s\n", string(m.Data))
})
// Responding to a request message
nc.Subscribe("request", func(m *nats.Msg) {
m.Respond([]byte("answer is 42"))
})
// Simple Sync Subscriber
sub, err := nc.SubscribeSync("foo")
m, err := sub.NextMsg(timeout)
// Channel Subscriber
ch := make(chan *nats.Msg, 64)
sub, err := nc.ChanSubscribe("foo", ch)
msg := <- ch
// Unsubscribe
sub.Unsubscribe()
// Drain
sub.Drain()
// Requests
msg, err := nc.Request("help", []byte("help me"), 10*time.Millisecond)
// Replies
nc.Subscribe("help", func(m *nats.Msg) {
nc.Publish(m.Reply, []byte("I can help!"))
})
// Drain connection (Preferred for responders)
// Close() not needed if this is called.
nc.Drain()
// Close connection
nc.Close()
```
## JetStream
[](https://pkg.go.dev/github.com/nats-io/nats.go/jetstream)
JetStream 是 NATS 的内置持久化系统。`nats.go` 提供了一个内置的
API，既可以管理 JetStream 资产，也可以发布/消费
持久化消息。
### 基本用法
```
// connect to nats server
nc, _ := nats.Connect(nats.DefaultURL)
// create jetstream context from nats connection
js, _ := jetstream.New(nc)
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
// get existing stream handle
stream, _ := js.Stream(ctx, "foo")
// retrieve consumer handle from a stream
cons, _ := stream.Consumer(ctx, "cons")
// consume messages from the consumer in callback
cc, _ := cons.Consume(func(msg jetstream.Msg) {
fmt.Println("Received jetstream message: ", string(msg.Data()))
msg.Ack()
})
defer cc.Stop()
```
要了解有关 `nats.go` JetStream API 的更多信息，请访问
[`jetstream/README.md`](jetstream/README.md)
## Service API
Service API (`micro`) 允许您[轻松构建 NATS 服务](micro/README.md)。
Service API 目前处于 beta 发布阶段。
## 新的身份验证（Nkeys 和 User Credentials）
这要求服务器版本 >= 2.0.0
NATS 服务器具有新的安全和身份验证机制，可使用 user credentials 和 Nkeys 进行身份验证。
最简单的形式是使用辅助方法 UserCredentials(credsFilepath)。
```
nc, err := nats.Connect(url, nats.UserCredentials("user.creds"))
```
该辅助方法会创建两个回调 handler，用于出示用户 JWT 并对来自服务器的 nonce challenge 进行签名。
核心客户端库永远不会直接访问您的私钥，而只是执行回调以对服务器 challenge 进行签名。
该辅助程序会在每次连接或重连时加载、擦除并清除其所使用的内存。
该辅助程序还可以接受两个条目，一个用于 JWT，另一个用于 NKey seed 文件。
```
nc, err := nats.Connect(url, nats.UserCredentials("user.jwt", "user.nk"))
```
您也可以直接设置回调 handler，并自行管理 challenge 签名。
```
nc, err := nats.Connect(url, nats.UserJWT(jwtCB, sigCB))
```
也支持 Bare Nkeys。nkey seed 应位于只读文件中，例如 seed.txt
```
> cat seed.txt
# 这是我的 seed nkey！
SUAGMJH5XLGZKQQWAWKRZJIGMOU4HPFUYLXJMXOO5NLFEO2OOQJ5LPRDPM
```
这是一个辅助函数，它将加载、解码并为服务器 nonce 执行正确的签名。
它会在调用之间清除内存。
您可以选择使用低级选项，并自行提供公钥和签名回调。
```
opt, err := nats.NkeyOptionFromSeed("seed.txt")
nc, err := nats.Connect(serverUrl, opt)
// Direct
nc, err := nats.Connect(serverUrl, nats.Nkey(pubNkey, sigCB))
```
## TLS
```
// tls as a scheme will enable secure connections by default. This will also verify the server name.
nc, err := nats.Connect("tls://nats.demo.io:4443")
// If you are using a self-signed certificate, you need to have a tls.Config with RootCAs setup.
// We provide a helper method to make this case easier.
nc, err = nats.Connect("tls://localhost:4443", nats.RootCAs("./configs/certs/ca.pem"))
// If the server requires client certificate, there is a helper function for that too:
cert := nats.ClientCert("./configs/certs/client-cert.pem", "./configs/certs/client-key.pem")
nc, err = nats.Connect("tls://localhost:4443", cert)
// You can also supply a complete tls.Config
certFile := "./configs/certs/client-cert.pem"
keyFile := "./configs/certs/client-key.pem"
cert, err := tls.LoadX509KeyPair(certFile, keyFile)
if err != nil {
t.Fatalf("error parsing X509 certificate/key pair: %v", err)
}
config := &tls.Config{
ServerName: opts.Host,
Certificates: []tls.Certificate{cert},
RootCAs: pool,
MinVersion: tls.VersionTLS12,
}
nc, err = nats.Connect("nats://localhost:4443", nats.Secure(config))
if err != nil {
t.Fatalf("Got an error on Connect with Secure Options: %+v\n", err)
}
```
## 通配符订阅
```
// "*" matches any token, at any level of the subject.
nc.Subscribe("foo.*.baz", func(m *Msg) {
fmt.Printf("Msg received on [%s] : %s\n", m.Subject, string(m.Data))
})
nc.Subscribe("foo.bar.*", func(m *Msg) {
fmt.Printf("Msg received on [%s] : %s\n", m.Subject, string(m.Data))
})
// ">" matches any length of the tail of a subject, and can only be the last token
// E.g. 'foo.>' will match 'foo.bar', 'foo.bar.baz', 'foo.foo.bar.bax.22'
nc.Subscribe("foo.>", func(m *Msg) {
fmt.Printf("Msg received on [%s] : %s\n", m.Subject, string(m.Data))
})
// Matches all of the above
nc.Publish("foo.bar.baz", []byte("Hello World"))
```
## 队列组
```
// All subscriptions with the same queue name will form a queue group.
// Each message will be delivered to only one subscriber per queue group,
// using queuing semantics. You can have as many queue groups as you wish.
// Normal subscribers will continue to work as expected.
nc.QueueSubscribe("foo", "job_workers", func(_ *Msg) {
received += 1
})
```
## 高级用法
```
// Normally, the library will return an error when trying to connect and
// there is no server running. The RetryOnFailedConnect option will set
// the connection in reconnecting state if it failed to connect right away.
nc, err := nats.Connect(nats.DefaultURL,
nats.RetryOnFailedConnect(true),
nats.MaxReconnects(10),
nats.ReconnectWait(time.Second),
nats.ReconnectHandler(func(_ *nats.Conn) {
// Note that this will be invoked for the first asynchronous connect.
}))
if err != nil {
// Should not return an error even if it can't connect, but you still
// need to check in case there are some configuration errors.
}
// Flush connection to server, returns when all messages have been processed.
nc.Flush()
fmt.Println("All clear!")
// FlushTimeout specifies a timeout value as well.
err := nc.FlushTimeout(1*time.Second)
if err != nil {
fmt.Println("Flushed timed out!")
} else {
fmt.Println("All clear!")
}
// Auto-unsubscribe after MAX_WANTED messages received
const MAX_WANTED = 10
sub, err := nc.Subscribe("foo")
sub.AutoUnsubscribe(MAX_WANTED)
// Multiple connections
nc1 := nats.Connect("nats://host1:4222")
nc2 := nats.Connect("nats://host2:4222")
nc1.Subscribe("foo", func(m *Msg) {
fmt.Printf("Received a message: %s\n", string(m.Data))
})
nc2.Publish("foo", []byte("Hello World!"))
```
## 集群用法
```
var servers = "nats://localhost:1222, nats://localhost:1223, nats://localhost:1224"
nc, err := nats.Connect(servers)
// Optionally set ReconnectWait and MaxReconnect attempts.
// This example means 10 seconds total per backend.
nc, err = nats.Connect(servers, nats.MaxReconnects(5), nats.ReconnectWait(2 * time.Second))
// You can also add some jitter for the reconnection.
// This call will add up to 500 milliseconds for non TLS connections and 2 seconds for TLS connections.
// If not specified, the library defaults to 100 milliseconds and 1 second, respectively.
nc, err = nats.Connect(servers, nats.ReconnectJitter(500*time.Millisecond, 2*time.Second))
// You can also specify a custom reconnect delay handler. If set, the library will invoke it when it has tried
// all URLs in its list. The value returned will be used as the total sleep time, so add your own jitter.
// The library will pass the number of times it went through the whole list.
nc, err = nats.Connect(servers, nats.CustomReconnectDelay(func(attempts int) time.Duration {
return someBackoffFunction(attempts)
}))
// Optionally disable randomization of the server pool
nc, err = nats.Connect(servers, nats.DontRandomize())
// Setup callbacks to be notified on disconnects, reconnects and connection closed.
nc, err = nats.Connect(servers,
nats.DisconnectErrHandler(func(nc *nats.Conn, err error) {
fmt.Printf("Got disconnected! Reason: %q\n", err)
}),
nats.ReconnectHandler(func(nc *nats.Conn) {
fmt.Printf("Got reconnected to %v!\n", nc.ConnectedUrl())
}),
nats.ClosedHandler(func(nc *nats.Conn) {
fmt.Printf("Connection closed. Reason: %q\n", nc.LastError())
})
)
// When connecting to a mesh of servers with auto-discovery capabilities,
// you may need to provide a username/password or token in order to connect
// to any server in that mesh when authentication is required.
// Instead of providing the credentials in the initial URL, you will use
// new option setters:
nc, err = nats.Connect("nats://localhost:4222", nats.UserInfo("foo", "bar"))
// For token based authentication:
nc, err = nats.Connect("nats://localhost:4222", nats.Token("S3cretT0ken"))
// You can even pass the two at the same time in case one of the servers
// in the mesh requires token instead of user name and password.
nc, err = nats.Connect("nats://localhost:4222",
nats.UserInfo("foo", "bar"),
nats.Token("S3cretT0ken"))
// Note that if credentials are specified in the initial URLs, they take
// precedence on the credentials specified through the options.
// For instance, in the connect call below, the client library will use
// the user "my" and password "pwd" to connect to localhost:4222, however,
// it will use username "foo" and password "bar" when (re)connecting to
// a different server URL that it got as part of the auto-discovery.
nc, err = nats.Connect("nats://my:pwd@localhost:4222", nats.UserInfo("foo", "bar"))
```
## Context 支持（+Go 1.7）
```
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel()
nc, err := nats.Connect(nats.DefaultURL)
// Request with context
msg, err := nc.RequestWithContext(ctx, "foo", []byte("bar"))
// Synchronous subscriber with context
sub, err := nc.SubscribeSync("foo")
msg, err := sub.NextMsgWithContext(ctx)
```
## 向后兼容性
在开发 nats.go 时，我们致力于保持向后兼容性，并确保为所有用户提供稳定可靠的体验。通常，我们遵循标准的 Go 兼容性准则。
但是，有必要澄清我们对某些类型更改的立场：
- **扩展结构体：**
向 struct 添加新字段不被视为破坏性更改。
- **向导出的 interface 添加方法：**
在此项目的背景下，使用新方法扩展公共 interface 也不被视为破坏性更改。需要注意的是，不会向 interface 添加任何非导出方法，从而允许用户自行实现它们。
此外，此库始终支持至少 2 个最新的 Go 次要版本。例如，如果最新的 Go 版本是 1.22，则该库将支持 Go 1.21 和 1.22。
## 许可证
除非另有说明，NATS 源文件均按 LICENSE 文件中的 Apache Version 2.0 许可证分发。
[](https://app.fossa.io/projects/git%2Bgithub.com%2Fnats-io%2Fgo-nats?ref=badge_large)
标签：EVTX分析, Golang, NATS, 分布式通信, 安全编程, 客户端组件, 底层编程, 日志审计, 消息队列