nats的消息传递模型

@

What is NATS

nats是一个go语言开发的开源的、轻量、高性能的原生消息系统。nats消息由主题处理，不依赖于网络位置。它提供了应用程序或服务与底层物理网络之间的抽象层。数据被编码并作为消息，由发布者发送。消息由一个或多个订阅者接收、解码和处理。

NATS使程序可以很容易地跨不同的环境、语言、云提供商和内部系统进行通信。客户机通常通过单个URL连接到NATS系统，然后向主题订阅或发布消息。通过这种简单的设计，NATS允许程序共享通用的消息处理代码，隔离资源和相互依赖。

本文主要介绍介绍nats有哪些消息传递模型

主题式消息(Subject-Based Messaging)

从根本上说，NATS是关于发布和监听消息的。这两者在很大程度上都依赖于消息的主题。简单地说，subject就是一串字符，它们是发布者和订阅者可以用来查找彼此的名称。

主题的层次结构

.
time.us

time.us.east

time.us.east.atlanta

time.eu.east

time.eu.warsaw

通配符

NATS提供了两个通配符，可以代替点分隔的主题中的一个或多个元素。订阅者可以使用这些通配符侦听多个主题，但发布者将始终使用完全指定的主题，而不使用通配符。

匹配单个token*
匹配多个tokens>>time.us.>time.us.easttime.us.east.atlantatime.us.*time.us.east

发布订阅(Publish-Subscribe)

NATS实现了一对多通信的发布-订阅消息分发模型。发布者发送关于主题的消息，而监听该主题的任何活动订阅者将接收该消息。订阅者还可以注册对通配符主题感兴趣的内容，通配符的工作原理有点像正则表达式(但只有一点点)。这种一对多的模式有时被称为扇出(fan-out)。

example

pub
nc, _ := nats.Connect(nats.DefaultURL)

defer nc.Close()

nc.Publish("foo", []byte("Hello World!"))


sub
nc, _ := nats.Connect(nats.DefaultURL)

defer nc.Close()

nc.Subscribe("foo", func(m *nats.Msg) {

    fmt.Printf("Received a message: %s\n", string(m.Data))

})

请求应答(Request-Reply)

请求-应答是现代分布式系统中的一种常见模式。发送请求时，应用程序要么使用特定超时等待响应，要么异步接收响应。现代系统日益增加的复杂性需求，许多技术需要额外的组件来完成完整的特性集。

NATS通过其核心通信机制(发布和订阅)支持这种模式。请求在给定的主题上与应答主题一起发布，应答者侦听该主题并将响应发送到应答主题。应答主题通常是一个称为_INBOX的主题，它将被动态地定向回请求者，而不考虑任何一方的位置。

NATS的能力甚至允许多个响应，其中第一个响应被利用，而系统有效地丢弃了附加的响应。这允许一个复杂的模式有多个响应器减少响应延迟和抖动。

example

nc, _ := nats.Connect(nats.DefaultURL)

defer nc.Close()

nc.Subscribe("foo", func(m *nats.Msg) {

    nc.Publish(m.Reply, []byte("I will help you"))

})

reply, _ := nc.Request("foo", []byte("help"), 50*time.Millisecond)

fmt.Println(string(reply.Data))

队列组

NATS提供了一个称为分布式队列的内置负载平衡特性。使用队列订阅者将在一组订阅者之间平衡消息传递，这些订阅者可用于提供应用程序容错和大规模工作负载处理。

要创建队列订阅，只需要订阅者注册队列名称。具有相同队列名称的所有订阅者组成队列组。不需要任何配置。当注册主题上的消息发布时，将随机选择组中的一个成员来接收消息。尽管队列组有多个订阅者，但每个消息仅被一个订阅者使用。

NATS的一个重要特性是队列组是由应用程序及其队列订阅者定义的，而不是在服务器配置上定义的。

队列订阅者是扩展服务的理想对象。向上扩展与运行另一个应用程序一样简单，向下扩展是使用一个信号终止应用程序，该信号将耗尽正在运行的请求。这种灵活性和缺少任何配置更改使NATS成为一种优秀的服务通信技术，可以与所有平台技术一起工作。

example

nc, _ := nats.Connect(nats.DefaultURL)

defer nc.Close()

received := 0

nc.QueueSubscribe("foo", "worker_group", func(_ *nats.Msg) {

    received++

})

Acknowledgements

在具有最多一次语义的系统中，有时会丢失消息。如果您的应用程序正在执行请求-应答，那么它应该使用超时来处理任何网络或应用程序故障。在请求上设置超时并使用处理超时的代码总是一个好主意。在发布事件或数据流时，确保消息传递的一种方法是将其转换为带有确认消息(ACKs)概念的请求-应答。在NATS中，ACK可以是一个空消息，一个没有有效负载的消息。

example

nc, _ := nats.Connect(nats.DefaultURL)

defer nc.Close()

nc.Subscribe("foo", func(m *nats.Msg) {

	//nc.Publish(m.Reply, []byte("I will help you"))

	m.Respond([]byte(""))

})

reply, _ := nc.Request("foo", []byte("help"), 50*time.Millisecond)

fmt.Println("ack:", string(reply.Data))