Быстрая реализация Pube-Sub Python: потоки (II)

Fast Pub Sub (2 серии деталей)

Продолжая улучшать наш простой паб. Теперь мы упустим нить на каждом подписке.

Реализация

class ThreadedEventChannel(EventChannel):
    def __init__(self):
        super(ThreadedEventChannel, self).__init__()

    def publish(self, event, *args, **kwargs):
        threads = []
        if event in self.subscribers.keys():
            for callback in self.subscribers[event]:
                threads.append(threading.Thread(
                  target=callback,
                  args=args,
                  kwargs=kwargs
                ))

            for th in threads:
                th.start()

            for th in threads:
                th.join()

Этот класс запустит все резьбы, подписанные и подожду, они все будут продолжать выполнять

Создание издателя не блокирует

class ThreadedEventChannel(EventChannel):
    def __init__(self, blocking=True):
        self.blocking = blocking
        super(ThreadedEventChannel, self).__init__()

    def publish(self, event, *args, **kwargs):
        threads = []
        if event in self.subscribers.keys():
            for callback in self.subscribers[event]:
                threads.append(threading.Thread(
                  target=callback,
                  args=args,
                  kwargs=kwargs
                ))
            for th in threads:
                th.start()

            if self.blocking:
                for th in threads:
                    th.join()

Этот класс запустит все подписанные нити и нет Подождите, чтобы они продолжили выполнять

Преимущества этого метода

Мы делаем вызовы параллель, например, улучшит нашу производительность, если событие стреляет 2 длинным обратным вызовам, мы ждем только самых длинных времен наших обратных вызовов.

import time
from event_channel.event_channel import EventChannel
from event_channel.threaded_event_channel import ThreadedEventChannel

non_thread = EventChannel()
threaded = ThreadedEventChannel()
non_blocking_threaded = ThreadedEventChannel(blocking=False)

non_thread.subscribe("myevent", time.sleep)
non_thread.subscribe("myevent", time.sleep)
start = time.time()
non_thread.publish("myevent", 3)
end = time.time()
print("non threaded function elapsed time {0}".format(end - start))
#non threaded function elapsed time 6.0080871582
threaded.subscribe("myevent", time.sleep)
threaded.subscribe("myevent", time.sleep)
start = time.time()
threaded.publish("myevent", 3)
end = time.time()
print("threaded function elapsed time {0}".format(end - start))
# threaded function elapsed time 3.00581121445

non_blocking_threaded.subscribe("myevent", time.sleep)
non_blocking_threaded.subscribe("myevent", time.sleep)
start = time.time()
non_blocking_threaded.publish("myevent", 3)
end = time.time()
print("threaded function non blocking elapsed time {0}".format(end - start))
# threaded function non blocking elapsed time 0.00333380699158

Ссылки

Что вы улучшите?

Fast Pub Sub (2 серии деталей)

Оригинал: “https://dev.to/mandrewcito/fast-pub-sub-python-implementation-threading-ii-1khp”