Multiplex on queue.Queue?

Ответ 1

Существует множество различных реализаций очередей производителей-потребителей, таких как queue.Queue. Обычно они отличаются множеством свойств, как указано в этой отличной статье Дмитрия Вюкова. Как вы можете видеть, существует более 10 000 различных комбинаций. Алгоритмы, используемые для таких очередей, также сильно различаются в зависимости от требований. Невозможно просто расширить существующий алгоритм очереди, чтобы гарантировать дополнительные свойства, поскольку для этого обычно требуются разные внутренние структуры данных и разные алгоритмы.

Go предлагают относительно высокое количество гарантированных свойств, поэтому эти каналы могут быть пригодны для множества программ. Один из самых сложных требований - поддержка чтения/блокировки сразу по нескольким каналам (оператор выбора) и выбор канала справедливо, если более чем одна ветвь в операторе select может продолжать работу, чтобы сообщения не оставались позади, Python queue.Queue не предлагает этих функций, поэтому просто невозможно архивировать то же поведение с ним.

Итак, если вы хотите продолжить использование queue.Queue, вам необходимо найти обходные пути для этой проблемы. Однако обходные пути имеют собственный список недостатков и их сложнее поддерживать. Поиск другой очереди производителей-потребителей, которая предлагает нужные вам функции, может быть лучшей идеей! Во всяком случае, вот два возможных способа обхода:

опрос

while True:
  try:
    i1 = c1.get_nowait()
    print "received %s from c1" % i1
  except queue.Empty:
    pass
  try:
    i2 = c2.get_nowait()
    print "received %s from c2" % i2
  except queue.Empty:
    pass
  time.sleep(0.1)

Это может использовать много циклов процессора во время опроса каналов и может быть медленным, когда есть много сообщений. Использование time.sleep() с экспоненциальным временем отсрочки (вместо показанных здесь констант 0,1 с) может значительно улучшить эту версию.

Единая очередь уведомлений

queue_id = notify.get()
if queue_id == 1:
  i1 = c1.get()
  print "received %s from c1" % i1
elif queue_id == 2:
  i2 = c2.get()
  print "received %s from c2" % i2

С помощью этой настройки вы должны отправить что-то в очередь уведомлений после отправки на c1 или c2. Это может сработать для вас, если для вас достаточно только одной очереди уведомлений (т.е. У вас нет нескольких "выборок", каждая из которых блокируется другим подмножеством ваших каналов).

В качестве альтернативы вы также можете использовать Go. Go goroutines и concurrency намного мощнее, чем Python, ограниченные возможности потоковой передачи.

Ответ 2

Если вы используете queue.PriorityQueue, вы можете получить аналогичное поведение, используя объекты канала в качестве приоритетов:

import threading, logging
import random, string, time
from queue import PriorityQueue, Empty
from contextlib import contextmanager

logging.basicConfig(level=logging.NOTSET,
                    format="%(threadName)s - %(message)s")

class ChannelManager(object):
    next_priority = 0

    def __init__(self):
        self.queue = PriorityQueue()
        self.channels = []

    def put(self, channel, item, *args, **kwargs):
        self.queue.put((channel, item), *args, **kwargs)

    def get(self, *args, **kwargs):
        return self.queue.get(*args, **kwargs)

    @contextmanager
    def select(self, ordering=None, default=False):
        if default:
            try:
                channel, item = self.get(block=False)
            except Empty:
                channel = 'default'
                item = None
        else:
            channel, item = self.get()
        yield channel, item


    def new_channel(self, name):
        channel = Channel(name, self.next_priority, self)
        self.channels.append(channel)
        self.next_priority += 1
        return channel


class Channel(object):
    def __init__(self, name, priority, manager):
        self.name = name
        self.priority = priority
        self.manager = manager

    def __str__(self):
        return self.name

    def __lt__(self, other):
        return self.priority < other.priority

    def put(self, item):
        self.manager.put(self, item)


if __name__ == '__main__':
    num_channels = 3
    num_producers = 4
    num_items_per_producer = 2
    num_consumers = 3
    num_items_per_consumer = 3

    manager = ChannelManager()
    channels = [manager.new_channel('Channel#{0}'.format(i))
                for i in range(num_channels)]

    def producer_target():
        for i in range(num_items_per_producer):
            time.sleep(random.random())
            channel = random.choice(channels)
            message = random.choice(string.ascii_letters)
            logging.info('Putting {0} in {1}'.format(message, channel))
            channel.put(message)

    producers = [threading.Thread(target=producer_target,
                                  name='Producer#{0}'.format(i))
                 for i in range(num_producers)]
    for producer in producers:
        producer.start()
    for producer in producers:
        producer.join()
    logging.info('Producers finished')

    def consumer_target():
        for i in range(num_items_per_consumer):
            time.sleep(random.random())
            with manager.select(default=True) as (channel, item):
                if channel:
                    logging.info('Received {0} from {1}'.format(item, channel))
                else:
                    logging.info('No data received')

    consumers = [threading.Thread(target=consumer_target,
                                  name='Consumer#{0}'.format(i))
                 for i in range(num_consumers)]
    for consumer in consumers:
        consumer.start()
    for consumer in consumers:
        consumer.join()
    logging.info('Consumers finished')

Пример вывода:

Producer#0 - Putting x in Channel#2
Producer#2 - Putting l in Channel#0
Producer#2 - Putting A in Channel#2
Producer#3 - Putting c in Channel#0
Producer#3 - Putting z in Channel#1
Producer#1 - Putting I in Channel#1
Producer#1 - Putting L in Channel#1
Producer#0 - Putting g in Channel#1
MainThread - Producers finished
Consumer#1 - Received c from Channel#0
Consumer#2 - Received l from Channel#0
Consumer#0 - Received I from Channel#1
Consumer#0 - Received L from Channel#1
Consumer#2 - Received g from Channel#1
Consumer#1 - Received z from Channel#1
Consumer#0 - Received A from Channel#2
Consumer#1 - Received x from Channel#2
Consumer#2 - Received None from default
MainThread - Consumers finished

В этом примере ChannelManager представляет собой всего лишь оболочку вокруг queue.PriorityQueue, которая реализует метод select как contextmanager, чтобы он выглядел аналогично оператору select в Go.

Несколько замечаний:

• Сортировка

  • В примере Go порядок, в котором каналы записываются внутри оператора select, определяет, какой код канала будет выполнен, если имеются данные для более чем одного канала.

  • В примере на python порядок определяется приоритетом, назначенным каждому каналу. Тем не менее, приоритет может быть динамически назначен каждому каналу (как показано в примере), поэтому изменение порядка возможно с помощью более сложного метода select, который заботится о назначении новых приоритетов на основе аргумента метода. Кроме того, старый менеджер может быть восстановлен после завершения менеджера контекста.

• Блокировка

  • В примере Go оператор select блокируется, если существует случай default.

  • В примере на python логический аргумент должен быть передан методу select, чтобы очистить его, когда требуется блокировка/неблокирование. В неблокирующем случае канал, возвращаемый манаганом контекста, является просто строкой 'default', поэтому в коде внутри легко обнаружить это в коде внутри оператора with.

• Threading: объект в модуле queue уже готов к многопроцессорным сценариям с несколькими контентом, как уже было показано в этом примере.

Ответ 3

Проект pychan дублирует каналы Go на Python, включая мультиплексирование. Он реализует тот же алгоритм, что и Go, поэтому он отвечает всем вашим желаемым свойствам:

• Несколько производителей и потребителей могут общаться через Chan. Когда производитель и потребитель готовы, пара из них заблокирует
• Производители и потребители обслуживаются в том порядке, в котором они прибыли (FIFO)
• Пустая (полная) очередь блокирует потребителей (производителей).

Здесь будет выглядеть ваш пример:

c1 = Chan(); c2 = Chan(); c3 = Chan()

try:
    chan, value = chanselect([c1, c3], [(c2, i2)])
    if chan == c1:
        print("Received %r from c1" % value)
    elif chan == c2:
        print("Sent %r to c2" % i2)
    else:  # c3
        print("Received %r from c3" % value)
except ChanClosed as ex:
    if ex.which == c3:
        print("c3 is closed")
    else:
        raise

(Полное раскрытие: я написал эту библиотеку)

Ответ 4

from queue import Queue

# these imports needed for example code
from threading import Thread
from time import sleep
from random import randint

class MultiQueue(Queue):

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.queues = []

    def addQueue(self, queue):
        queue.put = self._put_notify(queue, queue.put)
        queue.put_nowait = self._put_notify(queue, queue.put_nowait)
        self.queues.append(queue)

    def _put_notify(self, queue, old_put):
        def wrapper(*args, **kwargs):
            result = old_put(*args, **kwargs)
            self.put(queue)
            return result
        return wrapper

if __name__ == '__main__':
    # an example of MultiQueue usage

    q1 = Queue()
    q1.name = 'q1'
    q2 = Queue()
    q2.name = 'q2'
    q3 = Queue()
    q3.name = 'q3'

    mq = MultiQueue()
    mq.addQueue(q1)
    mq.addQueue(q2)
    mq.addQueue(q3)

    queues = [q1, q2, q3]
    for i in range(9):
        def message(i=i):
            print("thread-%d starting..." % i)
            sleep(randint(1, 9))
            q = queues[i%3]
            q.put('thread-%d ending...' % i)
        Thread(target=message).start()

    print('awaiting results...')
    for _ in range(9):
        result = mq.get()
        print(result.name)
        print(result.get())

Вместо того, чтобы пытаться использовать метод .get() нескольких очередей, идея здесь состоит в том, чтобы очереди уведомляли MultiQueue, когда у них есть готовые данные, - сортировка a select в обратном порядке. Это достигается тем, что MultiQueue обертывает различные методы Queue put() и put_nowait(), так что когда что-то добавляется в эти очереди, эта очередь затем put() в MultiQueue и соответствующая MultiQueue.get() получит Queue, у которого есть готовые данные.

Этот MultiQueue основан на очереди FIFO, но вы также можете использовать очереди LIFO или Priority в качестве базы в зависимости от ваших потребностей.