구독자와 메시지가 손실되기 전에 게시자가 완료됩니다. 이유가 무엇입니까?

Question

zeromq에 새로 입문하고 기본적인 pub/sub를 작동 시키려고합니다. 다음을 실행하면 (게시 전 시작 하위) 게시자가 완료되지만 구독자가 모든 메시지를받지 못하고 응답하지 않습니다. 이유가 무엇입니까?

소켓이 닫혀 있지만 메시지가 전송 된 것 같습니까? 모든 메시지가 수신되도록하는 방법이 있습니까?

제작사 :

import zmq 
import random 
import time 
import tnetstring 

context=zmq.Context() 
socket=context.socket(zmq.PUB) 
socket.bind("tcp://*:5556") 

y=0 
for x in xrange(5000): 
    st = random.randrange(1,10) 
    data = [] 
    data.append(random.randrange(1,100000)) 
    data.append(int(time.time())) 
    data.append(random.uniform(1.0,10.0)) 

    s = tnetstring.dumps(data) 
    print 'Sending ...%d %s' % (st,s) 

    socket.send("%d %s" % (st,s)) 
    print "Messages sent: %d" % x 
    y+=1 

print '*** SERVER FINISHED. # MESSAGES SENT = ' + str(y) 

가입자 : -

import sys 
import zmq 
import tnetstring 

# Socket to talk to server 
context = zmq.Context() 
socket = context.socket(zmq.SUB) 

socket.connect("tcp://localhost:5556") 
filter = "" # get all messages 

socket.setsockopt(zmq.SUBSCRIBE, filter) 

x=0 
while True: 
    topic,data = socket.recv().split() 
    print "Topic: %s, Data = %s. Total # Messages = %d" % (topic,data,x) 
    x+=1 

Answer 1

ZeroMQ에서, 클라이언트와 서버가 항상 다시 연결을 시도; 그들은 다른 쪽이 끊어지면 내려 가지 않을 것입니다. 왜냐하면 많은 경우에 상대방이 다시 오면 다시 말하기를 원하기 때문입니다. 따라서 테스트 코드에서 클라이언트는 어느 시점에서 메시지를 보내지 않으면 서버가 메시지를 다시 보내기 시작할 때까지 기다립니다.

Answer 2

특정 경우에 socket.close() 및 context.term()을 사용하여 조사 할 수 있습니다. 모든 메시지가 전송 될 때까지 차단됩니다. 느린 결합 자의 문제도 있습니다. 바인드 후 수면을 추가 할 수 있지만 게시를 시작하기 전에 수면을 추가 할 수 있습니다. 이것은 테스트 케이스에서 작동하지만 솔루션과 밴치드가 실제로 무엇인지 이해하고 싶어합니다.

PUB/SUB 패턴을 라디오처럼 생각해야합니다. 발신자와 수신자 모두 비동기입니다. 게시자는 아무도 듣지 않아도 계속 보내게됩니다. 구독자는 청취중인 경우에만 데이터를 수신합니다. 네트워크가 중간에 다운되면 데이터가 손실됩니다.

메시지를 디자인하려면이 내용을 이해해야합니다. 예를 들어 메시지를 "멱등 원"으로 디자인하면 데이터를 잃어 버리지도 상관 없습니다. 이것의 예는 상태 유형 메시지입니다. 이전 상태가 있다면 상관 없습니다. 최신 메시지가 정확하고 메시지 손실은 중요하지 않습니다. 이 접근 방식의 이점은보다 견고하고 효율적인 시스템으로 끝나는 것입니다. 단점은 이러한 방식으로 메시지를 디자인 할 수없는 경우입니다.

이 예에는 손실이 필요없는 메시지 유형이 포함되어 있습니다. 다른 유형의 메시지는 트랜잭션입니다. 예를 들어 방금 시스템에서 변경된 항목의 델타를 보낸 경우 메시지를 잃을 수 없습니다. 데이터베이스 복제는 종종 이런 방식으로 관리되기 때문에 db 복제가 종종 매우 취약합니다. 보장을 제공하기 위해, 당신은 몇 가지 일을해야합니다. 한 가지는 영구 캐시를 추가하는 것입니다. 전송 된 각 메시지는 지속적 캐시에 기록되어야합니다. 클라이언트가 메시지가 누락되었는지 판별 할 수 있도록 각 메시지에는 고유 한 ID (순서가 바람직 함)가 지정되어야합니다. 누락 된 메시지를 개별적으로 요청하기 위해 클라이언트에 대해 두 번째 소켓 (ROUTER/REQ)을 추가해야합니다. 또는 보조 소켓을 사용하여 PUB/SUB를 통한 재전송을 요청할 수 있습니다. 클라이언트는 모든 메시지를 다시 수신합니다 (멀티 캐스트 버전에서 작동 함). 클라이언트는 이미 본 메시지를 무시합니다. 참고 : 이것은 ZeroMQ 가이드에있는 MAJORDOMO 패턴을 따릅니다.

다른 방법은 ROUTER/DEALER 소켓을 사용하여 사용자 자신의 브로커를 만드는 것입니다. ROUTER 소켓은 각 대리점이 연결될 때 ID를 저장합니다. ROUTER가 데이터를 보내야 할 때 모든 클라이언트 ID를 반복하고 메시지를 게시합니다. 각 메시지에는 클라이언트가 요청할 누락 된 메시지를 알 수 있도록 시퀀스가 ​​있어야합니다. 참고 : 이것은 linkedin에서 Kafka의 일종의 reimplementation입니다.