zeroMQ初体验-11.节点间的协作

上一篇讲到了线程间的协作，通过zeroMQ的pair模式可以很优雅的实现。而在各节点间(进程级)，则适用度不高(虽然也能用)。这里给出了两个理由：
1.节点间是可以调节的，而线程间不是（线程是稳定的），pair模式是非自动连接的.
2.线程数是固定的，可预估的。而节点则是变动、不可预估的。
由此得出结论：pair适用于稳定、可控的环境。

所以，有了本章节。不知诸位还记得前面所讲的发布/订阅模式，在那里曾说过这种模式是不太稳定的(主要是指初始阶段)，容易在连接未建立前就发布、废弃部分数据。在这里，通过节点间的协作来解决那个难题。

模型图：


发布端：

import zmq

SUBSCRIBERS_EXPECTED = 2

def main():
    context = zmq.Context()
    
    publisher = context.socket(zmq.PUB)
    publisher.bind('tcp://*:5561')

    syncservice = context.socket(zmq.REP)
    syncservice.bind('tcp://*:5562')

    subscribers = 0
    while subscribers < SUBSCRIBERS_EXPECTED:
        msg = syncservice.recv()
        syncservice.send('')
        subscribers += 1
        print "+1 subscriber"
    
    for i in range(1000000):
       publisher.send('Rhubarb');

    publisher.send('END')

if name == 'main':
    main()

订阅端：

import zmq

def main():
    context = zmq.Context()
    
    subscriber = context.socket(zmq.SUB)
    subscriber.connect('tcp://localhost:5561')
    subscriber.setsockopt(zmq.SUBSCRIBE, "")

    syncclient = context.socket(zmq.REQ)
    syncclient.connect('tcp://localhost:5562')
    
    syncclient.send('')
    
    syncclient.recv()

    nbr = 0
    while True:
        msg = subscriber.recv()
        if msg == 'END':
            break
        nbr += 1
    
    print 'Received %d updates' % nbr

if name == 'main':
    main()

由上例可见，通过应答模式解决了之前的困扰，如果还不放心的话，也可以通过发布特定参数，当订阅端得到时再应答，安全系数便又升了一级。不过这里有个大前提，得先通过某种方式得到或预估一个概念数来确保应用的可用性。

（未完待续）

评论

3 楼 iyuan 2011-12-09  

guozhiwei 写道

同redis的订阅发布服务和icestorm的订阅发布服务相比 zeromq的显得格外的不方便

redis什么的都用到了中间件缓存数据(话说回来好像都要的..)想象成udp,至于你收不收，反正我是发了..

2 楼 guozhiwei 2011-12-08  

同redis的订阅发布服务和icestorm的订阅发布服务相比 zeromq的显得格外的不方便

1 楼 guozhiwei 2011-12-07  

这样子的订阅发布没多大实际用处