无法使用两个进程流式传输到具有SQLAlchemy的数据库
无法使用两个进程流式传输到具有SQLAlchemy的数据库
提问于 2019-07-21 23:35:49
我有两个while True循环，每个循环都从一个外部API读取一些数据。
db = Database(env='dev')
aStream = AStream(db)
proc1 = Process(target=aStream.stream_a_to_db())
proc1.start()
bStream = BStream(db)
proc2 = Process(target=bStream.stream_b_to_db())
proc2.start()我的Database类如下所示
class Database:
def __init__(self, env='dev'):
"""DB setup"""
self.db_url = self._set_db_url_by_env(env)
self.engine = create_engine(self.db_url, echo=True)
def create_db_session(self):
# Create all tables that do not already exist
Base.metadata.create_all(self.engine, Base.metadata.tables.values(), checkfirst=True)
# SqlAlchemy :: Session setup
Session = sessionmaker(bind=self.engine)
# SqlAlchemy :: Starts a session
return Session()我不明白运行这段代码的结果。如果我把proc2放在proc1之前，那么到db的proc2流就是唯一运行的。
我试过两件事。
- 使用伪函数而不是实际的流。
def func1():
print("func1 up and running.")
def func2():
print("func2 up and running.")
proc1 = Process(target=func1)
proc1.start()
proc2 = Process(target=func2)
proc2.start()这将按预期运行，func1和func2都在运行和打印。
- 只需将
bStream移动到一个单独的python文件中，并在两个终端选项卡中手动运行两个文件，python a_stream.py和python b_stream.py，而不需要对其他代码进行任何更改。它们各自都运行得很好，都将数据流式传输到数据库中，没有任何问题。
基本上，我的目标不是拥有两个脚本并手动运行这两个脚本，而是拥有一个包含两个进程的脚本。这样做的正确方法是什么？这一定与SQLAlchemy引擎/会话的工作方式有关。我是SQLAlchemy的新手。感谢这里的任何帮助！
回答 1
Stack Overflow用户
回答已采纳
发布于 2019-07-21 23:56:39
每个the docs
对于使用os.fork系统调用的多进程应用程序，或者例如Python多处理模块，通常需要为每个子进程使用单独的引擎。这是因为引擎维护了对最终引用DBAPI连接的连接池的引用-这些连接往往不能跨进程边界移植。配置为不使用池化(通过使用NullPool实现)的引擎没有此要求。
因此，不是在主流程中创建引擎并将其传递给两个子流程，而是在每个子流程中创建一个新引擎：
def worker(Stream, methodname):
db = Database(env='dev')
stream = Stream(db)
getattr(stream, methodname)()
args = [(AStream, 'stream_a_to_db'), (BStream, 'stream_b_to_db')]
procs = [Process(target=worker, args=a) for a in args]
for proc in procs:
proc.start()
for proc in procs:
proc.join()页面原文内容由Stack Overflow提供。腾讯云小微IT领域专用引擎提供翻译支持
原文链接：
https://stackoverflow.com/questions/57134630
复制相关文章
相似问题