如何在python代碼內(nèi)部將獲取的數(shù)據(jù)分流？

比如我使用kafka或rabbitmq從消息隊(duì)列中獲取數(shù)據(jù)，來源只有一條隊(duì)列，不能從數(shù)據(jù)源分流數(shù)據(jù)。而計(jì)算難度較高，所以要在內(nèi)部進(jìn)行分流。比如我的消息處理類為：

class Worker(threading.Thread):
    def __init__(self):
        self.raw = []
    
    def run(self):
        while True:
            if self.raw:
                d = self.raw.pop()
                處理d數(shù)據(jù)
                將結(jié)果保存到批量插入的類中

然后在程序運(yùn)行時(shí)，我創(chuàng)建若干個(gè)Worker，然后將消息源傳入的數(shù)據(jù)，分流保存到這若干個(gè)Worker的raw屬性中。問題就在于，如何能開銷比較低的分流這些數(shù)據(jù)？
還有這樣多線程處理的思路是否正確？我之前測試過，感覺python自帶的Queue的效率并不是特別高。
另外，假如我要設(shè)計(jì)彈性創(chuàng)建Worker，應(yīng)該如何用python代碼完成，就是當(dāng)數(shù)據(jù)流較大時(shí)，我就追加創(chuàng)建一些Worker，當(dāng)數(shù)據(jù)流降低時(shí)，就銷毀一些Worker。

回答

編輯回答

夏木

如果確定是計(jì)算密集確實(shí)不適合使用python中的多線程，但是是可以考慮使用多進(jìn)程的。你不需要通過自己創(chuàng)建一個(gè)queue來進(jìn)行內(nèi)部分流，即使需要一個(gè)Queue, 也是需要通過給Queue設(shè)置大小來限制Queue的流量。

以rabbitmq為例，請看https://www.rabbitmq.com/tuto...

在rabbitmq的官方例子中，是使用pika做為rabbitmq的客戶端的，消息模型應(yīng)該是和你的是一致的，稍微修改一下官方的work.py例子，通過建立多個(gè)rabbitmq客戶端來消費(fèi)消息：

#!/usr/bin/env python
import pika
import time
from concurrent.futures import ProcessPoolExecutor
# from concurrent.futures import ThreadPoolExecutor


connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))

channels = [
    (1, connection.channel(1)),
    (2, connection.channel(2)),
    (3, connection.channel(3)),
]

def callback(ch, method, properties, body):
    print(" [x] Received %r" % body)
    time.sleep(0.2)
    print(" [x] Done: %s" % ch.channel_number)
    ch.basic_ack(delivery_tag=method.delivery_tag)


for _, channel in channels:
    channel.queue_declare(queue='task_queue', durable=True)
    channel.basic_qos(prefetch_count=1)
    channel.basic_consume(callback, queue='task_queue')


def start_consumer(channel_name):
    dict(channels).get(channel_name).start_consuming()


print(' [*] Waiting for messages. To exit press CTRL+C')

with ProcessPoolExecutor(max_workers=len(channels)) as e:
    e.map(start_consumer, range(1, 4))

# with ThreadPoolExecutor(max_workers=len(channels)) as e:
#     e.map(start_consumer, range(1, 4))

彈性的創(chuàng)建worker我覺的從程序(worker.py)內(nèi)部去實(shí)現(xiàn)是比較困難的，從程序外部來看更容易實(shí)現(xiàn)，首先監(jiān)控流量，流量增大可以通過啟動(dòng)更多的worker.py腳本來加快消息的消費(fèi)；反之，減少worker.py啟動(dòng)的數(shù)量。

2017年11月9日 21:20

編輯回答