hpfeeds协议解析

电玩女神 2022-05-08 13:12 346阅读 0赞

## 一. hpfeeds协议简介 ##

hpfeeds是一个轻量级的验证发布-订阅协议(authenticated publish-subscribe protocol)。

**发布-订阅协议：**发布/订阅协议定义了一种一对多的依赖关系，让多个订阅者对象同时监听某一个主题对象。这个主题对象在自身状态变化时，会通知所有订阅者对象，使它们能够自动更新自己的状态。它是为了解决这样一种情况的发生，一个对象的改变需要同时改变其他对象，而且它不知道具体有多少对象需要改变。这就类似于发传单，目标发送通知，没有指定特定的对象，通知会自动传播，观察者自己决定是否需要看传单，发送者对于传单是否被看一无所知。

hpfeeds把不同的数据用频道来划分，支持传送任意的二进制数据。由频道的使用者决定传送数据结构的形式。通道的验证通过Authkey来完成，它由两部分组成：ident和secret，相当于身份和密码。把密码和每次连接的一个随机数进行hash摘要算法然后一起发送给服务器，这样保证不会被窃听，所以它可以运行在SSL、TLS下。

整个协议实现了三方的功能独立。使用这种方式降低了应用与业务逻辑之间的耦合，统一一个对外的发布接口，只需要关心监听的类型，不关心监听的具体处理人。频道的发布者只管发，不管订阅者有没有收到，很方便的建立一种一对多的依赖关系。在当一个对象的改变需要同时改变其他对象，而且它不知道具体有多少对象需要改变时，就可以使用订阅发布模式。

# #

## 二. hpfeeds的原理 ##

hpfeeds协议通过以下几部分实现：hpfeeds server， hpfeeds client，mongodb数据库。

### **1. hpfeeds server: ** ###

*  负责为每个client的连接生成一个连接标志；
 *  检查请求连接的client的id和sha1(nonce+Authkey)；
 *  检查client的请求类型，发布还是接收；

### 2. **hpfeeds client: ** ###

**　　**每个hpfeeds client都即可以作为发布者也可以作为订阅者，发布者和订阅者并不要求必须同时存在。

### 3. **Mongodb: ** ###

mongodb数据库用来存储每个client的id和secret，并且每当有client请求连接server时，server都会从mangodb中取出该client注册时的id和secret进行对比。 若对比一致则认证通过，client可以和server正常建立连接；若不一致则client与server建立连接失败。

### **4. client和server的认证过程：** ###

**　　**Client和server的认证及发布/订阅过程如下图1所示：

![531223818cc4a3280a87249bdbc8c3d7.png][]

**hpfeeds协议建立连接及通信的过程：**

1.  Client发起连接请求；
2.  server为每个client的连接生成一个连接标志，并将其发送给请求连接的client；
3.  client发送自己的id和sha1(nonce+Authkey)到server进行认证；
4.  server从mongodb中取出相应的信息检查验证，若认证通过，保持连接并执行后续操作。否则，断开连接；
5.  client发起publish/subscribe请求；
6.  server检查client请求消息的类型，发布/订阅。

# #

## 三. hpfeeds的消息格式 ##

### **1. Wire Protocol: ** ###

**　　**每个hpfeeds协议消息都携带了一个固定格式的消息头，有两个参数：消息总长度和请求类型。如下代码所示。

1 struct MsgHeader {
    2      unit32_t_messageLength;           // total message size, including this request type
    3      unit8_t_opCode;
    4 };

请求类型有以下几种：

*  error(0): errormessage
 *  info(1): server name, nonce　　
 *  auth (2): client id, sha1(nonce+authkey)
 *  publish (3): client id, channelname, payload 
 *  subscribe (4): client id, channelname

一个完整的发布类型的消息如下图所示，由消息头、client\_id的长度、client\_id、channelname的长度、channelname、传输内容payload。payload可以是任意类型的二进制数据。

![31ba958f063c9863219e5cb4da92f8be.png][]

## 四. hpfeeds源码解析 ##

Hpfeeds协议server与client的通讯主要也是使用TCP套接字的方式。

Hpfeeds server采用了事件驱动的方式来处理client的请求。这样做是为了应对高连接数高吞吐量的client请求，使用这种方法可以同时接收数百、数千甚至数万个事件，无论它们是内部请求还是网络连接，都可以高效地处理它们的操作。同时还能够极大的降低资源占用，增大服务接待能力，并提高网络传输效率。

Hpfeeds server与mongodb的连接及数据交互并没有使用Python自带的pymongo模块，而是使用了自己编写的一个基于事件驱动的MongoConn模块。这样做的目的也是为了处理高连接数的client请求。下面主要对hpfeeds的server和client的源码进行解析。

### 1. hpfeeds server 源码 ###

Hpfeeds server的工作方式，首先连接mongodb数据库，监听hpfeeds server的服务端口，设置事件监听器，关联相应处理函数，将事件监听器加入事件循环，启动事件循环进行监听。如果有client请求来，则会触发相应的事件，调用与事件相关联的函数进行处理操作。Hpfeeds server的主程序代码如下。

1 #!/usr/bin/env python
      2 
      3 
      4 import sys
      5 
      6 import struct
      7 import hashlib
      8 import collections
      9 import random
     10 
     11 import logging
     12 logging.basicConfig(level=logging.INFO)
     13 
     14 from evnet import loop, unloop, listenplain, EventGen       # 用于实现事件循环的模块
     15 from evnet.mongodb import MongoConn                         
     16 # 注意：Python本身有对mongodb进行操作的模块，但在hpfeeds server中没有使用，
     17 # 这里它自己实现了一个对mongodb进行操作的模块MongoConn,为了实现使用事件循
     18 # 的方式来对数据库进行操作
     19 
     20 FBIP = '0.0.0.0'        # hpfeeds server监听的地址和端口号
     21 FBPORT = 10000
     22 FBNAME = '@hp2'
     23 MONGOIP = '127.0.0.1'
     24 MONGOPORT = 27017
     25 
     26 OP_ERROR    = 0
     27 OP_INFO     = 1
     28 OP_AUTH     = 2
     29 OP_PUBLISH  = 3
     30 OP_SUBSCRIBE    = 4
     31 OP_UNSUBSCRIBE  = 5
     32 
     33 MAXBUF = 10* (1024**2)
     34 SIZES = {
     35     OP_ERROR: 5+MAXBUF,
     36     OP_INFO: 5+256+20,
     37     OP_AUTH: 5+256+20,
     38     OP_PUBLISH: 5+MAXBUF,
     39     OP_SUBSCRIBE: 5+256*2,
     40     OP_UNSUBSCRIBE: 5+256*2,
     41 }
     42 
     43 class BadClient(Exception):
     44     pass
     45 
     46 class FeedUnpack(object):            # 对client传来的数据进行解码
     47     def __init__(self):
     48         self.buf = bytearray()
     49     def __iter__(self):
     50         return self
     51     def next(self):
     52         return self.unpack()
     53     def feed(self, data):           # 将client传来的数据存入self.buf
     54         self.buf.extend(data)       
     55     def unpack(self):
     56         if len(self.buf) < 5:       # 如果self.buf的总长度小于5，说明请求消息为空
     57                                     # 因为client各种请求类型的消息长度都是大于5的
     58             raise StopIteration('No message.')
     59 
     60         ml, opcode = struct.unpack('!iB', buffer(self.buf,0,5))
     61                                                 # 解码出ml和opcode
     62         if ml > SIZES.get(opcode, MAXBUF):      # ml为hpfeeds消息的总长度
     63             raise BadClient('Not respecting MAXBUF.')
     64 
     65         if len(self.buf) < ml:      # self.buf中的数据长度小于该条消息的总长度，抛出异常
     66             raise StopIteration('No message.')
     67         
     68         data = bytearray(buffer(self.buf, 5, ml-5))
     69         del self.buf[:ml]       # 删除self.buf中的数据
     70         return opcode, data     
     71         # data中包含了len(client_id),client_id,length(channelname), channelname,payload
     72 
     73 
     74 class FeedConn(EventGen):
     75     def __init__(self, conn, addr, db):
     76         EventGen.__init__(self)
     77         self.conn = conn
     78         self.addr = addr
     79         self.db = db
     80         self.pubchans = set()
     81         self.subchans = set()
     82         self.idents = set()
     83         self.delay = False
     84 
     85         self.rand = struct.pack('<I', random.randint(2**31,2**32-1))    # 产生一个随机数
     86         self.fu = FeedUnpack()
     87 
     88         conn._on('read', self.io_in)
     89         conn._on('close', self.closed)
     90 
     91         self.sendinfo()
     92 
     93     def sendinfo(self):
     94         self.conn.write(self.msginfo())
     95 
     96     def auth(self, ident, hash):    # server和client的认证函数
     97         p = self.db.query('hpfeeds.auth_key', {'identifier': str(ident)}, limit=1)
     98         # 查询mongodb中的数据,返回的p为一个Promise()对象
     99         p._when(self.checkauth, hash)   # 调用checkauth函数对client进行认证
    100 
    101         def dbexc(e):       # mongodb查询异常处理函数
    102             logging.critical('Database query exception. {0}'.format(e))
    103             self.error('Database query exception.')
    104         
    105         p._except(dbexc)    # 如果出现异常则执行响应的处理函数
    106 
    107         self.delay = True
    108 
    109     def checkauth(self, r, hash):   # server与client的认证处理函数
    110         if len(r) > 0:              # r是self._result
    111             akobj = r[0]            
    112             akhash = hashlib.sha1('{0}{1}'.format(self.rand, akobj['secret'])).digest()
    113             if akhash == hash:      # 将数据库中取得的secret与self.rand进行hash摘要算法进行对比
    114                 self.pubchans.update(akobj.get('publish', []))  # 更新发布频道
    115                 self.subchans.update(akobj.get('subscribe', []))# 更新订阅频道
    116                 self.idents.add(akobj['identifier'])        # 将认证成功的client_id添加到self.idents
    117                 logging.info('Auth success by {0}.'.format(akobj['identifier']))
    118             else:
    119                 self.error('authfail.')
    120                 logging.info('Auth failure by {0}.'.format(akobj['identifier']))
    121         else:
    122             self.error('authfail.')
    123         self.delay = False
    124         self.io_in(b'')
    125     
    126     def closed(self, reason):
    127         logging.debug('Connection closed, {0}'.format(reason))
    128         self._event('close', self)
    129 
    130     def may_publish(self, chan):
    131         return chan in self.pubchans
    132 
    133     def may_subscribe(self, chan):
    134         return chan in self.subchans
    135 
    136     def io_in(self, data):      # 传送请求发布和订阅的数据函数
    137         self.fu.feed(data)      # 数据存入self.buf
    138         if self.delay:          # 经FeedUnpack的实例处理后的数据为opcode, data
    139             return              # data中包含了len(client_id),client_id,length(channelname), channelname,payload
    140         try:
    141             for opcode, data in self.fu:
    142                 if opcode == OP_PUBLISH:                        # 处理发布请求
    143                     rest = buffer(data, 0)                      # 数据存入buffer
    144                     ident, rest = rest[1:1+ord(rest[0])], buffer(rest, 1+ord(rest[0]))
    145                     chan, rest = rest[1:1+ord(rest[0])], buffer(rest, 1+ord(rest[0]))
    146                                                                 # 解码出发布请求包中的数据
    147                     if not ident in self.idents:
    148                         self.error('identfail.')
    149                         continue
    150 
    151                     if not self.may_publish(chan):
    152                         self.error('accessfail.')
    153                         continue
    154                     
    155                     self._event('publish', self, chan, data)    # 触发发布请求的处理事件
    156                 elif opcode == OP_SUBSCRIBE:                    # 处理订阅请求
    157                     rest = buffer(data, 0)
    158                     ident, chan = rest[1:1+ord(rest[0])], rest[1+ord(rest[0]):]
    159 
    160                     if not ident in self.idents:
    161                         self.error('identfail.')
    162                         continue
    163 
    164                     checkchan = chan
    165                     if chan.endswith('..broker'): checkchan = chan.rsplit('..broker', 1)[0]
    166 
    167                     if not self.may_subscribe(checkchan):
    168                         self.error('accessfail.')
    169                         continue
    170 
    171                     self._event('subscribe', self, chan, ident) # 触发订阅请求的处理事件
    172                 elif opcode == OP_UNSUBSCRIBE:                  # 处理取消订阅请求
    173                     rest = buffer(data, 0)
    174                     ident, chan = rest[1:1+ord(rest[0])], rest[1+ord(rest[0]):]
    175 
    176                     if not ident in self.idents:
    177                         self.error('identfail.')
    178                         continue
    179 
    180                     if not self.may_subscribe(chan):
    181                         self.error('accessfail.')
    182                         continue
    183 
    184                     self._event('unsubscribe', self, chan, ident)# 触发取消订阅请求的事件
    185                 elif opcode == OP_AUTH:                         # 处理认证请求
    186                     rest = buffer(data, 0)
    187                     ident, hash = rest[1:1+ord(rest[0])], rest[1+ord(rest[0]):]
    188                     self.auth(ident, hash)                      # 认证函数
    189                     if self.delay:
    190                         return
    191 
    192         except BadClient:
    193             self.conn.close()           # 关闭客户端与服务器的连接
    194             logging.warn('Disconnecting bad client: {0}'.format(self.addr))
    195     def forward(self, data):
    196         self.conn.write(self.msghdr(OP_PUBLISH, data))
    197 
    198     def error(self, emsg):
    199         self.conn.write(self.msgerror(emsg))
    200 
    201     def msgerror(self, emsg):
    202         return self.msghdr(OP_ERROR, emsg)
    203 
    204     def msginfo(self):
    205         return self.msghdr(OP_INFO, '{0}{1}{2}'.format(chr(len(FBNAME)%0xff), FBNAME, self.rand))
    206 
    207     def msghdr(self, op, data):                 # 对消息进行封包处理的函数
    208         return struct.pack('!iB', 5+len(data), op) + data
    209 
    210     def msgpublish(self, ident, chan, data):    # 发布消息预处理函数
    211         return self.msghdr(OP_PUBLISH, struct.pack('!B', len(ident)) + ident + struct.pack('!B', len(chan)) + chan + data)
    212 
    213     def publish(self, ident, chan, data):       # 发布消息函数
    214         self.conn.write(self.msgpublish(ident, chan, data))
    215 
    216 class FeedBroker(object):
    217     def __init__(self):
    218         self.ready = False
    219 
    220         self.db = None
    221         self.initdb()           # 初始化mongodb数据库
    222 
    223         self.listener = listenplain(host=FBIP, port=FBPORT)     # hpfeeds server 开始监听端口，返回的listener是一个监听事件
    224         self.listener._on('close', self._lclose)                # 为事件关联函数
    225         self.listener._on('connection', self._newconn)          # 有新的client连接则触发该事件
    226 
    227         self.connections = set()                                # 连接的client集合
    228         self.subscribermap = collections.defaultdict(list)      
    229         self.conn2chans = collections.defaultdict(list)
    230 
    231     def initdb(self):
    232         self.db = MongoConn(MONGOIP, MONGOPORT)     # 连接mongodb
    233         self.db._on('ready', self._dbready)         # 关联事件和回调函数
    234         self.db._on('close', self._dbclose)
    235 
    236     def _dbready(self):
    237         self.ready = True
    238         logging.info('Database ready.')
    239 
    240     def _dbclose(self, e):
    241         logging.critical('Database connection closed ({0}). Exiting.'.format(e))
    242         unloop()
    243 
    244     def _lclose(self, e):
    245         logging.critical('Listener closed ({0}). Exiting.'.format(e))
    246         unloop()
    247 
    248     def _newconn(self, c, addr):                                # client请求连接server的处理函数
    249         logging.debug('Connection from {0}.'.format(addr))
    250         fc = FeedConn(c, addr, self.db)                         # 处理client的各种类型的请求的监听事件
    251         self.connections.add(fc)                                
    252         fc._on('close', self._connclose)                        # 为fc关联事件和回调函数
    253         fc._on('subscribe', self._subscribe)
    254         fc._on('unsubscribe', self._unsubscribe)
    255         fc._on('publish', self._publish)
    256 
    257     def _connclose(self, c):                                    # 关闭server与client连接
    258         self.connections.remove(c)
    259         for chan in self.conn2chans[c]:
    260             self.subscribermap[chan].remove(c)
    261             for ident in c.idents:
    262                 self._brokerchan(c, chan, ident, 0)
    263     def _publish(self, c, chan, data):
    264         logging.debug('broker publish to {0} by {1}'.format(chan, c.addr))
    265         for c2 in self.subscribermap[chan]:             # 该频道中的订阅者
    266             if c2 == c: continue                        # 把发布者本身除外
    267             c2.forward(data)                            # 向该频道的所有订阅者推送要发布的数据
    268         
    269     def _subscribe(self, c, chan, ident):               # 订阅请求的处理仅把订阅者添加到频道中，然后触发推送数据的循环
    270         logging.debug('broker subscribe to {0} by {2} @ {1}'.format(chan, c.addr, ident))
    271         self.subscribermap[chan].append(c)
    272         self.conn2chans[c].append(chan)
    273         self._brokerchan(c, chan, ident, 1)
    274     
    275     def _unsubscribe(self, c, chan, ident):             # 某个客户端取消对某个频道的订阅
    276         logging.debug('broker unsubscribe to {0} by {1}'.format(chan, c.addr))
    277         self.subscribermap[chan].remove(c)
    278         self.conn2chans[c].remove(chan)
    279         self._brokerchan(c, chan, ident, 0)
    280 
    281     def _brokerchan(self, c, chan, ident, subscribe=0):     # 触发推送数据循环
    282         data = 'join' if subscribe else 'leave'
    283         if self.subscribermap[chan+'..broker']:
    284             for c2 in self.subscribermap[chan+'..broker']:
    285                 if c2 == c: continue
    286                 c2.publish(ident, chan+'..broker', data)
    287 
    288 def main():
    289     fb = FeedBroker()
    290 
    291     loop()                  # 启动事件监听循环
    292     return 0
    293  
    294 if __name__ == '__main__':
    295     sys.exit(main())

### 2. hpfeeds client源码 ###

Hpfeeds client的工作方式，与server成功建立连接后，开始相应的publish/subscribe操作。如果是做为订阅者，则会与server一直保持连接状态，不断读取订阅频道中的内容；如果是作为发布者，则每次推送完数据后，不管订阅者有没有收到信息，都立刻关闭与server的连接。

1 #!/usr/bin/env python
      2 
      3 import sys
      4 import optparse     # optparse模块用于处理命令行参数
      5 import datetime
      6 import logging
      7 import string
      8 
      9 import hpfeeds
     10 
     11 def log(msg):
     12     print '[feedcli] {0}'.format(msg)
     13 
     14 def main(opts, action, pubdata=None):
     15     outfd = None
     16     if opts.output:
     17         try: outfd = open(opts.output, 'a')
     18         except:
     19             log('could not open output file for message log.')
     20             return 1
     21     else:
     22         outfd = sys.stdout
     23 
     24     try: hpc = hpfeeds.new(opts.host, opts.port, opts.ident, opts.secret, certfile=opts.certfile)
     25     # 连接hpfeeds server，返回值hpc为hpfeeds模块中HPC类对象，
     26     # 如果client与server成功连接，并认证成功，则程序继续往后执行；否则抛出异常，程序退出
     27     except hpfeeds.FeedException, e:
     28         log('Error: {0}'.format(e))
     29         return 1
     30     
     31     log('connected to {0}'.format(hpc.brokername))
     32 
     33     if action == 'subscribe':                           # 订阅请求
     34         def on_message(ident, chan, payload):           # 显示记录收到的订阅信息
     35             if [i for i in payload[:20] if i not in string.printable]:
     36                 log('publish to {0} by {1}: {2}'.format(chan, ident, payload[:20].encode('hex') + '...'))
     37             else:
     38                 log('publish to {0} by {1}: {2}'.format(chan, ident, payload))
     39 
     40         def on_error(payload):                          # 记录错误信息
     41             log('Error message from broker: {0}'.format(payload))
     42             hpc.stop()                                  # 停止循环
     43 
     44         hpc.subscribe(opts.channels)
     45         try: hpc.run(on_message, on_error)      # 接收server推送过来的数据，调用on_message(),on_error()进行处理
     46         except hpfeeds.FeedException, e:        # 抛出异常，程序退出
     47             log('Error: {0}'.format(e))
     48             return 1
     49 
     50     elif action == 'publish':                   # 发布请求
     51         hpc.publish(opts.channels, pubdata)     # 推送数据
     52         emsg = hpc.wait()                       # 若推送成功，返回None；否则，返回其它值
     53         if emsg: print 'got error from server:', emsg
     54 
     55     elif action == 'sendfile':
     56         pubfile = open(pubdata, 'rb').read()
     57         hpc.publish(opts.channels, pubfile)
     58 
     59     log('closing connection.')
     60     hpc.close()
     61 
     62     return 0
     63 
     64 def opts():                         # 获取命令行参数
     65     usage = "usage: %prog -i ident -s secret --host host -p port -c channel1 [-c channel2, ...] <action> [<data>]"
     66     parser = optparse.OptionParser(usage=usage)
     67     parser.add_option("-c", "--chan",                       # 定义命令行参数
     68         action="append", dest='channels', nargs=1, type='string',
     69         help="channel (can be used multiple times)")
     70     parser.add_option("-i", "--ident",
     71         action="store", dest='ident', nargs=1, type='string',
     72         help="authkey identifier")
     73     parser.add_option("-s", "--secret",
     74         action="store", dest='secret', nargs=1, type='string',
     75         help="authkey secret")
     76     parser.add_option("--host",
     77         action="store", dest='host', nargs=1, type='string',
     78         help="broker host")
     79     parser.add_option("-p", "--port",
     80         action="store", dest='port', nargs=1, type='int',
     81         help="broker port")
     82     parser.add_option("-o", "--output",
     83         action="store", dest='output', nargs=1, type='string',
     84         help="publish log filename")
     85     parser.add_option("--certfile",
     86         action="store", dest='certfile', nargs=1, type='string',
     87         help="certfile for ssl verification (CA)", default=None)
     88     parser.add_option("--debug",
     89         action="store_const", dest='debug',
     90         help="enable debug log output", default=False, const=True)
     91 
     92     options, args = parser.parse_args()                     # 定义好所有的命令行参数，调用 parse_args()来解析程序的命令行
     93 
     94     if len(args) < 1:
     95         parser.error('You need to give "subscribe" or "publish" as <action>.')
     96     if args[0] not in ['subscribe', 'publish', 'sendfile']:
     97         parser.error('You need to give "subscribe" or "publish" as <action>.')
     98     if options.debug:
     99         logging.basicConfig(level=logging.DEBUG)
    100     else:
    101         logging.basicConfig(level=logging.CRITICAL)
    102 
    103     action = args[0]
    104     data = None
    105     if action == 'publish':
    106         data = ' '.join(args[1:])
    107     elif action == 'sendfile':
    108         data = ' '.join(args[1:])
    109 
    110     return options, action, data
    111 
    112 if __name__ == '__main__':
    113     options, action, data = opts()      # 获取命令行参数
    114     try:
    115         sys.exit(main(options, action, pubdata=data))   # 从main()函数开始执行
    116     except KeyboardInterrupt:
    117         sys.exit(0)

[531223818cc4a3280a87249bdbc8c3d7.png]: /images/20220508/35a950ac7a014e848bb623b0a55af504.png
[31ba958f063c9863219e5cb4da92f8be.png]: /images/20220508/d9b78540e54444368d6017bc0808f104.png