我很难理解如何有效地打开和关闭数据库会话,正如sqlalchemy文档所理解的,如果我使用scoped_session来构造我的会话对象,然后使用返回的session对象来创建会话,它是线程安全的,所以基本上每个线程都会得到自己的会话,并且不会有有问题。下面的示例可以工作了,我将它放在一个无限循环中,看看它是否正确地关闭了会话,如果我正确地监视了它(在mysql中,通过执行“SHOW PROCESSLIST;”),连接只会继续增长,它不会关闭它们,即使我使用了session.close(),甚至在每次运行结束时删除作用域的会话对象。我做错什么了?我在大型应用程序中的目标是使用所需的最少数据库连接数,因为我当前的工作实现在每个需要会话的方法中创建一个新会话,并在返回之前关闭它,这看起来效率很低。
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, scoped_session
from threading import Thread
from Queue import Queue, Empty as QueueEmpty
from models import MyModel
DATABASE_CONNECTION_INFO = 'mysql://username:password@localhost:3306/dbname'
class MTWorker(object):
def __init__(self, worker_count=5):
self.task_queue = Queue()
self.worker_count = worker_count
self.db_engine = create_engine(DATABASE_CONNECTION_INFO, echo=False)
self.DBSession = scoped_session(
sessionmaker(
autoflush=True,
autocommit=False,
bind=self.db_engine
)
)
def _worker(self):
db_session = self.DBSession()
while True:
try:
task_id = self.task_queue.get(False)
try:
item = db_session.query(MyModel).filter(MyModel.id == task_id).one()
# do something with item
except Exception as exc:
# if an error occurrs we skip it
continue
finally:
db_session.commit()
self.task_queue.task_done()
except QueueEmpty:
db_session.close()
return
def start(self):
try:
db_session = self.DBSession()
all_items = db_session.query(MyModel).all()
for item in all_items:
self.task_queue.put(item.id)
for _i in range(self.worker_count):
t = Thread(target=self._worker)
t.start()
self.task_queue.join()
finally:
db_session.close()
self.DBSession.remove()
if __name__ == '__main__':
while True:
mt_worker = MTWorker(worker_count=50)
mt_worker.start()
您应该每次只调用
create_engine
和scoped_session
一次 进程(每个数据库)。每个人都有自己的连接或会话池 (分别),所以您要确保只创建一个池。只是让它成为模块级的全局。如果需要更精确地管理会话,则可能不应该使用scoped_session
另一个改变是直接使用
DBSession
,就好像它是 会议。在作用域会话上调用会话方法将透明 如果需要,创建一个线程本地会话,并将方法调用转发给 会议。另一件需要注意的事情是 ^{}
连接池的
默认为5。对于许多应用程序来说这很好,但是如果您正在创建
很多线程,您可能需要调整该参数
相关问题 更多 >
编程相关推荐