入门指南

关于 NATS

软件应用和服务需要交换数据。NATS 是一种支持此类数据交换的基础设施,数据以消息的形式进行分段传输。我们将其称为"面向消息的中间件"(message oriented middleware)。

借助 NATS,应用开发人员可以:

• 轻松构建分布式且可扩展的客户端-服务器应用。
• 以通用的方式实时存储和分发数据。这可以灵活地跨越各种环境、语言、云服务提供商以及本地部署系统实现。

What is NATS, NATS docs

• NATS 是一个使用 Go 构建的消息代理(message broker)。

嵌入式 NATS

如果您的应用使用 Go 编写,并且符合您的使用场景和部署方案,您甚至可以将 NATS 服务器嵌入到您的应用中。

Embedding NATS, NATS docs

• NATS 还有一个特别之处,即对于使用 Go 构建的应用,它支持嵌入式模式(embedded mode)。
• 也就是说,不同于消息代理的常规方式(即单独启动代理服务器,再通过客户端与该服务器通信)的结构,可以将代理本身嵌入到使用 Go 构建的应用中。

嵌入式 NATS 的优势与使用场景

• 有一个讲解得很好的Youtube 视频,因此以该视频链接代替说明。
• 即使不部署单独的消息代理服务器,也可以构建模块化单体(modular monolith)应用,从而实现关注点分离(separate of concern),并利用将 NATS 以嵌入方式植入的优点。此外,还可以实现单一二进制文件部署(single binary deployment)。
• 不仅适用于无网络平台(wasm),在离线优先(offline-first)应用中也能有效使用。

官方文档示例

 1package main
 2
 3import (
 4    "fmt"
 5    "time"
 6
 7    "github.com/nats-io/nats-server/v2/server"
 8    "github.com/nats-io/nats.go"
 9)
10
11func main() {
12    opts := &server.Options{}
13
14    // 使用选项初始化新服务器
15    ns, err := server.NewServer(opts)
16
17    if err != nil {
18        panic(err)
19    }
20
21    // 通过 goroutine 启动服务器
22    go ns.Start()
23
24    // 等待服务器准备好接受连接
25    if !ns.ReadyForConnections(4 * time.Second) {
26        panic("not ready for connection")
27    }
28
29    // 连接到服务器
30    nc, err := nats.Connect(ns.ClientURL())
31
32    if err != nil {
33        panic(err)
34    }
35
36    subject := "my-subject"
37
38    // 订阅该主题
39    nc.Subscribe(subject, func(msg *nats.Msg) {
40        // 打印消息数据
41        data := string(msg.Data)
42        fmt.Println(data)
43
44        // 关闭服务器(可选)
45        ns.Shutdown()
46    })
47
48    // 向该主题发布数据
49    nc.Publish(subject, []byte("Hello embedded NATS!"))
50
51    // 等待服务器关闭
52    ns.WaitForShutdown()
53}

• 这是 NATS 官方文档中链接的Embedded NATS 示例,但如果按照该示例代码操作,通信并不会以嵌入模式进行。

1Every 2.0s: netstat -an | grep 127.0.0.1         pravdalaptop-home.local: 02:34:20
2                                                                     in 0.017s (0)
3...
4tcp4       0      0  127.0.0.1.4222         127.0.0.1.63769        TIME_WAIT

• 通过 watch 'netstat -an | grep 127.0.0.1' 命令监控往返于 localhost(127.0.0.1) 的网络,同时使用 go run . 运行该 Go 文件,可以看到从 NATS 默认端口 4222 发起的新网络请求不断增加。

嵌入模式的正确配置

• 要按预期以嵌入模式进行通信,需要以下两个选项。

  • 客户端:必须传入 InProcessServer 选项。
  • 服务器:必须在 Server.Options 中将 DontListen 标志显式设置为 true。

• 上述内容并未在官方文档中记载,该功能的起源可以通过此 PR 了解。

  This PR adds three things:

  1. InProcessConn() function to Server which builds a net.Pipe to get a connection to the NATS server without using TCP sockets
  2. DontListen option which tells the NATS server not to listen on the usual TCP listener
  3. startupComplete channel, which is closed right before we start AcceptLoop, and readyForConnections will wait for it

  The main motivation for this is that we have an application that can run either in a monolith (single-process) mode or a polylith (multi-process) mode. We'd like to be able to use NATS for both modes for simplicity, but the monolith mode has to be able to cater for a variety of platforms where opening socket connections either doesn't make sense (mobile) or just isn't possible (WASM). These changes will allow us to use NATS entirely in-process instead.

  An accompanying PR nats-io/nats.go#774 adds support to the client side.

  This is my first PR to this project so apologies in advance if I've missed anything obvious anywhere.

  /cc @nats-io/core

嵌入模式的工作示例

 1package main
 2
 3import (
 4	"fmt"
 5	"time"
 6
 7	"github.com/nats-io/nats-server/v2/server"
 8	"github.com/nats-io/nats.go"
 9)
10
11func main() {
12	opts := &server.Options{
13		// 用于配置嵌入式 NATS 服务器
14		// 将 DonListen 设置为 true
15		DontListen: true,
16	}
17
18	// 使用选项初始化新服务器
19	ns, err := server.NewServer(opts)
20
21	if err != nil {
22		panic(err)
23	}
24
25	// 通过 goroutine 启动服务器
26	go ns.Start()
27
28	// 等待服务器准备好接受连接
29	if !ns.ReadyForConnections(10 * time.Second) {
30		panic("not ready for connection")
31	}
32
33	// 通过进程内连接连接到服务器
34	nc, err := nats.Connect(ns.ClientURL(), nats.InProcessServer(ns))
35
36	if err != nil {
37		panic(err)
38	}
39
40	subject := "my-subject"
41
42	// 订阅该主题
43	nc.Subscribe(subject, func(msg *nats.Msg) {
44		// 打印消息数据
45		data := string(msg.Data)
46		fmt.Println(data)
47
48		// 关闭服务器(可选)
49		ns.Shutdown()
50	})
51
52	// 向该主题发布数据
53	nc.Publish(subject, []byte("Hello embedded NATS!"))
54
55	// 等待服务器关闭
56	ns.WaitForShutdown()
57}
58

1Every 2.0s: netstat -an | grep 127.0.0.1         pravdalaptop-home.local: 02:37:50
2                                                                     in 0.023s (0)
3...no additional logs
4

• 现在可以看到,按预期不再产生额外的网络跳转(network hop)。

底层原理

TL;DR

• 这是在 main.go 中运行该代码时,内部各函数如何运作的时序图(sequence diagram),其要点如下。
  • 通过 DontListen: true,服务器会跳过名为 AcceptLoop 的客户端监听阶段(client listening phase)。
  • 如果客户端的 Connect 选项中启用了 InProcessServer,则会创建内存中的连接(in-memory connection),通过 net.Pipe 建立管道,并将管道的一端以 net.Conn 类型返回给客户端。
  • 客户端与服务器通过该连接进行进程内通信(in-process communication)。

服务器

AcceptLoop

1// nats-server/server/server.go
2
3// 等待客户端。
4if !opts.DontListen {
5	s.AcceptLoop(clientListenReady)
6}

• 首先,当 DontListen 为 true 时,会跳过名为 AcceptLoop 的客户端监听阶段。

 1// nats-server/server/server.go
 2
 3// AcceptLoop 为便于测试而导出。
 4func (s *Server) AcceptLoop(clr chan struct{}) {
 5	// 如果在监听器正确设置之前退出,
 6	// 确保关闭该通道。
 7	defer func() {
 8		if clr != nil {
 9			close(clr)
10		}
11	}()
12
13	if s.isShuttingDown() {
14		return
15	}
16
17	// 快照服务器选项。
18	opts := s.getOpts()
19
20	// 设置可用于关闭的状态
21	s.mu.Lock()
22	hp := net.JoinHostPort(opts.Host, strconv.Itoa(opts.Port))
23	l, e := natsListen("tcp", hp)
24	s.listenerErr = e
25	if e != nil {
26		s.mu.Unlock()
27		s.Fatalf("Error listening on port: %s, %q", hp, e)
28		return
29	}
30	s.Noticef("Listening for client connections on %s",
31		net.JoinHostPort(opts.Host, strconv.Itoa(l.Addr().(*net.TCPAddr).Port)))
32
33	// 提示已启用 TLS。
34	if opts.TLSConfig != nil {
35		s.Noticef("TLS required for client connections")
36		if opts.TLSHandshakeFirst && opts.TLSHandshakeFirstFallback == 0 {
37			s.Warnf("Clients that are not using \"TLS Handshake First\" option will fail to connect")
38		}
39	}
40
41	// 如果服务器以 RANDOM_PORT (-1) 启动,则函数开始时 opts.Port 将等于 0。
42	// 因此需要获取实际端口
43	if opts.Port == 0 {
44		// 将解析后的端口写回选项。
45		opts.Port = l.Addr().(*net.TCPAddr).Port
46	}
47
48	// 现在端口已设置(如果是随机端口),使用来自 Options 或
49	// ClientAdvertise 的值设置服务器的 info Host/Port。
50	if err := s.setInfoHostPort(); err != nil {
51		s.Fatalf("Error setting server INFO with ClientAdvertise value of %s, err=%v", opts.ClientAdvertise, err)
52		l.Close()
53		s.mu.Unlock()
54		return
55	}
56	// 记录客户端连接 URL,之后可能需要用到。
57	s.clientConnectURLs = s.getClientConnectURLs()
58	s.listener = l
59
60	go s.acceptConnections(l, "Client", func(conn net.Conn) { s.createClient(conn) },
61		func(_ error) bool {
62			if s.isLameDuckMode() {
63				// 通知不再接受新客户端
64				s.ldmCh <- true
65				// 现在等待关闭...
66				<-s.quitCh
67				return true
68			}
69			return false
70		})
71	s.mu.Unlock()
72
73	// 通知调用方已准备就绪
74	close(clr)
75	clr = nil
76}

• 供参考,AcceptLoop 函数会执行以下过程。其中 TLS 和 hostPort 等与网络通信相关的部分,在进程内通信中是不需要的,因此可以省略。

客户端

InProcessServer

 1
 2// nats-go/nats.go
 3
 4// Connect 将尝试连接到 NATS 系统。
 5// url 可以包含用户名/密码信息,例如 nats://derek:pass@localhost:4222。
 6// 同样支持以逗号分隔的数组,例如 urlA, urlB。
 7// 选项从默认值开始,但可以被覆盖。
 8// 若要连接到 NATS 服务器的 websocket 端口,请使用 `ws` 或 `wss` 协议,例如
 9// `ws://localhost:8080`。注意,websocket 协议不能与其他协议(nats/tls)混用。
10func Connect(url string, options ...Option) (*Conn, error) {
11	opts := GetDefaultOptions()
12	opts.Servers = processUrlString(url)
13	for _, opt := range options {
14		if opt != nil {
15			if err := opt(&opts); err != nil {
16				return nil, err
17			}
18		}
19	}
20	return opts.Connect()
21}

 1// nats-go/nats.go
 2
 3// Options 可用于创建自定义连接。
 4type Options struct {
 5	// Url 表示客户端将要连接的单个 NATS 服务器 url。
 6	// 如果同时设置了 Servers 选项,则它
 7	// 将成为 Servers 数组中的第一个服务器。
 8	Url string
 9
10	// InProcessServer 表示运行在
11	// 同一进程内的 NATS 服务器。如果设置了该选项,我们将尝试直接
12	// 连接到该服务器,而不是使用外部 TCP 连接。
13	InProcessServer InProcessConnProvider
14
15	//...
16}

1// nats-go/nats.go
2
3type InProcessConnProvider interface {
4	InProcessConn() (net.Conn, error)
5}

• 用于建立 nats 服务器与 nats 客户端连接的 Connect 函数,可以设置客户端 URL 和连接选项;而汇集这些选项的 Options 结构体中,存在一个类型为 InProcesConnProvider 接口的 InProcessServer 字段。

1// main.go of example code
2
3// 使用选项初始化新服务器
4ns, err := server.NewServer(opts)
5
6//...
7
8// 通过进程内连接连接到服务器
9nc, err := nats.Connect(ns.ClientURL(), nats.InProcessServer(ns))

• 在 nats 客户端进行 Connect 时,如果将 nats.InProcessServer(ns) 作为 InProcessServer 字段传入,

 1// nats-go/nats.go
 2
 3// InProcessServer 是一个 Option,它将尝试建立到运行在
 4// 进程内的 NATS 服务器的直接连接,而不是通过 TCP 拨号。
 5func InProcessServer(server InProcessConnProvider) Option {
 6	return func(o *Options) error {
 7		o.InProcessServer = server
 8		return nil
 9	}
10}

• 选项中的 InProcessServer 就会被替换为嵌入式 nats 服务器,

 1// nats-go/nats.go
 2
 3// createConn 将连接到服务器并包装相应的
 4// bufio 结构。当已存在连接时,它会正确处理。
 5func (nc *Conn) createConn() (err error) {
 6	if nc.Opts.Timeout < 0 {
 7		return ErrBadTimeout
 8	}
 9	if _, cur := nc.currentServer(); cur == nil {
10		return ErrNoServers
11	}
12
13	// 如果我们持有进程内服务器的引用,则使用它建立连接。
14	if nc.Opts.InProcessServer != nil {
15		conn, err := nc.Opts.InProcessServer.InProcessConn()
16		if err != nil {
17			return fmt.Errorf("failed to get in-process connection: %w", err)
18		}
19		nc.conn = conn
20		nc.bindToNewConn()
21		return nil
22	}
23
24	//...
25}

• 在用于创建连接的 createConn 函数中,当 InProcessServer 选项不为 nil(即有效)时,会执行选项中 InProcessServer 的 InProcesConn 方法,

 1// nats-server/server/server.go
 2
 3// InProcessConn 返回到服务器的进程内连接,
 4// 无需使用 TCP 监听器即可在同一进程内进行本地通信。
 5// 无论 DontListen 选项的状态如何,均可使用此方法。
 6func (s *Server) InProcessConn() (net.Conn, error) {
 7	pl, pr := net.Pipe()
 8	if !s.startGoRoutine(func() {
 9		s.createClientInProcess(pl)
10		s.grWG.Done()
11	}) {
12		pl.Close()
13		pr.Close()
14		return nil, fmt.Errorf("failed to create connection")
15	}
16	return pr, nil
17}

• 调用服务器中实现的 InProcessConn 并执行。
• 当 nats 的 Go 客户端 nats.go 中 nc(nats 连接)的 InProcessServer 不为 nil 时会调用该函数,创建连接(net.Conn),并将其绑定到服务器的连接上。

Go 的消费者定义接口

A type implements an interface by implementing its methods. There is no explicit declaration of intent, no "implements" keyword. Implicit interfaces decouple the definition of an interface from its implementation, which could then appear in any package without prearrangement.

Interfaces are implemented implicitly, A Tour of Go

If a type exists only to implement an interface and will never have exported methods beyond that interface, there is no need to export the type itself.

Generality, Effective Go

• 这种接口设计很好地体现了 Go 中常说的消费者定义接口(consumer defined interface)和结构化类型(structural typing,即鸭子类型 duck typing),因此想借此机会一并介绍这一主题。

 1// nats-go/nats.go
 2
 3// Options 可用于创建自定义连接。
 4type Options struct {
 5	// Url 表示客户端将要连接的单个 NATS 服务器 url。
 6	// 如果同时设置了 Servers 选项,则它
 7	// 将成为 Servers 数组中的第一个服务器。
 8	Url string
 9
10	// InProcessServer 表示运行在
11	// 同一进程内的 NATS 服务器。如果设置了该选项,我们将尝试直接
12	// 连接到该服务器,而不是使用外部 TCP 连接。
13	InProcessServer InProcessConnProvider
14
15	//...
16}

1// nats-go/nats.go
2
3type InProcessConnProvider interface {
4	InProcessConn() (net.Conn, error)
5}

• 让我们回到代码。在 nats.go 客户端中,InProcessServer 选项结构体字段被定义为仅包含 InProcessConn 方法的 InProcessConnProvider 接口。

 1// nats-server/server/server.go
 2
 3// InProcessConn 返回到服务器的进程内连接,
 4// 无需使用 TCP 监听器即可在同一进程内进行本地通信。
 5// 无论 DontListen 选项的状态如何,均可使用此方法。
 6func (s *Server) InProcessConn() (net.Conn, error) {
 7	pl, pr := net.Pipe()
 8	if !s.startGoRoutine(func() {
 9		s.createClientInProcess(pl)
10		s.grWG.Done()
11	}) {
12		pl.Close()
13		pr.Close()
14		return nil, fmt.Errorf("failed to create connection")
15	}
16	return pr, nil
17}

• 然而,传入其中的类型是 nats-server 的 Server,它除了 InProcessConn 之外还实现了许多其他功能。
• 这是因为在这种情况下,客户端所关心的只是是否提供了 InProcessConn 这一接口,其他方面并不重要。
• 因此,nats.go 客户端仅定义并使用了只包含 InProcessConn() (net.Conn, error) 功能的 InProcessConnProvider 这一消费者定义接口。

结论

• 本文简要介绍了 NATS 的嵌入模式及其工作方式,以及通过 NATS 代码可以看到的 Go 消费者定义接口。
• 希望这些信息能对出于上述目的而使用 NATS 的人们有所帮助,本文就此结束。