使用MySQL表作为Python中的线程队列

Question

我有一个带有 queue 表的数据库，新条目在队列中不断插入。

我想要一个 Python 脚本尽可能快地执行队列，我认为我需要一些线程化的代码来作为守护进程运行。

但是我不知道如何将数据库用作队列。

我正在查看这个示例：

import MySQLdb
from Queue import Queue
from threading import Thread

def do_stuff(q):
    while True:
        print q.get()
        q.task_done()

q = Queue(maxsize=0)
num_threads = 10

for i in range(num_threads):
    worker = Thread(target=do_stuff, args=(q,))
    worker.setDaemon(True)
    worker.start()

// TODO:  Use the DB
db = MySQLdb.connect(...)
cursor = db.cursor()
q = cursor.execute("SELECT * FROM queue")

for x in range(100):
    q.put(x)
q.join()

- pkdkk

---

我不明白你想用数据库做什么。 无论如何，Python有GIL，这意味着在并行化此操作时，您很可能不会获得任何性能提升。 - mkorpela

Answer 1

两个要点：

1. 假设你正在使用cPython，全局解释器锁（GIL）将有效地使线程无用，一次只允许一个线程通过解释器。有几种解决方法：

   • Gevent库[源代码]

     gevent是一种基于协程的Python网络库，它使用greenlet在libev事件循环之上提供高级同步API。

   • multiprocessing模块，您可以生成多个进程-这是Python中的真正并发。

   • concurrent.futures模块-在python 3中新增，在python 2中有可用端口。[源代码]

     这是一个新的高级库，仅在“作业”级别操作，这意味着您不再需要烦恼
     同步或管理线程或进程。您只需指定具有一定数量的“工作者”的线程或进程池，提交
     作业，并整理结果。它在Python 3.2中是新的，但Python 2.6+的端口可在http://code.google.com/p/pythonfutures上获得。

您可以使用MySQLdb的SSDictCursor（）并执行fetchone()。这是一个流式游标，您可以在一个无限的while（）循环中运行它以类似于队列：

cur = MySQLdb.cursors.SSDictCursor()

cur.execute(query)

while True:

row = cursor.fetchone()

if not row : break # (or sleep()!)

else: # other

2. 尽管如此，我建议您考虑实施像celery或mongodb这样的工具来模拟队列和工作者。关系型数据库不适合处理这种任务，容易出现不必要的碎片化。如果您想了解更多有关mysql中碎片化的信息，可以参考此处。

- Abhishek Pathak

---

gevent 不允许您绕过 GIL，concurrent.futures.ThreadPoolExecutor 也不行。只有 multiprocessing 和 concurrent.futures.ProcessPoolExecutor 可以做到这一点。请注意，对于 I/O 绑定操作，线程和 gevent 仍然很有用，因为执行 I/O 会释放 GIL。 - dano

这是正确的。只有通过生成多个进程，我们才能在这里实现真正的并发。然而，很多时候，我们尝试通过并发来实现的实际上是异步处理，而gevent在这方面非常适用。 - Abhishek Pathak

Answer 2

我不确定这是最好的解决方案，但我考虑了一个主线程的结构，它读取数据库并填充队列。确保避免重复项。也许通过使用递增数字的主键可以轻松检查。

Worker-Structure很好，但正如评论中所提到的：GIL将避免任何提升。但如果您的“do_stuff”与脚本本身无关（例如，任务是图片，“do_stuff”是“将每张图片旋转90°”），则可以使用多进程。据我所知，它不会受到GIL的影响。

https://docs.python.org/2/library/subprocess.html 给出了一些相关信息。

PS：英语不是我的母语。

- Kuishi

---

我认为在这种情况下，“multiprocessing”模块比“subprocess”模块更相关。 - Abhishek Pathak