如何在数据库中使龙卷风请求成为原子

Posted 2021-04-24

我有一个用Tornado异步框架编写的python应用程序。当HTTP请求进入时，会调用此方法：

@classmethod
def my_method(cls, my_arg1):

    # Do some Database Transaction #1
    x = get_val_from_db_table1(id=1, 'x')
    y = get_val_from_db_table2(id=7, 'y')
    x += x + (2 * y) 

    # Do some Database Transaction #2
    set_val_in_db_table1(id=1, 'x', x)

    return True

这三个数据库操作是相互关联的。这是一个并发应用程序，因此多个此类HTTP调用可以同时发生并命中同一个数据库。

出于数据完整性的目的，重要的是这个方法中的三个数据库操作都被调用而没有另一个进程读取或写入其间的那些数据库行。

如何确保此方法具有数据库原子性？龙卷风有装饰师吗？

答案

Synchronous database access

您尚未说明如何访问数据库。如果，您可能在get_val_from_db_table1和朋友（例如pymysql）和my_method中有同步数据库访问阻塞（不会将控制权返回到IO循环），那么您将阻止服务器（这会影响服务器的性能和响应能力） ）但有效地序列化您的客户，只有一个可以一次执行my_method。因此，就数据一致性而言，您不需要做任何事情，但通常这是一个糟糕的设计。您可以在短期内使用@ xyres的解决方案解决这两个问题（因为大多数Tornado的功能isn't thread-safe都需要考虑线程安全问题）。

Asynchronous database access

如果您在get_val_from_db_table1和朋友（例如tornado-mysql）中有异步数据库访问权限，那么您可以使用tornado.locks.Lock。这是一个例子：

from tornado import web, gen, locks, ioloop


_lock = locks.Lock()

def synchronised(coro):
    async def wrapper(*args, **kwargs):  
        async with _lock:
            return await coro(*args, **kwargs)

    return wrapper


class MainHandler(web.RequestHandler):

    async def get(self):
        result = await self.my_method('foo')
        self.write(result)

    @classmethod
    @synchronised
    async def my_method(cls, arg):
        # db access
        await gen.sleep(0.5)
        return 'data set for {}'.format(arg)


if __name__ == '__main__':
    app = web.Application([('/', MainHandler)])
    app.listen(8080)
    ioloop.IOLoop.current().start()

---

请注意，上面说的是关于正常的单进程Tornado应用程序。如果你使用tornado.process.fork_processes，那么你只能使用multiprocessing.Lock。

另一答案

由于您希望一个接一个地运行这三个db操作，因此函数my_method必须是非异步的。

但这也意味着my_method将阻止服务器。你绝对不希望如此。我能想到的一种方法是在另一个线程中运行此函数。这不会阻止服务器，并将在操作运行时继续接受新请求。因为，它将是非同步的，db原子性是有保证的。

以下是帮助您入门的相关代码：

import concurrent.futures

executor = concurrent.futures.ThreadPoolExecutor(max_workers=1)
# Don't set `max_workers` more than 1, because then multiple 
# threads will be able to perform db operations

class MyHandler(...):
    @gen.coroutine
    def get(self):

        yield executor.submit(MyHandler.my_method, my_arg1)
        # above, `yield` is used to wait for 
        # db operations to finish
        # if you don't want to wait and return
        # a response immediately remove the 
        # `yield` keyword

        self.write('Done')

    @classmethod
    def my_method(cls, my_arg1):
        # do db stuff ...
        return True