基于Http协议订阅发布系统设计

Posted 2020-09-26 神机小道

 

基于Http协议订阅发布系统设计

--物联网系统架构设计

 

1,订阅发布(subscriber-publisher)

     订阅发布模式最典型的应用场景就是消息系统的设计。在消息系统的架构中，消息的发送者称作(publisher)，消息的接收者称作(subscriber)，参见wikipedia: Publish–subscribe pattern。整个消息系统的架构可以用如下图1来描述:

 

图1

  由图1可知消息系统主要包括3个组件: 发布者，订阅者和消息代理(Broker)，而整个消息系统的核心即是Broker，而目前就业务能力而言Broker的实现难点主要在于它的吞吐量。拿手机消息推送举例，在当前的移动互联时代，就我们很常见的大多数app用户数基本都是百万级别以上（流行app基本是千万级别），这意味着Broker至少要能支持百万台设备的订阅，使用单台服务器做Broker显然不能解决问题。而在物联网时代，订阅者将不再只有手机，订阅者可以是任何电子设备，这种场景的级别将是手机数量的百倍。

 

2,Mqtt协议的发布订阅系统实现方案

2.1，Mqtt协议

     根据官方的定义，mqtt协议即是machine-to-machine (M2M)的连接协议，该协议就是为发布订阅模式设计的非常轻量的消息传输协议。具体参见:http://mqtt.org/

从mqtt协议定义可知，该mqtt就是为发布订阅系统而设计，并且非常轻量。

2.2，实现方案

     实现一套完整的发布订阅系统，主要就是两个组件(client和broker)一个协议规范(mqtt)。

     目前流行的开源mqtt client实现是paho(http://eclipse.org/paho); 流行的开源mqtt broker实现包括 apache apollo 和 Eclipse Mosquitto(http://eclipse.org/mosquitto), mosquitto的优点是非常轻量，使用一台树莓派(或路由器)这样的小型设备足够服务一个家庭的设备连接。

2.3， 架构设计

     发布订阅的服务系统架构非常简单，基本都遵照图1的基本架构模式。对于一个家庭的物联网应用，如果设备仅想要在局域网内访问，则broker只需要安装在(基于NanoPi或RasPi开发的)小型的设备中或者直接集成到路由器中。当然对于真正的物联网应用，我们还是希望设备可以通过互联网就可以管理和控制，所以很多broker实际应当在互联网服务器中。

2.4， Mqtt协议的订阅发布系统交互原理

     首先引用一下开源项目paho提供的python版客户端执行订阅和发布动作的demo，代码非常简短

 1 #susbscriber
 2 import paho.mqtt.client as mqtt
 3 
 4 # The callback for when the client receives a CONNACK response from the server.
 5 def on_connect(client, userdata, rc):
 6     client.subscribe("$SYS/#")
 7 
 8 # The callback for when a PUBLISH message is received from the server.
 9 def on_message(client, userdata, msg):
10     print(msg.topic+" "+str(msg.payload))
11 
12 client = mqtt.Client()
13 client.on_connect = on_connect
14 client.on_message = on_message
15 client.connect("iot.eclipse.org", 1883, 60)
16  
17 # Blocking call that processes network traffic
18 client.loop_forever()

View Code

 

     Subscriber: 从订阅者客户端代码可知，订阅者只需做2个动作（连接broker和建立循环等待的长连接）和提供2个接口函数（订阅请求函数和处理broker响应结果的函数）。基本要素无非请求连接、订阅指定topic消息、和处理响应结果，但loop_forever()是一个无限循环，这意味着客户端和borker之间保持着一个socket长连接，所以从这里可以认识到broker的瓶颈之一便是能处理多少个这样的长连接。

1 #publisher
2 import paho.mqtt.client as mqtt
3  
4 client = mqtt.Client()
5 client.connect("iot.eclipse.org")
6 client.loop_start()
7 res = mqttc.publish("$SYS/#", "HELLO")
8 client. loop_stop(force=False)

     Publisher: 从发布者客户端代码可知，发布者操作比订阅者更加简单，基本要素无非是建立连接、向broker发布指定topic消息，忽略结果响应处理过程。

     subscriber和publisher的交互逻辑本质是基于tcp协议的socket实现，对于server端的socket打开mqtt协议端口，并开启一个异步线程来持续监听端口，等待client端(subscriber和publisher )的socket发出mqtt请求，client端的subscriber的mqtt请求有些不一样，那就是subscriber的socket实际和server一直保持长连接，随时等待server那边推送过来的消息，直到连接关闭 。所以抛开细节处理问题，完全可以使用netty框架，基于mqtt协议很快的开发出一套server和client端的应用。

 

3,http协议broker设计实现

 

图2 订阅发布系统Broker设计

  http协议和mqtt协议比较：

          优点：http在互联网时代得到最广泛的应用， 充分检验了它的有效性和稳定性，充分的社区支持和成熟的开源资源可用

          缺陷：相对mqtt协议太重，对网络要求更高，直接基于http1.x无法实现发布订阅(http1.x是单工协议，需要依赖websocket、servlet3.0等技术实现双工，也可以使用http2.0，但目前支持较少)

 

 

   本文是使用servlet3.0的技术实现基于http协议的发布/订阅系统broker, 图2所示即为物联网broker系统设计架构。后台broker分成两大模块：发布中心(用户和设备)和订阅中心(用户和设备)，以及事件总线。这样的设计或许会有疑惑，为什么不直接抽象成事件的发布和订阅中心，如此不久和mqtt broker一致了么？ 的确，既然是使用http协议实现，那为什么要完全仿照mqtt协议的模式呢，而且我们要设计的实际是一个“物联网的业务系统“而不是一个“中间件“，所以如果你换了一个业务场景，你又得重新设计系统，而恰巧基于http协议servlet应用正是为业务系统提供了丰富的开源资源。

 

  下面详细解释用户发布中心和订阅中心的设计，因为在物联网的应用场景中，主要业务交互逻辑是围绕用户和设备之间做publish和subscribe.

  用户发布中心(publisher):

     在物联网场景中用户充当了核心业务的publisher，对于broker的发布中心，接收到所有的前端用户请求过来的数据都将被封装成event在broker的内部系统中由发布中心广播到订阅中心。以摩拜单车为例，app是publisher的终端，摩拜单车的核心业务逻辑就是开锁指令和一系列的交易逻辑。就开锁动作而言，发布中心收到开锁event，在publish这个event之前，针对这个event不同业务场景或许有不同的业务需求，典型场景有：该事件是否需要群发、该事件是否需要定时功能，该事件是否需要可靠发布。特别的，对于事件的可靠发布，在交易类系统中属于必备要求。拿摩拜单车来说，开锁指令发出后就会开始计时计费和扣钱，这时候就需要依赖broker在应用层面对数据做事务保证，而不能依赖基础系统服务的稳定。

  订阅中心(subscriber):

      对于订阅中心(无论是用户或设备)的设计，完全遵照table或key-value的数据结构来设计，也即是对于每一个请求，broker都将为其关联一个handler以及和其对应id标识。当事件被发布到订阅中心，订阅中心的processor便会用事件ID(或唯一标识的设备ID)去查询对应的handler，并作结果响应。由于是基于http协议，所以在具体实现时需要依赖servlet3.0或websoket技术。

 

4,领域建模

     4.1 发布中心领域建模

     发布中的核心功能是发布事件，因此Event是发布中心的核心领域对象。在图2中已经阐明，事件发布所需要实现的基本功能要素，Event设计也就主要是达到第3部分所描述功能。

图3 发布中心领域模型抽象

     

  在图3中可知，AbstractEvent即是Event的顶层的Entity抽象设计。因为发布中心可能会发布多种不同类型的Event，所以AbstractEvent必须有EventType属性来表述事件的类型。无论是那种类型的Event实际都是一个Entity，既然是Entity就意味着有自己的ID，EventId作为event的唯一标识符，需要有一个明确的说明的是EventId表示意义实际相当于topic，这就是说不是每发布一个Event就会生成一个新的EventId。例如在摩拜单车的应用中，就开锁这一类事件，对于每一辆单车，都有对应一个唯一的EventId。对于之前第3部分提到的关于事件需要实现周期、延时以及可靠发布，AbstractEvent定义了cronExpression和deliveryStatus属性，其中cron表达式可以非常简洁的描述和实现周期和延时的事件设定， 而deliveryStatus则需要使用状态来保证分布式网络环境下事件动作的事务。此外，定义GroupEvent是为了解决第3部分中提到的发布一个事件，响应多台设备。

     4.2 订阅中心领域建模

  订阅中心领域核心抽象是Handler，每一个handler对应为一个订阅http request。每一个订阅请求handler都持有其希望响应的EventId，携带的业务数据以及结果响应的回调方法。订阅者期望的是当EventId标识的event发生时，可以立刻收到对应的事件响应，也即是说订阅http request是作为一个保持长时间等待的网络连接。因此所有的handler应当有一个holder将其缓存起来管理，这就是单例模式的HandlerHolder存在的意义。对于HandlerHolder在对handler缓存策略可以有两种选择：1， 以table形式缓存；2，以map形式缓存。相较两种缓存策略各有优缺点，table形式节约存储但查找代价高(可有序存放提高速度，实际在使用Java HashMap或ConcurrentHashMap实现的保持10^5个连接时，并不会多消耗太多内存，相对于List也仅仅是多几十M的内存而已，因为map中的空桶实际存储量极小)，map形式查找快但耗存储，但无论那种形式缓存都可以通过分级缓存来提高缓存能力(例如一级缓存简要数据在系统内存，二级缓存主要数据在redis等缓存系统)。

  在图4中的状态图描述了(用户和设备)订阅中心以event驱动的handler转移流程。初始时刻，设备发起订阅CommandEvent请求，等待发布中心收到用户发过来的CommandEvent请求，此时发布中心会去判断该事件是否需要记录事件交付状态，如需要得到设备响应的OKEvent，则会去订阅中心生成对应handler。此时，响应给设备的handler将携带deliveryStatus=Waiting 标识，等待设备返回确认结果。随后，设备返回的确认响应即可通过发布中心发布OKEvent响应用户处理结果。（实际处理流程应当更复杂，因为没有考虑异常情况，如设备没有收到响应结果、备响应结果丢失等，这些都需要做一些补偿策略）

 

图4 订阅中心领域建模抽象