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))