zeroMQ初体验-19.应答模式进阶(五)-异步式应答

恩，这应该算是比较实用的部分了。

模式图：

import zmq
import threading
import time
from random import choice

class ClientTask(threading.Thread):
    """ClientTask"""
    def init(self):
        threading.Thread.init (self)

    def run(self):
        context = zmq.Context()
        socket = context.socket(zmq.XREQ)
        identity = 'worker-%d' % (choice([0,1,2,3,4,5,6,7,8,9]))
        socket.setsockopt(zmq.IDENTITY, identity )
        socket.connect('tcp://localhost:5570')
        print 'Client %s started' % (identity)
        poll = zmq.Poller()
        poll.register(socket, zmq.POLLIN)
        reqs = 0
        while True:
            for i in xrange(5):
                sockets = dict(poll.poll(1000))
                if socket in sockets:
                    if sockets[socket] == zmq.POLLIN:
                        msg = socket.recv()
                        print '%s: %s\n' % (identity, msg)
                        del msg
            reqs = reqs + 1
            print 'Req #%d sent..' % (reqs)
            socket.send('request #%d' % (reqs))

        socket.close()
        context.term()

class ServerTask(threading.Thread):
    """ServerTask"""
    def init(self):
        threading.Thread.init (self)

    def run(self):
        context = zmq.Context()
        frontend = context.socket(zmq.XREP)
        frontend.bind('tcp://*:5570')

        backend = context.socket(zmq.XREQ)
        backend.bind('inproc://backend')

        workers = []
        for i in xrange(5):
            worker = ServerWorker(context)
            worker.start()
            workers.append(worker)

        poll = zmq.Poller()
        poll.register(frontend, zmq.POLLIN)
        poll.register(backend,  zmq.POLLIN)

        while True:
            sockets = dict(poll.poll())
            if frontend in sockets:
                if sockets[frontend] == zmq.POLLIN:
                    msg = frontend.recv()
                    print 'Server received %s' % (msg)
                    backend.send(msg)
            if backend in sockets:
                if sockets[backend] == zmq.POLLIN:
                    msg = backend.recv()
                    frontend.send(msg)

        frontend.close()
        backend.close()
        context.term()
 
class ServerWorker(threading.Thread):
    """ServerWorker"""
    def init(self, context):
        threading.Thread.init (self)
        self.context = context

    def run(self):
        worker = self.context.socket(zmq.XREQ)
        worker.connect('inproc://backend')
        print 'Worker started'
        while True:
            msg = worker.recv()
            print 'Worker received %s' % (msg)
            replies = choice(xrange(5))
            for i in xrange(replies):
                time.sleep(1/choice(range(1,10)))
                worker.send(msg)
            del msg

        worker.close()

def main():
    """main function"""
    server = ServerTask()
    server.start()
    for i in xrange(3):
        client = ClientTask()
        client.start()
    
    server.join()
    

if name == "main":
    main()

作为一个异步的服务器，详图应该是这样的：


这里的数据传递顺序是这样的：

 client        server     frontend     worker
    [ XREQ ]<---->[ XREP <----> XREQ <----> XREQ ]
            1 part       2 parts     2 parts

在这里有可能碰到一个比较经典的c/s问题:
c端太多，耗尽s端资源怎么办？
这就需要靠谱些的机制了，比如通过“心跳”来确定是否应该释放这个c端的资源等。当然，那就是另外一个话题了。

（未完待续）

评论

2 楼 iyuan 2012-02-06  

guozhiwei 写道

这个例子应该就是 官方提供的包里面的example包里面的Python中的asyncsrv.py

你这个例子有些地方跟它不一样,

你的例子 跑不起来 

而官方的例子能跑起来...

你的例子在send 和 recv的时没有identity,加上以后就能正常跑了.

就是直接拿的官网的例子。原来是声明了setsockopt就可以用的。也可能是zmq版本更新了导致的吧。zmq这块目前还没什么机会用到，有时间还得再捋下，初体验还是太浅了，哈哈

1 楼 guozhiwei 2012-02-06  

这个例子应该就是 官方提供的包里面的example包里面的Python中的asyncsrv.py

你这个例子有些地方跟它不一样,

你的例子 跑不起来 

而官方的例子能跑起来...

你的例子在send 和 recv的时没有identity,加上以后就能正常跑了.