使用NATS消息中间件实现云边协同
作者：互联网
1.NATS介绍
NATS(Message bus): 从CloudFoundry的总架构图看，位于各模块中心位置的是一个叫nats的组件。NATS是由CloudFoundry的架构师Derek开发的一个开源的、轻量级、高性能的，支持发布、订阅机制的分布式消息队列系统。它的核心基于EventMachine开发，代码量不多，可以下载下来慢慢研究。其核心原理就是基于消息发布订阅机制。每个台服务 器上的每个模块会根据自己的消息类别，向MessageBus发布多个消息主题；而同时也向自己需要交互的模块，按照需要的信息内容的消息主题订阅消息。 NATS原来是使用Ruby编写，可以实现每秒150k消息，后来使用Go语言重写，能够达到每秒8-11百万个消息，整个程序很小只有3M Docker image，它不支持持久化消息，如果你离线，你就不能获得消息。
NATS适合云基础设施的消息通信系统、IoT设备消息通信和微服务架构。Apcera团队负责维护NATS服务器（Golang语言开发）和客户端（包括Go、Python、Ruby、Node.js、Elixir、Java、Nginx、C和C#），开源社区也贡献了一些客户端库，包括Rust、PHP、Lua等语言的库。目前已经采用了NATS系统的公司有：爱立信、HTC、百度、西门子、VMware。
市面上常见到的和Nats功能类似的消息通信系统有：
ActiveMQ（Java编写）、KafKa（Scala编写）、RabbitMq（Erlang编写）、Nats（之前是Ruby编写现已修改为Go）、Redis（C语言编写）、Kestrel（Scala编写不常用）、NSQ（Go语言编写），这些消息通信系统在Broker吞吐量方面的比较：(注：来自作者Derek Collison 对不同版本的消息系统进行的比较)
按照其官网的说法，NATS是一个开源的、高性能的、简洁的、灵活的 适用于现代的可靠灵活的云和分布式系统的中枢系统。 说的很玄乎，实际上就是一个分布式的消息队列系统，支持PubSub/ReqRsp 模型。其最初由Apcera领导开发，并实现了Ruby版本的服务器和客户端，其主要作者Derek Collison自称做了20多年的MQ，并经历过TIBOC、Rendezvous、EMC公司，这里有他自己的reddit回答。
根据github里面ruby-nats的日志显示在11年Derek实现了Ruby版本的NATS服务器以及对应的客户端。然后在12年末，姑且认为是13年Derek又用Golang将服务器重写了一遍，并最终发现其效果更好，于是现在慢慢将Ruby版本的服务器淘汰了，现在官网也只维护一个Golang版本的服务器，也就是我们这里的gnatsd。
2.NATS发布/订阅机制
概念
发布/订阅（Publish/subscribe 或pub/sub）是一种消息范式，消息的发送者（发布者）不是计划发送其消息给特定的接收者（订阅者）。而是发布的消息分为不同的类别，而不需要知道什么样的订阅者订阅。订阅者对一个或多个类别表达兴趣，于是只接收感兴趣的消息，而不需要知道什么样的发布者发布的消息。这种发布者和订阅者的解耦可以允许更好的可扩展性和更为动态的网络拓扑.
发布/订阅是消息队列范式的兄弟，通常是更大的消息导向的中间件的系统的一部分。大多数消息系统在应用程序接口（API）中同时支持消息队列模型和发布/订阅模型，例如Java消息服务（JMS）。
现实中，并不是所有请求都期待答复，而不期待答复，自然就没有了状态。广播听过吧？收音机用过吧？就这个意思。
发布/订阅模式定义了一种一对多的依赖关系，让多个订阅者对象同时监听某一个主题对象。这个主题对象在自身状态变化时，会通知所有订阅者对象，使它们能够自动更新自己的状态。
消息过滤
在发布/订阅模型中，订阅者通常接收所有发布的消息的一个子集。选择接受和处理的消息的过程被称作过滤。有两种常用的过滤形式：基于主题的和基于内容的。
在 基于主题 的系统中，消息被发布到主题或命名通道上。订阅者将收到其订阅的主题上的所有消息，并且所有订阅同一主题的订阅者将接收到同样的消息。发布者负责定义订阅者所订阅的消息类别。
在 基于内容 的系统中，订阅者定义其感兴趣的消息的条件，只有当消息的属性或内容满足订阅者定义的条件时，消息才会被投递到该订阅者。订阅者需要负责对消息进行分类。
一些系统支持两者的混合：发布者发布消息到主题上，而订阅者将基于内容的订阅注册到一个或多个主题上。
拓扑
在许多发布/订阅系统中，发布者发布消息到一个中间的 消息代理，然后订阅者向该消息代理注册订阅，由消息代理来进行过滤。消息代理通常执行 存储转发 的功能将消息从发布者发送到订阅者。
使用场景
很多项目中都有消息分发或者事件通知机制，尤其是模块化程度高的项目。
比如：在你的系统中，很多模块都对 新建用户 感兴趣。权限模块希望给新用户设置默认权限，报表模块希望重新生成当月的报表，邮件系统希望给用户发送激活邮件…诸如此类的代码都写到新建用户的业务逻辑后面，会加大耦合度，降低可维护性，并且对于每个模块都是一个独立系统的情况，这种方式更是不可取。
对于简单的情形，观察者模式 The Observer Pattern 就足够了。如果系统中有很多地方都需要收发消息，那么它就不适用了。否则会造成类数量的膨胀，增加类的复杂性，这时候就需要一种更集中的机制来处理这些业务逻辑。
特点
一个订阅者可以订阅多个发布者
消息是会到达所有订阅者处，订阅者根据 filter 丢掉自己不需要的消息(filter 是在订阅端起作用的)
每个订阅者都会接收到每条消息的一个副本
基于推送 push，其中消息自动地向订阅者广播，它们无须请求或轮询主题来获得新消息，发布/订阅模式内部，有多种不同类型的订阅者。
非持久订阅者是临时订阅类型，它们只是在主动侦听主题时才接收消息。
持久订阅者将接收到发布的每条消息的一个副本，即便在发布消息，它们处于"离线"状态时也是如此。
另外还有动态持久订阅者和受管的持久订阅者等类型。
优势
降低了模块间的耦合度：发布者与订阅者松散地耦合，并且不需要知道对方的存在。相关操作都集中在 Publisher 中。
可扩展性强：系统复杂后，可以把消息订阅和分发机制单独作为一个模块来实现，增加新特性以满足需求
缺陷
与其说缺陷，不如说它设计本身就有如下特点。但不管怎么说，这种模式在逻辑上不可靠的。主要体现在：
发布者不知道订阅者是否收到发布的消息
订阅者不知道自己是否收到了发布者发出的所有消息
发送者不能获知订阅者的执行情况
没人知道订阅者何时开始收到消息
已存在的案例
ctiveMQ（Java编写）、KafKa（Scala编写）、RabbitMq（Ruby编写）、Nats（之前是Ruby编写现已修改为Go）
3. NATS—消息通信模型
消息通信模型
NATS的消息通信是这样的：应用程序的数据被编码为一条消息，并通过发布者发送出去；订阅者接收到消息，进行解码，再处理。订阅者处理NATS消息可以是同步的或异步的。
- 异步处理
异步处理使用回调消息句柄处理消息，当有消息到来时，已注册的回调句柄接收并控制处理消息。整个过程客户端不会被阻塞，可以同步执行其它任务。异步处理可以采用多线程调度的设计。 - 同步处理
同步处理需要应用程序显示调用方法来处理到来的消息。这种显示调用是阻塞式的调用，会暂停任务直到消息可用。如果没有可用的消息，消息处理阻塞的周期由客户端设置。同步处理通常用于服务器等待并处理传入的请求消息，并发送响应给客户端。
NATS支持以下三种消息通信模型：
1. 发布/订阅模型
NATS的发布/订阅通信模型是一对多的消息通信。发布者在一个主题上发送消息，任何注册（订阅）了此主题的客户端都可以接收到该主题的消息。订阅者可以使用主题通配符订阅感兴趣的主题。
对于订阅者，可以选择异步处理或同步处理接收到的消息。如果异步处理消息，消息交付给订阅者的消息句柄。如果客户端没有句柄，那么该消息通信是同步的，那么客户端可能会被阻塞，直到它处理了当前消息。
服务质量（QoS）
至多发送一次 (TCP reliability)：如果客户端没有注册某个主题（或者客户端不在线），那么该主题发布消息时，客户端不会收到该消息。NATS系统是一种“发送后不管”的消息通信系统，故如果需要高级服务，可以选择"NATS Streaming" 或 在客户端开发相应的功能
至少发送一次(NATS Streaming) ：一些使用场景需要更高级更严格的发送保证，这些应用依赖底层传送消息，不论网络是否中断或订阅者是否在线，都要确保订阅者可以收到消息
2. 请求/响应模型
NATS支持两种请求-响应消息通信：P2P（点对点）和O2M（一对多）。P2P最快、响应也最先。而对于O2M，需要设置请求者可以接收到的响应数量界限（默认只能收到一条来自订阅者的响应——随机）
在请求-响应模式，发布请求操作会发布一个带预期响应的消息到Reply主题。
请求创建了一个收件箱，并在收件箱执行调用，并进行响应和返回
多个订阅者(reply 例子)订阅了同一个 主题，请求者向该主题发送一个请求，默认只收到一个订阅者的响应(随机)
事实上，NATS协议中并没有定义 “请求” 或 "响应"方法，它是通过 SUB/PUB变相实现的：请求者先 通过SUB创建一个收件箱，然后发送一个带 reply-to 的PUB，响应者收到PUB消息后，向 reply-to 发送 响应消息，从而实现 请求/响应。reply-to和收件箱都是一个 subject，前者是后者的子集（O2M的情况）
3. 队列模型
NATS支持P2P消息通信的队列。要创建一个消息队列，订阅者需注册一个队列名。所有的订阅者用同一个队列名，形成一个队列组。当消息发送到主题后，队列组会自动选择一个成员接收消息。尽管队列组有多个订阅者，但每条消息只能被组中的一个订阅者接收。
队列的订阅者可以是异步的，这意味着消息句柄以回调方式处理交付的消息。同步队列订阅者必须建立处理消息的逻辑
NATS支持P2P消息通信的队列。要创建一个消息队列，订阅者需注册一个队列名。所有的订阅者用同一个队列名，形成一个队列组。当消息发送到主题后，队列组会自动选择一个成员接收消息。尽管队列组有多个订阅者，但每条消息只能被组中的一个订阅者接收。
队列的订阅者可以是异步的，这意味着消息句柄以回调方式处理交付的消息。异步队列订阅者必须建立处理消息的逻辑。
队列模型一般常用于数据队列使用，例如：从网页上采集的数据经过处理直接写入到该队列，接收端一方可以起多个线程同时读取其中的一个队列，其中某些数据被一个线程消费了，其他线程就看不到了，这种方式为了解决采集量巨大的情况下，后端服务可以动态调整并发数来消费这些数据。说白了就一点，上游生产数据太快，下游消费可能处理不过来，中间进行缓冲，下游就可以根据实际情况进行动态调整达到动态平衡。
NATS特性
NATS提供了以下独特的功能：
1）纯发布/订阅
永远不假定有接收者
总是在线
2）集群模式的服务器
NATS服务器可以集群；
发布式的队列可以跨域集群；
集群感知的客户端
3）订阅者的自动修剪
要支持可伸缩性，NATS提供了客户端连接的自动修剪功能；
如果某个客户端APP处理消息很慢，NATS会自动关闭此客户端的连接；
如果某个客户端在ping-pong时间间隔内未做响应，服务器会自动关闭此连接；
客户端实现重连逻辑
4）基于文本的协议
开发上手比较容易；
不影响服务器的性能；
可以直接用Telnet连接服务器
5）多种 QoS
至多发送一次（TCP level reliability）—NATS立即向符合条件的订阅者发送消息，并不存留消息
https://www.zhihu.com/question/49596182
至少发送一次（via NATS Streaming）— 如果匹配的订阅者一时不在线，Message 将被存储直到它被传送给订阅者，并得到订阅者确认。除非 该消息超时或存储空间耗尽
6）持久性订阅（via NATS Streaming）
服务端维护 持久性订阅者的 订阅推送状态，这样，持久性订阅者就可以知道它们在上一次会话中是在哪儿断开的
7）Event 流服务(via NATS Streaming)
根据时间戳、序列号或相对位差，消息被持久化存储在 内存、文件或其它二级存储设备中
8）缓存 最新一个或第一个值 (via NATS Streaming)
订阅者连接上服务器以后，先向订阅者推送最近一次的publish消息
4.NATS—协议详解（nats-protocol）
参考：https://www.cnblogs.com/yorkyang/p/8393080.html
5.NATS使用
这个nats云边协同基于openstack keystone去做的，实现了一个发布nats client处理，另外一个服务器订阅同步数据，此demo只作为参考，因为优化方面还没来得及做，希望能给到小伙伴们更多的帮助。
nats client
def dispose_nats(response):
'''
response : flask response.
send request data to nats
'''
print response.status_code
if response.status_code == 200 or response.status_code == 201 or response.status_code == 204:
response_body = response.get_data()
response_body = json.loads(response_body)
request_body = request.get_json()
if request.method == 'POST':
if request.url == 'http://10.121.11.163:5000/v3/users/accounts' or request.url == 'http://10.121.11.163:5000/v3/policies':
pass
else:
data_id = {'id': ''}
for data in response_body.values():
data_id['id'] = data.get('id', '')
if data_id.get('id') == '':
LOG.warning('The data ID is obtained when publishing the information')
for body in request_body.values():
body['id'] = data_id.get('id', '')
datas = {'url': request.url, 'body': request_body,
'method': request.method,
'headers': {'Content-Type': request.headers['Content-Type'],
'X-Auth-Token': request.headers['X-Auth-Token']}
}
tornado.ioloop.IOLoop.instance().run_sync(partial(send_nats, datas))
elif request.method == 'PATCH':
datas = {'url': request.url, 'body': request_body,
'method': request.method,
'headers': {'Content-Type': request.headers['Content-Type'],
'X-Auth-Token': request.headers['X-Auth-Token']}
}
tornado.ioloop.IOLoop.instance().run_sync(partial(send_nats, datas))
elif request.method == 'DELETE':
datas = {'url': request.url, 'headers': {
'X-Auth-Token': request.headers['X-Auth-Token']},
'method': request.method}
tornado.ioloop.IOLoop.instance().run_sync(partial(send_nats, datas))
return response
@tornado.gen.coroutine
def send_nats(datas):
"""
nats发布信息，datas是从flask中response之前拼接的数据，做为订阅那边进行数据同步，这边指定了
一下nats的连接地址，如果是默认的，请注意改动，nats使用tornado异步进行处理，笔者这边占时没有使用nats client回应的机制，可能会存在发布过程中订阅者消息超时未处理，导致未同步成功，建议订阅者收到信息后给到nats client回应，防止数据不同步
"""
nc = NATS()
yield nc.connect(servers=["nats://10.121.121.241:4222"], connect_timeout=10)
yield nc.publish("synchronization", json.dumps(datas).encode())
nats server
笔者使用的是http发送请求来实现同步，因为nats server默认不会启动，所以把订阅放到了docker运行，一些抛错信息使用了logging存放到了指定的log文件，方便查看过程中是否出现同步异常，最后对main函数做了一个优化，如有问题请在最下方评论或私信联系我，欢迎吐槽！
import datetime
import json
import logging
import requests
import sys
import getopt
import tornado.gen
import tornado.ioloop
from nats.io import Client as NATS
logging.basicConfig(level=logging.DEBUG,
format='%(asctime)s %(filename)s[line:%(lineno)d] %(levelname)s %(message)s',
filename='sub-nats.log',
filemode='a')
@tornado.gen.coroutine
def send():
nc = NATS()
ip = 'http://10.121.121.241:5000'
yield nc.connect("nats://10.121.121.241:4222", connect_timeout=10)
@tornado.gen.coroutine
def subscriber(msg):
data = json.loads(msg.data.decode())
url = data['url'].split('5000')
urls = ip + url[1]
if data['method'] == 'POST':
try:
res = requests.post(url=urls, data=json.dumps(data['body']), headers=data['headers'])
logging.info(res.json())
logging.info(res.status_code)
except BaseException as e:
logging.error(str(e))
elif data['method'] == 'PATCH':
try:
res = requests.post(url=urls, data=json.dumps(data['body']), headers=data['headers'])
logging.info(res.json())
logging.info(res.status_code)
except BaseException as e:
logging.error(str(e))
elif data['method'] == 'DELETE':
try:
res = requests.delete(url=urls, headers=data['headers'])
logging.info(res.json())
logging.info(res.status_code)
except BaseException as e:
logging.error(str(e))
yield nc.flush()
yield nc.subscribe("synchronization", "", subscriber)
while 1:
yield tornado.gen.sleep(100)
# Sends a PING and wait for a PONG from the server, up to the given timeout.
# This gives guarantee that the server has processed above message.
class Usage(Exception):
def __init__(self, msg):
self.msg = msg
def main(argv=None):
if argv is None:
argv = sys.argv
try:
try:
opts, args = getopt.getopt(argv[1:], "h", ["help"])
tornado.ioloop.IOLoop.instance().run_sync(send)
except getopt.error, msg:
raise Usage(msg)
# more code, unchanged
except Usage, err:
print >> sys.stderr, err.msg
print >> sys.stderr, "for help use --help"
return 2
if __name__ == '__main__':
sys.exit(main())
标签：订阅,队列,nats,NATS,云边,消息,消息中间件,data 来源： https://blog.csdn.net/weixin_49319422/article/details/119103235