ZMQ-python：一对多的推/拉

Question

我正在使用以下代码尝试 zmq，但订阅者正在一个接一个地获取对象。

以下是我的 PUSH 脚本：

# zmq server -- run it once

import zmq
import time
# server
# print(zmq.Context)
ctx = zmq.Context()
sock = ctx.socket(zmq.PUSH)
sock.bind('ipc:///tmp/zmqtest')
i=0
while True:
    i+=1
    time.sleep(0.5)
    sock.send_pyobj((i))

以下是 PULL 脚本：

# zmq client -- run it 2,3 times in parallel

import zmq
ctx = zmq.Context() # create a new context to kick the wheels
sock = ctx.socket(zmq.PULL)
sock.connect('ipc:///tmp/zmqtest')

i=0
while True:
    i+=1
    o = sock.recv_pyobj()
    print('received python object:', o,i)
    if o == 'quit':
        print('exiting.')
        break

我从 PULL 脚本之一得到以下输出：

received python object: 1 1
received python object: 3 2
received python object: 5 3
received python object: 7 4

如何将对象并行推送到两个脚本？ 我尝试了 PUB/SUB，但它不是这样工作的。 （可以检查将PUSH/PULL替换为PUB/SUB）

Answer 1

PUB-side ：

# zmq PUB-server -- run it once

import zmq
import time

IPC  = 'ipc:///tmp/zmqtest'
ctx  = zmq.Context()
PUB  = ctx.socket( zmq.PUB )
PUB.bind( IPC )
#------------------------------------------------- SELF-DEFENSIVE CONFIGURATION
PUB.setsockopt( zmq.LINGER, 0 )
PUB.setsockopt( zmq...        )
#------------------------------------------------------------------------------
i = 0
while True:
    i += 1
    time.sleep( 0.5 )
    sock.send_pyobj( ( i ) )
#------------------------------------------------------------------------------

SUB-side(s)：

# zmq SUB-client -- run x-times concurrently ( or distributed, if other TransportClasses permit )

import zmq

IPC = 'ipc:///tmp/zmqtest'        # <TransportClass>://<address>, TCP,TIPC,...may follow
ctx = zmq.Context()               # create a new context to kick the wheels
SUB = ctx.socket( zmq.SUB )
SUB.connect( IPC )
#------------------------------------------------- SELF-DEFENSIVE CONFIGURATION
SUB.setsockopt( zmq.LINGER,     0 )
SUB.setsockopt( zmq.SUBSCRIBE, "" )
SUB.setsockopt( zmq...            )
#------------------------------------------------------------------------------
i    = 0
aClk = zmq.Stopwatch()
MASK = '(i:{1:_>9d}): After{2:_>+12d} [us] did .recv() a python object:[{0:}]'
while True:
    i += 1
    aClk.start()
    o = sock.recv_pyobj()
    _ = aClk.stop()
    print( MASK.format( repr( o ), i, _ ) )

    if o == 'quit':
        print( 'Will exit.' )
        #--------------------------------------- BE NICE & FAIR TO RESOURCES
        SUB.setsockopt( zmq.UNSUBSCRIBE, "" )
        SUB.disconnect( IPC )
        SUB.close()
        ctx.term()
        #-------------------------------------------------------------------
        break

“是的。我需要将每个对象发送到两个（或多个）脚本无损”

请注意，对此有零保修。可以构建自己的应用级协议来实现这一点。