Tornado 后台任务使用 cx_Oracle答案

【问题标题】：Tornado background task using cx_OracleTornado 后台任务使用 cx_Oracle
【发布时间】：2018-05-09 08:14:40
【问题描述】：

我正在运行一个Bokeh 服务器，使用底层的Tornado 框架。

我需要服务器在某个时候刷新一些数据。这是通过使用 Cx_Oracle 从 Oracle 数据库中获取行来完成的。

感谢 Tornado 的 PeriodicCallback，程序每 30 秒检查一次是否应该加载新数据：

server.start() 
from tornado.ioloop import PeriodicCallback
pcallback = PeriodicCallback(db_obj.reload_data_async, 10 * 1e3)
pcallback.start()
server.io_loop.start()

其中db_obj 是一个类的实例，它负责处理与数据库相关的功能（连接、获取等）。

基本上，reload_data_async 函数是这样的：

executor = concurrent.futures.ThreadPoolExecutor(4)

# methods of the db_obj class ...

@gen.coroutine
def reload_data_async(self):
  # ... first, some code to check if the data should be reloaded ...
  # ...
  if data_should_be_reloaded:
    new_data = yield executor.submit(self.fetch_data)

def fetch_data(self):
   """ fetch new data in the DB """
   cursor = cx.Cursor(self.db_connection) 
   cursor.execute("some SQL select request that takes time (select * from ...)")

   rows = cursor.fetchall()
   # some more processing thereafter 
   # ...

基本上，这是可行的。但是，当我尝试在fetch_data 中加载数据时读取数据（通过单击以在 GUI 中显示），程序由于竞争条件而崩溃（我猜？）：它正在同时获取数据时访问数据时间。

我刚刚发现tornado.concurrent.futures 不是线程安全的：

tornado.concurrent.Future 类似于 concurrent.futures.Future，但是不是线程安全的（因此与单线程一起使用更快事件循环）。

总而言之，我认为我应该创建一个新线程来处理 CX_Oracle 操作。我可以使用 Tornado 并继续使用 PerodicCallback 函数吗？如何将我的异步操作转换为线程安全的？这样做的方法是什么？

PS：我使用的是 Python 2.7

谢谢

【问题讨论】：

你想读什么？假设您将获取的数据存储在一个列表（或其他任何东西）中，然后您尝试在另一个线程（主线程）中读取此列表。我认为这不会导致崩溃。
@Sraw 很好，它在fetchall 期间崩溃了，并且它的重现性 100% 的时间。基本上，我正在做一个select * from ...，所以我尝试读取一堆或行，其中列是数字和字符串。

标签： python multithreading asynchronous tornado cx-oracle

【解决方案1】：

解决了！

@Sraw 是对的：它不应该导致崩溃。

解释：fetch_data() 正在使用 cx Oracle Connection 对象 (self.db_connection)，默认情况下它不是线程安全的。将 threaded 参数设置为 True 会使用互斥锁包装共享连接，如 Cx Oracle 文档中所述：

线程参数应该是一个布尔表达式指示 Oracle 是否应该包装对连接的访问带有互斥锁。在单线程应用程序中这样做会强加一个大约 10-15% 的性能损失，这就是默认值为 False 的原因。

所以我在我的代码中，我只是修改了以下内容，当用户在刷新数据时尝试访问数据时，它现在可以正常工作而不会崩溃：

# inside the connect method of the db_obj class
self.db_connection = cx.connect('connection string', threaded=True) # False by default

【讨论】：