ZeroMQ에서 팬 아웃하는 법? 여러 주제에서 여러 클라이언트로 전달

Question

제어 채널 (req-rep)을 통해 주제 세트를 지정하는 요청을 수신 한 서버를 구현해야합니다. 그러면 응답이 게시자 소켓을 가리키는 URL을 전송합니다 이 특정 클라이언트 또는에 대해 열렸습니다 (권한이 충분하지 않아 거부 메시지).

한 번에 하나의 클라이언트 (두 개의 무한 루프)를 처리 할 수있는 버전을 구현했지만 여러 클라이언트를 동시에 처리하는 데 사용할 패턴을 모르겠습니다.

다른 클라이언트의 소켓이 분리되어 있어야한다는 것이 중요합니다. 스레드를 사용하는 것이 작업을 수행하는

import zmq 

context = zmq.Context() 

upstream_port = 10000 
upstream_host = 'localhost' 
control_channel_port = 11000 
upstream_addr = 'tcp://{}:{}'.format(upstream_host, upstream_port) 

def should_grant(request): 
    ''' 
    Permission checking - irrelevant to the question 
    ''' 
    return True 


def bind_downstream(): 
    downstream = context.socket(zmq.PUB) 
    addr = 'tcp://*' 
    port = downstream.bind_to_random_port(addr) 
    return downstream, port 


def bind_control_channel(): 
    control_channel_sock = context.socket(zmq.REP) 
    control_channel_sock.bind('tcp://*:{}'.format(control_channel_port)) 
    return control_channel_sock 


def connect_upstream(topics): 
    raw_data = context.socket(zmq.SUB) 
    for t in topics: 
     raw_data.setsockopt_unicode(zmq.SUBSCRIBE, unicode(t)) 
    raw_data.connect(upstream_addr) 
    return raw_data 


if __name__ == '__main__': 
    print("Binding control channel socket on {}".format('tcp://*:{}'.format(control_channel_port))) 
    control_channel = bind_control_channel() 
    while True: 
     request = control_channel.recv_json() 
     print("Received request {}".format(request)) 
     if should_grant(request): 
      (downstream_sock, downstream_port) = bind_downstream() 
      print("Downstream socket open on {}".format('tcp://*:{}'.format(downstream_port))) 

      print("Connecting to upstream on {}".format(upstream_addr)) 
      upstream_sock = connect_upstream(request['topics']) 

      control_channel.send_json({'status': 'ok', 'port': downstream_port}) 
      while True: 
       parts = upstream_sock.recv_multipart() # Simple forwarding 
       downstream_sock.send_multipart(parts) 
     else: 
      control_channel.send_json({'status': 'rejected'}) 

Answer 1

올바른 방법 :

여기에 간단한 코드입니다.

주 프로그램이나 스레드가 제어 채널 루프를 처리합니다. 연결이 나타나 자마자 업스트림 및 다운 스트림 소켓을 만들지 만 실제 전송은 스레드에서 처리합니다. 아래의 코드가 작동하는 클라이언트가 없기 때문에 작동하는지 알 수는 없지만 진행 상황을 확인하고 어떤 일이 발생하는지 확인하십시오. 그럼에도 불구하고 아이디어를 얻을 수 있습니다.