在 Linux 上使用 Python 3.x 的多处理生产者消费者

Question

下面的代码模拟生产者和消费者模型，它将从外汇经纪商 FXCM 收集数据并写入数据库。每个生产者进程都将与代理建立基于会话的连接。

生产者和消费者都将运行无限期地 'poison pill' 放入队列中，这发生在营业结束时（星期五 22:00）。我省略了这部分代码，因为它与问题无关。我能找到的所有示例似乎都产生了一个进程，在短时间内做了一些工作，然后 join() 返回到父进程。喜欢这个here

如上所述，生产者将运行无限期，这是因为登录并与代理创建会话大约需要 3 秒。

当运行执行下面的代码时，您会看到队列积压，尽管运行执行实际代码时这似乎更糟。

不确定它是否相关，但会话是使用 python-forexconnect API 创建的，它是用 C++ 编写并使用 Boost。

问题是消费者从队列中 get() 项目花费的时间太长，我想知道这个模型是否是处理此类开发的正确方法。

谢谢

示例代码

from multiprocessing import Process, Queue, cpu_count
from datetime import datetime, timedelta

import numpy as np
import time

def dummy_data(dtto):
    dates = np.array(
          [dtto - timedelta(days=i) for i in range(300)])
    price_data = np.random.rand(len(dates),5)
    return np.concatenate(  
      (np.vstack(dates),price_data), axis=1)

def get_bars(q2, ms, symbol, dtfm, dtto, time_frame):
    stop_date = dtfm
    while dtto > stop_date:
        data = dummy_data(dtto)
        dtfm = data[-1,0]
        dtto = data[0,0]
        q2.put((symbol, dtfm, dtto))
        # Switch to date
        dtto = dtfm

def producer(q1,q2):  
    # client = fx.Client(....)
    client = 'broker session'
    while True:
        job = q1.get()
        if job == None:
            break
        sym, dtfm, dtto, tf = job
        # Get price data from broker
        get_bars(q2, client, sym, dtfm, dtto, tf)
    q2.put(None)

def consumer(q2):
    while True:
        bars = q2.get()
        if bars == None:
            break
        print(q2.qsize(), bars[0], bars[1], bars[2]) # write to db

q1, q2 = Queue(), Queue()
# instruments = client.get_offers()
# instruments = ['GBP/USD', 'EUR,USD',...]
instruments = range(63) # 62 dummy instruments
# Places jobs into the queue for each symbol
for symbol in instruments:
    q1.put((symbol,
        datetime(2000,1,14,22,0),
        datetime(2018,1,14,22,0),
        'D1'))
# Setup producers and consumers
pp, cp = range(6), range(2)
pro = [Process(target=producer, args=(q1, q2,)) for i in pp]
con = [Process(target=consumer, args=(q2,)) for i in cp]
for p in pro: p.start()
for p in con: p.start()
# This is just here to stop this script and does not
# exist in the real version
for i in pp: q1.put(None)
for p in pro: p.join()
for p in con: p.join()
print('stopped')

Answer 1

multiprocessing.Queue.get() 的糟糕表现是一个已知问题（在 Whosebug 上也有几个问题，但没有通常有用的答案）。

哪种情况表明您应该考虑其他型号。您可以看到与此相比有多少进程创建开销；根本不要永久使用运行进程，而是在准备好数据后立即启动进程。当您这样做时，当您的进程分叉时，您的子进程将收到一份内存中的数据副本。这增加了进程创建开销但删除了队列。您至少可以考虑这一点，因为您的消费者写入数据库并且不需要向父级报告任何内容。

Python 是一种很棒的语言，但在并行处理方面它并不是最好的语言。

在 Linux 上使用 Python 3.x 的多处理生产者消费者

Multiprocessing Producer Consumer with Python 3.x on Linux

python

producer-consumer

multiprocessing

python-multiprocessing