背景介绍
Odoo 是最好的开源企业应用系统，没有之一。虽然有很多技术落伍了，但是经过近20年的持续发展，Odoo积累了所有企业运营所需要的所有软件；即使没有提供企业所需要的功能，它本身还是一个完整的开发平台，可以开发出你能想象的应用，当然不会开发也没有关系，可以尽情参考已经开源的模块是如何做到的。
这里就讲一讲 Odoo 如何使用‘过时的’技术实现的即时通讯，通过对即时通讯的支持，Odoo可以让企业内部人员可以通过 Odoo 进行实时沟通，也可以让企业内部人员和网站客户进行实时沟通；同时 Odoo 将即时通讯的消息与邮件和 Robot 整合，为营销自动化和服务智能化乃至应用智能化交互提供了基础。
核心技术
通过数据库消息队列
先说最重要的，Odoo 即时通讯使用了 PostgreSQL 数据库的 listen 和 notify 的机制完成。这个机制是 PostgreSQL 数据库私有的，其它数据库未必支持。所以要用 Odoo 是必须要用 PostgreSQL，这是原因之一。参考这里可以了解更多关于 PostgreSQL listen notify 的信息。
使用 listen 和 notify 可以让连接数据库的各个客户端之间进行实时通讯。
通过长连接
连接数据库的客户端不是 Odoo 的客户端，数据库的客户端实际上是 Odoo 的服务端，是 Python 代码连接 数据库；而 Odoo 客户端是通过 Javascript 实现的 Web 应用，它通过长连接方式与 Odoo 后台保持信息的实时性。长连接的链接地址是 longpolling/poll ，Odoo 客户端会发起这个连接请求，Odoo 服务端处理这个请求，如果有这个请求关注的 Channels 的消息，那么这个请求就会立即返回，如果没有消息，这个连接会保持 TIMEOUT 秒，目前 TIMEOUT 是50秒。Channels 就是会话标记，可以理解为一个聊天室、一个群等等，客户 poll 数据的时候要写上它关注的 Channels。
异步处理
如果很多用户同时使用 Odoo，那么 Odoo 为每个客户保持一个连接，这是无疑问的，因为没有连接就没有办法读写数据；但是每个连接是在一个线程里面呢，还是多个呢？简单说，Odoo 只为 longpolling 维护了一个线程或者一个进程（gevent），如果你启动 Odoo 的时候使用了 worker 参数，就意味这 Odoo 要以多进程方式运作，如果没有指定 woker 就是多线程方式，如果你启动的是线程模式，longpolling 将是一个线程，如果你启动的是 worker （进程）模式那么 Odoo 会通过 Popen 一个全新进程，这个全新进程的命令行 加上 gevent，很怪异吧，确实就是这么干的。
def long_polling_spawn(self):
nargs = stripped_sys_argv()
cmd = [sys.executable, sys.argv[0], 'gevent'] + nargs[1:]
popen = subprocess.Popen(cmd)
self.long_polling_pid = popen.pid
把原来的命令行插入一个 gevent，再启动一遍。当然后续的代码会判断如果是以 gevent 启动命令的，这是要启动 longpolling。
gevent 在 Python 3 asyncio 的大环境下是个过时的技术了，它使用了 Monkey Patch 的方式对 Python 库进行了异步化，感觉代码的书写方式还是一样，但是已经异步化了。好处是代码在没有 gevent 的时候可以同步跑，引入 gevent 后不用改变代码逻辑就可以异步化。有人会问，异步化有啥好处啊？异步化可以让 Odoo 同时处理多个连接，就这么简单，如果没有异步化，一个连接就占用了 Odoo，别的连接进不来，解决这个问题的老方法是启动更多进程，但是进程的方式太重了，随着互联网服务的普及，开发人员发现实际上只需要维护 I/O 并不需要启动多个进程或者多个进程，只需要维护好文件描述符，并且能够正确发现这些描述符什么时候该读什么时候该写。selelct，poll，epoll一步一步把异步I/O的性能榨干了最后一滴血。
在 Python 2 的时候，Python 没有内置异步 I/O 的功能，所以 gevent，Tornado 都是解决 Python 异步 I/O 问题的。Odoo 使用了 gevent，当 longpolling 服务正在服务一个客户端的时候，也没有任何消息给这个客户端，那么这个客户端将保持连接 50 秒，这时候 longpolling 服务端会把基于这个连接的处理 wait 让出 CPU，让其它连接能进来。 当这个连接的文件描述符准备好读写的时候 Odoo 通过 select 调用得以了解。然后会通过 Thread 或者 gevent 的 Event 通知等待的客户连接。
def loop(self):
""" Dispatch postgres notifications to the relevant polling threads/greenlets """
_logger.info("Bus.loop listen imbus on db postgres")
with odoo.sql_db.db_connect('postgres').cursor() as cr:
conn = cr._cnx
cr.execute("listen imbus")
cr.commit();
while True:
if select.select([conn], [], [], TIMEOUT) == ([], [], []):
pass
else:
conn.poll()
channels = []
while conn.notifies:
channels.extend(json.loads(conn.notifies.pop().payload))
# dispatch to local threads/greenlets
events = set()
for channel in channels:
events.update(self.channels.pop(hashable(channel), set()))
for event in events:
event.set()
上边的这段代码在 bus 模块里面，Odoo 只有一个线程或者 gevent 程序去无差别的listen 系统所有的 imbus 上的消息，notify imbus 的消息都会让 select 返回准备好的文件描述符（不是空的，所以就不会等于 ([],[],[])），无差别就是它不判断 Channel，而每个客户端是需要关注 Channel的，所以这是系统级别的不是用户级别的，它取出数据后通过 event set 来通知那些 wait 在具体 event （关联了 Channel）上的客户。
Channels 的 Overload
每次 longpolling 的 poll 请求都要带上这个用户想要关注的 channels，而 用户怎么知道自己要 polling 什么channels呢？
Channels 一般来自两种可能，一个是同一种应用导致的会话数量的增加，比如在线客服，每个新访客都有可能跟 Odoo 的用户建立一个 Channel 就是会话，这样就会有很多会话。
还有一种可能就是，Odoo 有很多应用，每个应用都会有自己建立或者判断 Channel 的方式，在线客服是 Odoo 的一个应用，CRM 也是一个应用，每个应用对 Channel 的标记和维护方法各不相同，一般是一个元组 （db,table,id) 再 hashable 或者文本化一下，就变成字符串，作为 Channel 的唯一标记，具体有多少个这样的 Channels 也是存储在各自应用的表里面。所以 bus 应用的 Controller 提供了一个可以 Overload 的机会来修改 Channels，就是 _load。
# override to add channels
def _poll(self, dbname, channels, last, options):
# update the user presence
if request.session.uid and 'bus_inactivity' in options:
request.env['bus.presence'].update(options.get('bus_inactivity'))
request.cr.close()
request._cr = None
return dispatch.poll(dbname, channels, last, options)
它轻描淡写的注释暴露了它存在的意义。
再看 mail 应用下的 controller 对这个函数的 overload。
# --------------------------
# Extends BUS Controller Poll
# --------------------------
def _poll(self, dbname, channels, last, options):
if request.session.uid:
partner_id = request.env.user.partner_id.id
if partner_id:
channels = list(channels) # do not alter original list
for mail_channel in request.env['mail.channel'].search([('channel_partner_ids', 'in', [partner_id])]):
channels.append((request.db, 'mail.channel', mail_channel.id))
# personal and needaction channel
channels.append((request.db, 'res.partner', partner_id))
channels.append((request.db, 'ir.needaction', partner_id))
return super(MailChatController, self)._poll(dbname, channels, last, options)
把在 mail （就是讨论应用）中需要的 channels 都圈出来提供给 bus 应用去处理。
这里面又学到一个 Odoo 的知识，如何搜索 many2many 的字段 （channel partner ids），因为 many2many 是 Odoo 加了一个中间表实现的，没看过这段代码还不知道咋搜索呢。这里面的 channel partner ids 是在 mail channel 中对应的 partner id，在 res partner 表中也有 partner 对应的 mail channel。这是一个多对多的关系，一个 mail channel 可以含有多个 partner，一个partner 可以在多个 mail channel 中，这很自然，人可以在很多对话中，对话中含有很多人 。但是如果你想搜索哪些对话中含有我这个人的怎么搜？
其它
Odoo 的即时通讯几乎都在 bus 这个 addon 下面，但是在odoo 全局的代码中也有很多配合的 code，比如上文提到的 gevent 命令行；还有更加复杂的部分，就是 WSGI 和 数据连接的处理部分，由于 longpolling 同时重用了普通 httprequest 和数据库运行环境 （registry，Enviroments，Enviroment，cusor），这段代码比较乱，不如 addon 里面的结构清晰，当然可能也是为了让 addon 结构清晰，不得不做出的妥协。值得说明的是，当 longpolling 的请求来的时候，WSGI 请求自带的 Odoo 数据库执行环境会被抛弃，而是每次请求重新再次建立：
event.wait(timeout=timeout)
with registry.cursor() as cr:
env = api.Environment(cr, SUPERUSER_ID, {})
notifications = env['bus.bus'].poll(channels, last, options)
让我们知道了 Odoo 如何每次建立数据环境。如果不是每次建立环境那么这里的数据操作别的客户不会同时发现的。
通过分析 Odoo 的 IM 实现过程可以看出 Odoo 的技术的确有点过时了，跟踪的不够猛。因为 Python 3 已经支持 syncio 了，关于 asyncio 可以读读这个 blog 。
如果通过 asyncio 去实现，我的思路是在 asyncio 中加入 postgresql connection 的描述符，就是上边用来select 的，watching 这个描述符。当有数据的时候 callback 就会运行，再去通过 asyncio 的 locks 中的 Event 去 set()。用 asyncio.wait_for(event.wait(), timeout) 来响应用户的请求，用户的 HTTP 请求就会被阻塞直到 Event 被 set 或者超时，而 CPU 会被让出，完美。
`loop.``add_reader`(*fd*, *callback*, **args*)
Start monitoring the *fd* file descriptor for read availability and invoke *callback* with the specified arguments once *fd* is available for reading.
这样就用原生的 Python 3 解决了，不需要引入 gevent，也不需要引入异步的 PostgreSQL Python 库，重用原来的 psycopg2 阻塞库。
网友评论