【问题标题】：Accessing a database from a worker process从工作进程访问数据库
【发布时间】：2014-06-26 06:04:56
【问题描述】：

我正在尝试使用 Python、Redis 和 PostgreSQL 放置需要访问数据库的作业。我正在执行以下操作：

使用 RQ 将作业放入 Redis 队列：

def queue_data(self, json_data):
    Queue().enqueue(process_data, json.dumps(json_data))

在process_data 中，使用 psycopg2 对 PostgreSQL 数据库执行查询：

def process_data(json_data):
    with psycopg2.connect("dbname=pgtest2db user=pgtest2user") as conn:
        with conn.cursor() as cursor:
            # I'm not actually doing this query, but you get the idea
            cursor.execute("SELECT * FROM some_table")
    conn.close()

显然这远非最佳，因为每次调用process_data 都会创建一个新连接。实现这一目标的最佳实践是什么？谁应该负责管理数据库连接池？

请注意，我强迫自己不要使用 ORM，因为我这样做是为了自学，我想从更纯粹的角度理解这些模式。

编辑

我最终使用了像这样的自定义工作者：

import os
import node

import redis
from rq import Worker, Queue, Connection
from psycopg2.pool      import ThreadedConnectionPool

listen = ['high', 'default', 'low']
redis_url = os.getenv('REDISTOGO_URL', 'redis://localhost:6379')
redis_conn = redis.from_url(redis_url)
pool = ThreadedConnectionPool(minconn = 1, maxconn = 10, dsn = "dbname=pgtest2db user=pgtest2user")

def process_data(json_data):
    dbconn = pool.getconn()
    result = perform_some_db_actions(dbconn, json_data)
    pool.putconn(dbconn)
    return result

if __name__ == '__main__':
    with Connection(redis_conn):
        print "Setting up Redis"
        worker = Worker(map(Queue, listen))
        worker.work()

【问题讨论】：

psychopg2.pool 不包含连接池实现吗？
我确实需要一个游泳池，但谁来管理它？我无法通过队列传递光标或连接，因为这些值不能“腌制”，即以序列化方式表示。
池的意义在于你不需要腌制它。你可以简单地要求一个“新的”，你会得到一个现有的。但是，您似乎不仅想重用连接，还想重用光标状态。我说的对吗？
你用什么python模块来和redis交互？
@jjanes 抱歉我错过了，我正在使用RQ

标签： python postgresql redis psycopg2

【解决方案1】：

我知道现在为时已晚，但是每个想知道如何在 RQ 中使用 psycopg2 的人（尤其是当您遇到 SSL 错误时：解密失败或坏记录 mac - 只需打开连接 inside worker（即任务。最好是创建单例而不是每次都打开连接）。不是在worker.work()函数之前。

【讨论】：