Python 和 ZeroMQ 订阅者对发布者的确认

Question

我正在尝试使用Python中的ZeroMQ从我的订阅者那里获得对我的发布者的确认。

我尝试了一些代码示例，使用zmq.PUSH和zmq.PULL代码序列，但无济于事。

我的代码：

PUB_SERVER.PY

import zmq
import random
import sys
import time
port = "5556"
if len( sys.argv ) > 1:
   port =  sys.argv[1]
   int( port )
   context = zmq.Context()
   socket  = context.socket( zmq.PUB )
   socket.bind( "tcp://*:%s" % port )
   while True:
      # topic = random.randrange( 9999, 10005 )
      topic       = 10000
      messagedata = random.randrange( 1, 215 ) - 80
      print        "%d %d" % ( topic, messagedata )
      socket.send( "%d %d" % ( topic, messagedata ) )
      time.sleep( 1 )

SUB_CLIENT.PY

import sys
import zmq
port = "5556"
if len( sys.argv ) > 1:
   port =  sys.argv[1]
   int( port )
   if len( sys.argv ) > 2:
      port1   =  sys.argv[2]
      int( port1 )
      # Socket to talk to server
      context = zmq.Context()
      socket  = context.socket( zmq.SUB )
      print "Collecting updates from weather server..."
      socket.connect( "tcp://192.168.0.21:%s" % port )
if len( sys.argv ) > 2:
   socket.connect(    "tcp://192.168.0.21:%s" % port1 )
   # Subscribe to zipcode, default is NYC, 10001
   topicfilter = "10000"
   socket.setsockopt( zmq.SUBSCRIBE, topicfilter )
   # Process 5 updates
   total_value = 0
   for update_nbr in range( 5 ):
      string = socket.recv()
      topic, messagedata = string.split()
      total_value += int( messagedata )
      print topic, messagedata
   print "Average messagedata value for topic '%s' was %dF" % ( topicfilter, total_value / update_nbr )

该代码在一个SSH窗口（在Parallella上）中为我提供了服务器的输出，并在另一个SSH窗口（在RaspberryPi上）中为我提供了客户端的过滤消息，该窗口运行良好。

我迷路的地方是，一旦客户端从服务器获得过滤的消息，它将如何确认收到的过滤消息，然后让服务器记录这些确认的消息？

最终，我想做一些明智的决策，将文件发送给确认的订阅者。

Answer 1

如何确认？

可以为此目的为软信令创建并行消息传递结构。

使用 .SUB Rx 接入点扩展PUB_SERVER.PY：

anAckRxSOCKET  = context.socket( zmq.SUB )               # create SUB side
anAckRxSOCKET.bind( "tcp://*:%s" % aServerAckPORT ) )    # .bind()
anAckRxSOCKET.setsockopt( zmq.SUBSCRIBE, "" )            # SUB to *-anything
# ...
anAckRxSTRING = anAckRxSOCKET.recv()                     # .recv()

使用 .PUB Tx 套接字将SUB_CLIENT.PY扩展到服务器端接入点：

anAckTxSOCKET  = context.socket( zmq.PUB )               # create PUB side(s)
anAckTxSOCKET.connect( "tcp://192.168.0.21:%s" % aServerAckPORT ) )

和发送带有"a-proxy-ID"的 ACK，以进行您可能想要或需要的任何服务器端处理

anAckTxSOCKET.send( topicfilter )                        # ACK with an "identity"-proxy