- 浏览: 463325 次
- 性别:
- 来自: 上海
文章分类
最新评论
-
kc_hxd_jp:
博主问个问题,这个篇幅下的python代码无法达到应有的作用, ...
zeroMQ初体验-14.命名机制 进阶 -
kobe1029:
Map<String, Object> args ...
rabbitmq 队列长度预设的曲线方案 -
Sasoritattoo:
LZ,这都13年了,抽空把这篇文章的下文给表完了吧,这一口气喘 ...
nginx + gridfs + mongodb 大事记(残) -
3GQQ2012:
引用前文已经说过,XREP其实用以平衡负载,所以这里由它对请求 ...
zeroMQ初体验-15.应答模式进阶(一)-数据的封装 -
iyuan:
ustclz 写道图片怎么显示不了了。。我这看是可以显示的。不 ...
zeroMQ初体验-1.简介及C/S模式
从经典到超越经典。
首先,先回顾下经典:
然后,扩展:
然后,变异:
client发出的数据结构:
路由处理成:
再转给worker成:
工人处理的数据:
由worker到client是一个逆序过程,不过因为两边都是REQ类型,所以其实是一致的。
[补]:
通常,上层的api会帮我们做一些事,免去了逐步封装数据的麻烦,比如在python中,最终代码会是这个样子:
(未完待续)
首先,先回顾下经典:
然后,扩展:
然后,变异:
import threading import time import zmq NBR_CLIENTS = 10 NBR_WORKERS = 3 def worker_thread(worker_url, context, i): """ Worker using REQ socket to do LRU routing """ socket = context.socket(zmq.REQ) identity = "Worker-%d" % (i) socket.setsockopt(zmq.IDENTITY, identity) #set worker identity socket.connect(worker_url) # Tell the borker we are ready for work socket.send("READY") try: while True: # python binding seems to eat empty frames address = socket.recv() request = socket.recv() print("%s: %s\n" %(identity, request)) socket.send(address, zmq.SNDMORE) socket.send("", zmq.SNDMORE) socket.send("OK") except zmq.ZMQError, zerr: # context terminated so quit silently if zerr.strerror == 'Context was terminated': return else: raise zerr def client_thread(client_url, context, i): """ Basic request-reply client using REQ socket """ socket = context.socket(zmq.REQ) identity = "Client-%d" % (i) socket.setsockopt(zmq.IDENTITY, identity) #Set client identity. Makes tracing easier socket.connect(client_url) # Send request, get reply socket.send("HELLO") reply = socket.recv() print("%s: %s\n" % (identity, reply)) return def main(): """ main method """ url_worker = "inproc://workers" url_client = "inproc://clients" client_nbr = NBR_CLIENTS # Prepare our context and sockets context = zmq.Context(1) frontend = context.socket(zmq.XREP) frontend.bind(url_client) backend = context.socket(zmq.XREP) backend.bind(url_worker) # create workers and clients threads for i in range(NBR_WORKERS): thread = threading.Thread(target=worker_thread, args=(url_worker, context, i, )) thread.start() for i in range(NBR_CLIENTS): thread_c = threading.Thread(target=client_thread, args=(url_client, context, i, )) thread_c.start() # Logic of LRU loop # - Poll backend always, frontend only if 1+ worker ready # - If worker replies, queue worker as ready and forward reply # to client if necessary # - If client requests, pop next worker and send request to it # Queue of available workers available_workers = 0 workers_list = [] # init poller poller = zmq.Poller() # Always poll for worker activity on backend poller.register(backend, zmq.POLLIN) # Poll front-end only if we have available workers poller.register(frontend, zmq.POLLIN) while True: socks = dict(poller.poll()) # Handle worker activity on backend if (backend in socks and socks[backend] == zmq.POLLIN): # Queue worker address for LRU routing worker_addr = backend.recv() assert available_workers < NBR_WORKERS # add worker back to the list of workers available_workers += 1 workers_list.append(worker_addr) # Second frame is empty empty = backend.recv() assert empty == "" # Third frame is READY or else a client reply address client_addr = backend.recv() # If client reply, send rest back to frontend if client_addr != "READY": # Following frame is empty empty = backend.recv() assert empty == "" reply = backend.recv() frontend.send(client_addr, zmq.SNDMORE) frontend.send("", zmq.SNDMORE) frontend.send(reply) client_nbr -= 1 if client_nbr == 0: break # Exit after N messages # poll on frontend only if workers are available if available_workers > 0: if (frontend in socks and socks[frontend] == zmq.POLLIN): # Now get next client request, route to LRU worker # Client request is [address][empty][request] client_addr = frontend.recv() empty = frontend.recv() assert empty == "" request = frontend.recv() # Dequeue and drop the next worker address available_workers -= 1 worker_id = workers_list.pop() backend.send(worker_id, zmq.SNDMORE) backend.send("", zmq.SNDMORE) backend.send(client_addr, zmq.SNDMORE) backend.send(request) #out of infinite loop: do some housekeeping time.sleep (1) frontend.close() backend.close() context.term() if name == "main": main()
client发出的数据结构:
路由处理成:
再转给worker成:
工人处理的数据:
由worker到client是一个逆序过程,不过因为两边都是REQ类型,所以其实是一致的。
[补]:
通常,上层的api会帮我们做一些事,免去了逐步封装数据的麻烦,比如在python中,最终代码会是这个样子:
import threading import time import zmq NBR_CLIENTS = 10 NBR_WORKERS = 3 def worker_thread(worker_url, context, i): """ Worker using REQ socket to do LRU routing """ socket = context.socket(zmq.REQ) identity = "Worker-%d" % (i) socket.setsockopt(zmq.IDENTITY, identity) #set worker identity socket.connect(worker_url) # Tell the borker we are ready for work socket.send("READY") try: while True: [address, request] = socket.recv_multipart() print("%s: %s\n" %(identity, request)) socket.send_multipart([address, "", "OK"]) except zmq.ZMQError, zerr: # context terminated so quit silently if zerr.strerror == 'Context was terminated': return else: raise zerr def client_thread(client_url, context, i): """ Basic request-reply client using REQ socket """ socket = context.socket(zmq.REQ) identity = "Client-%d" % (i) socket.setsockopt(zmq.IDENTITY, identity) #Set client identity. Makes tracing easier socket.connect(client_url) # Send request, get reply socket.send("HELLO") reply = socket.recv() print("%s: %s\n" % (identity, reply)) return def main(): """ main method """ url_worker = "inproc://workers" url_client = "inproc://clients" client_nbr = NBR_CLIENTS # Prepare our context and sockets context = zmq.Context(1) frontend = context.socket(zmq.XREP) frontend.bind(url_client) backend = context.socket(zmq.XREP) backend.bind(url_worker) # create workers and clients threads for i in range(NBR_WORKERS): thread = threading.Thread(target=worker_thread, args=(url_worker, context, i, )) thread.start() for i in range(NBR_CLIENTS): thread_c = threading.Thread(target=client_thread, args=(url_client, context, i, )) thread_c.start() # Logic of LRU loop # - Poll backend always, frontend only if 1+ worker ready # - If worker replies, queue worker as ready and forward reply # to client if necessary # - If client requests, pop next worker and send request to it # Queue of available workers available_workers = 0 workers_list = [] # init poller poller = zmq.Poller() # Always poll for worker activity on backend poller.register(backend, zmq.POLLIN) # Poll front-end only if we have available workers poller.register(frontend, zmq.POLLIN) while True: socks = dict(poller.poll()) # Handle worker activity on backend if (backend in socks and socks[backend] == zmq.POLLIN): # Queue worker address for LRU routing message = backend.recv_multipart() assert available_workers < NBR_WORKERS worker_addr = message[0] # add worker back to the list of workers available_workers += 1 workers_list.append(worker_addr) # Second frame is empty empty = message[1] assert empty == "" # Third frame is READY or else a client reply address client_addr = message[2] # If client reply, send rest back to frontend if client_addr != "READY": # Following frame is empty empty = message[3] assert empty == "" reply = message[4] frontend.send_multipart([client_addr, "", reply]) client_nbr -= 1 if client_nbr == 0: break # Exit after N messages # poll on frontend only if workers are available if available_workers > 0: if (frontend in socks and socks[frontend] == zmq.POLLIN): # Now get next client request, route to LRU worker # Client request is [address][empty][request] [client_addr, empty, request ] = frontend.recv_multipart() assert empty == "" # Dequeue and drop the next worker address available_workers -= 1 worker_id = workers_list.pop() backend.send_multipart([worker_id, "", client_addr, request]) #out of infinite loop: do some housekeeping time.sleep (1) frontend.close() backend.close() context.term() if name == "main": main()
(未完待续)
发表评论
-
IM选型(初)
2016-08-23 19:12 1593主要参考文章: https://r ... -
关于python和rabbitmq的那点事儿
2011-10-19 14:15 7912rabbitmq是一个消息中间件,在之前的zmq介绍中有略带提 ... -
zeroMQ初体验-34.发布/订阅模式进阶-克隆模式-下,结言
2011-05-26 16:09 4075服务器: // // Clone server Mod ... -
zeroMQ初体验-33.发布/订阅模式进阶-克隆模式-中
2011-05-26 15:37 2857临时缓存 现实中,比如 ... -
zeroMQ初体验-32.发布/订阅模式进阶-克隆模式-上
2011-05-26 15:04 3602在发布/订阅模式中,特别是现实应用中,总会因为这样那样的问题导 ... -
zeroMQ初体验-31.发布/订阅模式进阶-黑盒的高速订阅者
2011-05-25 16:55 2708作为发布/订阅模式的一个常用场景,大数据量的组播是有必要的。虽 ... -
zeroMQ初体验-30.发布/订阅模式进阶-自裁的蜗牛订阅者
2011-05-25 16:24 4485在初次介绍发布/订阅模式的时候,就已经抖出了这个包袱:如果订阅 ... -
zeroMQ初体验-29.可靠性-自由模式
2011-05-24 17:02 5353好吧,本以为这可能是一个更靠谱的模式,谁知(其实是我一厢情愿了 ... -
zeroMQ初体验-28.可靠性-主从模式
2011-05-23 14:47 5457虽然"硬盘模式" ... -
zeroMQ初体验-27.可靠性-硬盘模式
2011-05-23 13:44 3748在之前的种种模式中, ... -
zeroMQ初体验-26.可靠性-管家模式
2011-05-12 19:05 5608上一节末尾有说到协议,zeromq自然做了充沛的封装,&quo ... -
zeroMQ初体验-26.可靠性-管家模式
2011-05-12 19:03 1上一节末尾有说到协议,zeromq自然做了充沛的封装,&quo ... -
zeroMQ初体验-25.可靠性-偏执的海盗模式
2011-05-05 19:05 3502虽然说“简单的海盗模 ... -
zeroMQ初体验-24.可靠性-简单的海盗模式
2011-05-05 16:41 3160相较于“懒惰的”做了 ... -
zeroMQ初体验-23.可靠性-懒惰的海盗模式
2011-05-05 16:15 5016相较于通常的阻塞模式,这里只是做了一点简单的动作来加强系统的可 ... -
zeroMQ初体验-22.可靠性-总览
2011-04-26 19:25 5867在开篇就从曾对zeromq的可靠性做过质疑,不过,作为一个雄心 ... -
rabbitmq 队列长度预设的曲线方案
2011-04-21 14:36 3317zeromq中倒是直接支持这个功能的。 类似于设定队列长度或 ... -
zeroMQ初体验-21.应答模式进阶(七)-云计算
2011-04-18 19:14 3479这里给出了一个最近很火的"云计算"案例。 ... -
zeroMQ初体验-20.应答模式进阶(六)-多对多路由模式
2011-04-18 17:22 3810某些时候,为了冗余的需要,可能会有这样的需求: impo ... -
zeroMQ初体验-19.应答模式进阶(五)-异步式应答
2011-04-15 15:23 4780恩,这应该算是比较实 ...
相关推荐
zeromq-2.1.7.tar.gz 的早期的一个版本,本人已安装成功 放心使用
zeromq-4.0.3.tar.gz zeromq-4.0.3.tar.gz zeromq-4.0.3.tar.gz
zeromq-4.3.2.tar.gz,可在linux下编译安装,能够使用zeromq进行sorket开发,多线程,提升性能,效率,可以配合msgpack进行使用,是个好的扩展插件
zeromq-4.1.3.tar.gz,最新的zeromq的开发工具包,希望对开发者有用
官方离线安装包,测试可用。使用rpm -ivh [rpm完整包名] 进行安装
zeromq-4.1.8.tar.gz 有问题请联系
zeromq-2.1.9.tar.gz 这是zeromq linux 官方原版 请放心下载
zeromq-4.2.3.tar.gz 一个稳定的版本,可以安装使用!
0MQ version 4.3.4 stable, released on 2021/01/17
storm搭建所需资源
在官网下载zeromq太慢了,网速极不稳定,特意下载放在这里供大家下载,当然象征性地赚点 资源分
ZeroMQ是一个网络通讯库,其主要用来为分布式应用程序开发提供进程间通信(此处的进程既可以是同一台机器上的两个进程也可以是不同机器上的两个进程)。ZeroMQ的特点在于灵活的通信手段和丰富的连接模型,并且它可以...
ZeroMQ是一个网络通讯库,其主要用来为分布式应用程序开发提供进程间通信(此处的进程既可以是同一台机器上的两个进程也可以是不同机器上的两个进程)。ZeroMQ的特点在于灵活的通信手段和丰富的连接模型,并且它可以...
ZMQ(以下ZeroMQ简称ZMQ)是一个简单好用的传输层,像框架一样的一个socket library,他使得Socket编程更加简单、简洁和性能更高。是一个消息处理队列库,可在多个线程、内核和主机盒之间弹性伸缩。
0MQ version 4.3.4 stable, released on 2021/01/17
ZeroMQ(也说明 ØMQ,0MQ 或 ZMQ)是一个高性能的异步消息库,旨在使用分布式或并行应用程序。它提供了一个消息队列,但 不同于面向消息的中间件,一个 ZeroMQ 系统可以在没有专用运行消息代理。jzmq安装包,本人也...
zeroMQ 4.2.3版本 zeromq-4.2.3.tar.gz 欢迎关注我的CSDN博客:https://mp.csdn.net/console/home 免积分下载
zeromq-4.2.5.tar.gz
VS2015 在Widows 10 上编译的 ZeroMQ 4.3.2,JZMQ 3.1 CZMQ 4.2,可以在 JDK 1.8 下运行。DLL 都是 64位,包含了编译及运行相关信息。分享一下,也给自己留个备份