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

Question

С ZeroMQ для распределенного обмена сообщениями я использую пример кода, предоставленного для параноидального пиратского паттерна в python. У меня есть один клиент (потенциально может быть больше клиентов), брокер и несколько работников.

Я изменил пример так, что клиент будет продолжать отправлять запросы в очередь посредника, даже если нет доступных рабочих, вместо того, чтобы повторять попытки и в конечном итоге завершать работу. Они будут переданы работникам, когда они в конечном итоге станут доступными. В моем сценарии каждый работник различается по количеству времени, которое требуется для обработки данного запроса.

Проблема, которую я вижу, состоит в том, что, когда брокер выходит из строя (становится недоступным), клиент не может сказать, что брокер недоступен, и он продолжает .send() Запросы. Эти запросы теряются. Только новые запросы обрабатываются после того, как брокер снова становится доступным.

Client.py

from random import randint
import time

import zmq

HEARTBEAT_LIVENESS = 3
HEARTBEAT_INTERVAL = 1
INTERVAL_INIT = 1
INTERVAL_MAX = 32

#  Paranoid Pirate Protocol constants
PPP_READY = "PPP_READY"      # Signals worker is ready
PPP_HEARTBEAT = "PPP_HEARTBEAT"  # Signals worker heartbeat

def worker_socket(context, poller):
    """Helper function that returns a new configured socket
       connected to the Paranoid Pirate queue"""
    worker = context.socket(zmq.DEALER) # DEALER
    identity = "%04X-%04X" % (randint(0, 0x10000), randint(0, 0x10000))
    worker.setsockopt(zmq.IDENTITY, identity)
    poller.register(worker, zmq.POLLIN)
    worker.connect("tcp://localhost:5556")
    worker.send(PPP_READY)
    return worker

context = zmq.Context(1)
poller = zmq.Poller()

liveness = HEARTBEAT_LIVENESS
interval = INTERVAL_INIT

heartbeat_at = time.time() + HEARTBEAT_INTERVAL

worker = worker_socket(context, poller)
cycles = 0
while True:
    socks = dict(poller.poll(HEARTBEAT_INTERVAL * 1000))

    # Handle worker activity on backend
    if socks.get(worker) == zmq.POLLIN:
        #  Get message
        #  - 3-part envelope + content -> request
        #  - 1-part HEARTBEAT -> heartbeat
        frames = worker.recv_multipart()
        if not frames:
            break # Interrupted

        if len(frames) == 3:
            print "I: Normal reply: ", frames
            liveness = HEARTBEAT_LIVENESS
            time.sleep(4)  # Do some heavy work
            worker.send_multipart(frames)
        elif len(frames) == 1 and frames[0] == PPP_HEARTBEAT:
           # print "I: Queue heartbeat"
            liveness = HEARTBEAT_LIVENESS
        else:
            print "E: Invalid message: %s" % frames
        interval = INTERVAL_INIT
    else:
        liveness -= 1
        if liveness == 0:
            print "W: Heartbeat failure, can't reach queue"
            print ("W: Reconnecting in")
            time.sleep(interval)

            if interval < INTERVAL_MAX:
                interval *= 2
            poller.unregister(worker)
            worker.setsockopt(zmq.LINGER, 0)
            worker.close()
            worker = worker_socket(context, poller)
            liveness = HEARTBEAT_LIVENESS
    if time.time() > heartbeat_at:
        heartbeat_at = time.time() + HEARTBEAT_INTERVAL
        #print "I: Worker heartbeat"
        worker.send(PPP_HEARTBEAT)

Broker.py

from collections import OrderedDict
import time
import threading
import zmq

HEARTBEAT_LIVENESS = 3     # 3..5 is reasonable
HEARTBEAT_INTERVAL = 1.0   # Seconds

#  Paranoid Pirate Protocol constants
PPP_READY = "PPP_READY"      # Signals worker is ready
PPP_HEARTBEAT = "PPP_HEARTBEAT"  # Signals worker heartbeat
PPP_BUSY = "PPP_BUSY"
PPP_FREE = "PPP_FREE"

class Worker(object):
    def __init__(self, address):
        self.address = address
        self.expiry = time.time() + HEARTBEAT_INTERVAL * HEARTBEAT_LIVENESS

class WorkerQueue(object):
    def __init__(self):
        self.queue = OrderedDict()

    def ready(self, worker):
        self.queue.pop(worker.address, None)
        self.queue[worker.address] = worker

    def purge(self):
        """Look for & kill expired workers."""
        t = time.time()
        expired = []
        for address,worker in self.queue.iteritems():
            if t > worker.expiry:  # Worker expired
                expired.append(address)
        for address in expired:
            print "W: Idle worker expired: %s" % address
            self.queue.pop(address, None)

    def next(self):
        address, worker = self.queue.popitem(False)
        return address

context = zmq.Context(1)
clcontext = zmq.Context()

frontend = context.socket(zmq.ROUTER) # ROUTER
backend = context.socket(zmq.ROUTER)  # ROUTER
frontend.bind("tcp://*:5555") # For clients
backend.bind("tcp://*:5556")  # For workers

poll_workers = zmq.Poller()
poll_workers.register(backend, zmq.POLLIN)

poll_both = zmq.Poller()
poll_both.register(frontend, zmq.POLLIN)
poll_both.register(backend, zmq.POLLIN)

workers = WorkerQueue()

heartbeat_at = time.time() + HEARTBEAT_INTERVAL

while True:
    if len(workers.queue) > 0:
        poller = poll_both
    else:
        poller = poll_workers

    socks = dict(poller.poll(HEARTBEAT_INTERVAL * 1000))

    # Handle worker activity on backend
    if socks.get(backend) == zmq.POLLIN:
        # Use worker address for LRU routing
        frames = backend.recv_multipart()
        if not frames:
            break

        address = frames[0]
        workers.ready(Worker(address))

        # Validate control message, or return reply to client
        msg = frames[1:]
        if len(msg) == 1:
            if msg[0] not in (PPP_READY, PPP_HEARTBEAT):
                print "E: Invalid message from worker: %s" % msg
        else:
            print ("sending: %s"%msg)
            frontend.send_multipart(msg)

        # Send heartbeats to idle workers if it's time
        if time.time() >= heartbeat_at:
            for worker in workers.queue:
                msg = [worker, PPP_HEARTBEAT]
                backend.send_multipart(msg)
            heartbeat_at = time.time() + HEARTBEAT_INTERVAL

    if socks.get(frontend) == zmq.POLLIN:
        frames = frontend.recv_multipart()
        print ("client frames: %s" % frames)
        if not frames:
            break

        frames.insert(0, workers.next())
        backend.send_multipart(frames)

    workers.purge()

Worker.py

from random import randint
import time

import zmq

HEARTBEAT_LIVENESS = 3
HEARTBEAT_INTERVAL = 1
INTERVAL_INIT = 1
INTERVAL_MAX = 32

#  Paranoid Pirate Protocol constants
PPP_READY = "PPP_READY"      # Signals worker is ready
PPP_HEARTBEAT = "PPP_HEARTBEAT"  # Signals worker heartbeat

def worker_socket(context, poller):
    """Helper function that returns a new configured socket
       connected to the Paranoid Pirate queue"""
    worker = context.socket(zmq.DEALER) # DEALER
    identity = "%04X-%04X" % (randint(0, 0x10000), randint(0, 0x10000))
    worker.setsockopt(zmq.IDENTITY, identity)
    poller.register(worker, zmq.POLLIN)
    worker.connect("tcp://localhost:5556")
    worker.send(PPP_READY)
    return worker

context = zmq.Context(1)
poller = zmq.Poller()

liveness = HEARTBEAT_LIVENESS
interval = INTERVAL_INIT

heartbeat_at = time.time() + HEARTBEAT_INTERVAL

worker = worker_socket(context, poller)
cycles = 0
while True:
    socks = dict(poller.poll(HEARTBEAT_INTERVAL * 1000))

    # Handle worker activity on backend
    if socks.get(worker) == zmq.POLLIN:
        #  Get message
        #  - 3-part envelope + content -> request
        #  - 1-part HEARTBEAT -> heartbeat
        frames = worker.recv_multipart()
        if not frames:
            break # Interrupted

        if len(frames) == 3:
            print "I: Normal reply: ", frames
            liveness = HEARTBEAT_LIVENESS
            time.sleep(4)  # Do some heavy work
            worker.send_multipart(frames)
        elif len(frames) == 1 and frames[0] == PPP_HEARTBEAT:
           # print "I: Queue heartbeat"
            liveness = HEARTBEAT_LIVENESS
        else:
            print "E: Invalid message: %s" % frames
        interval = INTERVAL_INIT
    else:
        liveness -= 1
        if liveness == 0:
            print "W: Heartbeat failure, can't reach queue"
            print ("W: Reconnecting in")
            time.sleep(interval)

            if interval < INTERVAL_MAX:
                interval *= 2
            poller.unregister(worker)
            worker.setsockopt(zmq.LINGER, 0)
            worker.close()
            worker = worker_socket(context, poller)
            liveness = HEARTBEAT_LIVENESS
    if time.time() > heartbeat_at:
        heartbeat_at = time.time() + HEARTBEAT_INTERVAL
        #print "I: Worker heartbeat"
        worker.send(PPP_HEARTBEAT)

Answer 1

Добавьте сигнализацию HeartBeat к контрольному слою

Если есть только

(cit.:) " Проблема, с которой я сталкиваюсь, заключается в том, что когда брокер выходит из строя (становится недоступным), клиент не может сказать, что брокер недоступен, и продолжает отправлять запросы".

просто добавьте тривиальную передачу сигналов (будь то PUB/SUB или другой архетип), который может сохранять только последнее сигнальное присутствие с меткой времени с использованием более нового API .setsockopt( zmq.CONFLATE ) или иметь еще более надежное двустороннее рукопожатие, которое позволяет отправителю получить разумное допущение с правом на желание отправить следующее сообщение процессу потребителя, что все заинтересованные стороны находятся в состоянии и находятся в таком состоянии, которое позволяет достичь целевая функциональность.

В этом смысле ZeroMQ является отличным средством, помогающим интегрировать как службы интеллектуальной сигнализации, так и службы обработки сообщений в нечто вроде полностью распределенных конечных состояний-автоматов или сети FSA-устройств, взаимодействующих по состоянию, если это необходимо, действительно замечательный шаг в гетерогенные распределенные системы (очереди, являющиеся некоторыми неосновными, дополнительными средствами для того, чтобы сделать это умным, быстрым и масштабируемым способом).

Вскоре можно осознать дополнительные риски распределенной взаимной блокировки, которые должны быть включены в проекты и проверки профессиональных систем. Этот новый тип проблем может возникать и возникает как в наивном REQ/REP через подверженную ошибкам доставку сообщений и, более сложными способами, в более сложных распределенных сетях сигнализации / обмена сообщениями FSA.