背景介绍

Odoo 是最好的开源企业应用系统，没有之一。虽然有些技术已经落伍了，Odoo 的前端通过 JQuery 做的 MVC 跟当今的 React 的革命性前端编程正好差一个时代，不过Odoo 14 已经开始了 OWL 貌似要追上了。

丁贵金：为什么 Odoo 选择自行开发 OWL 猫头鹰 Javascript 框架​zhuanlan.zhihu.com

经过近20年的持续发展，Odoo 积累了企业运营所需要的各种软件；它本身还是一个完整的 WEB 应用开发框架（非常非常独特的前端和后段统一的开发框架），可以自由开发各种 WEB 应用；开发中的遇到任何问题，可以通过阅读已开源的Odoo 应用，看看类似的特性是如何做到的。

本文关注 Odoo 的即时通讯特性，Odoo 的即时通讯是 Odoo 框架的基础部分，研究过后，发现其总体实现还是比较简单的，基于 Odoo 是一个极端模块化的系统，Odoo 的即时通讯为 Odoo 模块化系统支持专门进行了设计，非常值得研究学习。

Odoo 即时通讯可以让企业内部人员进行实时沟通，也可以让企业内部人员和网站客户进行实时沟通；同时 Odoo 将即时通讯的消息与Bot 整合，为营销自动化和服务智能化乃至应用智能化交互提供了基础。

Odoo 即时通讯还在基础架构上为 Odoo 应用提供了社交化支持，应用的数据可以很容易增加订阅者，当数据更改的时候，订阅者可以收到消息，用户也可以直接为数据添加评论，评论会推送给数据的订阅者。所以我们经常能在 Odoo 中看到针对一个具体的数据表单下面可以有用户的评论和这个数据本身更改的历史信息。

核心技术

数据库消息队列

先说最重要的，Odoo 即时通讯使用了 PostgreSQL 数据库的 listen 和 notify 的机制完成。这个机制是 PostgreSQL 数据库私有的，其它数据库未必支持。Odoo 依赖 PostgreSQL，这是原因之一。参考这里可以了解更多关于 PostgreSQL listen notify 的信息。

使用数据库的 listen 和 notify 可以让连接数据库的各个客户端之间进行实时通讯。值得注意的是，这个特性是 PostgreSQL 所特有的。

在 Odoo 中并没有让每个客户都去使用数据库的 listen 和 notify 这个功能，而是由 Odoo 统一使用的。客户通过浏览器访问 Odoo，只要说明自己关心的 Channels，并且通过 Event 异步等待 （Event.wait()）。Odoo 统一访问 listen，然后根据 listen 返回的数据，解析出来这些数据是哪些 Channels，从而通知 Event (Event.set())，这样客户就知道自己有消息要处理。

通过长连接

连接数据库的客户端不是 Odoo 的客户端，数据库的客户端实际上是 Odoo 的服务端，是 Python 代码连接 数据库；而 Odoo 客户端是通过 Javascript 实现的 Web 应用，它通过长连接方式与 Odoo 后台保持信息的实时性。长连接的链接地址 URL 是 /longpolling/poll ，Odoo 客户端会发起这个连接请求，如果有这个请求关注的 Channels 的消息，那么这个请求就会立即返回，如果没有消息，这个连接会尝试保持 TIMEOUT 秒，目前 TIMEOUT 是50秒。Channels 就是会话标记，可以理解为一个聊天室、一个群等等，客户 poll 数据的时候要写上它关注的 Channels。

长连接的 URL 是 /longpolling/poll，在服务端为每次客户请求创建一个 Event，然后把这个客户想关注的 Channels（这个 Channels 是个数组）关联这个 Event，异步等待这个Event；一个客户连接只有一个 Event，但是多个 Channels，这些 Channels 关联这个Event，一旦这些 Channels 中任何一个 Channel 有消息，那么这个 Event 就会被通知，客户的长连接就会返回，客户就知道有消息了，可以进行下一步动作，比如获取消息的详情。如果系统中没有任何消息跟客户的这些 Channels 有关系，客户就会一直在 Event 上异步等待，直到超时。超时结束后长连接就会结束，客户端会重新发起连接请求，再次进入等待的状态。

异步处理

如果很多用户同时使用 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 秒，这个等待过程并不会抓着 CPU 不放，因为它只是等待一个事件的发生，并不需要计算。

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._cnxcr.execute("listen imbus")cr.commit();while True:# 异步监听文件描述符，三个数组分别是指可读监控，可写监控，异常监控，# 如果数组中的文件描述符符合监控的条件就会返回if select.select([conn], [], [], TIMEOUT) == ([], [], []):# 如果超时了passelse:# 发现 conn 这个文件描述符可以读conn.poll()channels = []# 读出所有的 notifieswhile conn.notifies:# 每个 notify 的数据 payload 就是 channelchannels.extend(json.loads(conn.notifies.pop().payload))# dispatch to local threads/greenletsevents = set()for channel in channels:# 注意这里面用到 self.channels 这里面是 channel 和 event 的映射字典# 通过 set 能够排除重复的 event，每个channel 对应一个 event 的集合，# 就是可能很多客户在等候这个 channel，当然也可能没有任何客户在等候这个 channel# 那么就是空集合events.update(self.channels.pop(hashable(channel), set()))for event in events:# event 能够在多个协程之间通讯，客户使用 event wait在异步等待，这里面通过# event set 通知等待可以结束了。event.set()

上边的这段代码在 bus 模块里面，Odoo 只有一个线程或者 gevent 程序去 listen 系统所有的 imbus 上的消息，notify imbus 的消息都会让 select 返回准备好的文件描述符（不是空的，所以就不会等于 ([],[],[])），收到数据库的消息通知后，通过分析消息知道这些消息是什么 Channels （实际上，消息的内容就是 Channel 的 Hash），通过 Channel 找到其对应的 Event，Event 一定对应一个客户端的长连接。执行 Event set 来通知那些 wait 在 Event 上的客户。具体过程可以阅读我在代码里面写的注释。

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，就是 _poll。

# override to add channelsdef _poll(self, dbname, channels, last, options):# update the user presenceif 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)

‘ override to add channels‘ 轻描淡写的注释暴露了它存在的意义。 一个客户通过浏览器与 Odoo 建立的长连接就是一个longpolling 的HTTP 请求，这个HTTP请求通过 _poll 这个函数调用 dispatch poll 去异步等待数据库的 notify，重载这个函数可以有机会在真正执行 dispatch poll 之前收集 Channels。

再看 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.idif partner_id:channels = list(channels)       # do not alter original listfor 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 channelchannels.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 中所有的 人/用户/客户/访客/公司 都是用 res.partner 这个表来维护的。

上段代码潜藏了一个 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 中，这很自然，人可以在很多对话中，对话中含有很多人 。这段代码搜索 mail channel 表中 channel partner ids 中包含用户 id 的所有记录，然后把这些记录按照(db,table,id)的形式合成Channel。

最后加上 res partner 和 ir needaction 关于partner id 的 Channel，不负责任的猜测一下（尚未查证），res partner 这个 Channel 的含义可能是当这个用户以用户身份订阅其他数据记录的修改的 Channel，因为它指向这个访问者，所以如果以访问者身份去操作导致的消息可能需要这个Channel；ir needaction 这个 Channel 的含义可能是需要这个用户执行 Activity 的时候发送的通知所使用的 Channel，顾名思义哈，Odoo 中有 Activity 模块，这也是一个基础模块，各个应用模块都可以使用，它的意思是提醒 Odoo 用户去进行一些计划行为动作，比如让用户打电话，开会，处理工单等等。

其它

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 已经支持 asyncio 了，关于 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 阻塞库。