ØMQ指南：Python实现的零MQ入门教程

需积分: 10 91 浏览量更新于2024-07-16 收藏 5.28MB PDF 举报

身份认证购VIP最低享 7 折!

领优惠券(最高得80元）

资源详情

资源推荐

2018/2/14 ØMQ - The Guide - ØMQ - The Guide
http://zguide.zeromq.org/py:all#The-DEALER-to-DEALER-Combination 16/411
    # Python 3
 
    raw_input = input
context = zmq.Context()
# Socket to send messages on
 
sender = context.socket(zmq.PUSH)
 
sender.bind("tcp://*:5557")
# Socket with direct access to the sink: used to syncronize start of
batch
 
sink = context.socket(zmq.PUSH)
 
sink.connect("tcp://localhost:5558")
print("Press Enter when the workers are ready: ")
 
_ = raw_input()
 
print("Sending tasks to workers…")
# The first message is "0" and signals start of batch
 
sink.send(b'0')
# Initialize random number generator
 
random.seed()
# Send 100 tasks
 
total_msec = 0
 
for task_nbr in range(100):
    # Random workload from 1 to 100 msecs
 
    workload = random.randint(1, 100)
 
    total_msec += workload
    sender.send_string(u'%i' % workload)
print("Total expected cost: %s msec" % total_msec)
# Give 0MQ time to deliver
 
time.sleep(1)
C | C++ | C# | Clojure | CL | Delphi | Erlang | F# | Felix | Go | Haskell | Haxe | Java | Lua | Node.js |
Objective-C | Perl | PHP | Ruby | Scala | Tcl | Ada | Basic | ooc | Q | Racket
Here is the worker application. It receives a message, sleeps for that number of seconds, and
then signals that it's finished:
taskwork: Parallel task worker in Pythonin Python
# Task worker
 
# Connects PULL socket to tcp://localhost:5557
 
# Collects workloads from ventilator via that socket
 
# Connects PUSH socket to tcp://localhost:5558
 
# Sends results to sink via that socket
 
#
 
# Author: Lev Givon <lev(at)columbia(dot)edu>
import sys
 
import time
 

2018/2/14 ØMQ - The Guide - ØMQ - The Guide
http://zguide.zeromq.org/py:all#The-DEALER-to-DEALER-Combination 18/411
# Process 100 confirmations
 
for task_nbr in range(100):
 
    s = receiver.recv()
 
    if task_nbr % 10 == 0:
 
        sys.stdout.write(':')
 
    else:
 
        sys.stdout.write('.')
 
    sys.stdout.flush()
# Calculate and report duration of batch
 
tend = time.time()
 
print("Total elapsed time: %d msec" % ((tend-tstart)*1000))
C | C++ | C# | Clojure | CL | Delphi | Erlang | F# | Felix | Go | Haskell | Haxe | Java | Lua | Node.js |
Objective-C | Perl | PHP | Ruby | Scala | Tcl | Ada | Basic | ooc | Q | Racket
The average cost of a batch is 5 seconds. When we start 1, 2, or 4 workers we get results
like this from the sink:
1 worker: total elapsed time: 5034 msecs.
2 workers: total elapsed time: 2421 msecs.
4 workers: total elapsed time: 1018 msecs.
Let's look at some aspects of this code in more detail:
The workers connect upstream to the ventilator, and downstream to the sink. This
means you can add workers arbitrarily. If the workers bound to their endpoints, you
would need (a) more endpoints and (b) to modify the ventilator and/or the sink each
time you added a worker. We say that the ventilator and sink are stable parts of our
architecture and the workers are dynamic parts of it.
We have to synchronize the start of the batch with all workers being up and running.
This is a fairly common gotcha in ZeroMQ and there is no easy solution. The
zmq_connect method takes a certain time. So when a set of workers connect to the
ventilator, the first one to successfully connect will get a whole load of messages in that
short time while the others are also connecting. If you don't synchronize the start of the
batch somehow, the system won't run in parallel at all. Try removing the wait in the
ventilator, and see what happens.
The ventilator's PUSH socket distributes tasks to workers (assuming they are all
connected before the batch starts going out) evenly. This is called load balancing and
it's something we'll look at again in more detail.
The sink's PULL socket collects results from workers evenly. This is called fair-queuing.
Figure 6 - Fair Queuing

2018/2/14 ØMQ - The Guide - ØMQ - The Guide

http://zguide.zeromq.org/py:all#The-DEALER-to-DEALER-Combination 20/411

ZeroMQ applications always start by creating a context, and then using that for creating

sockets. In C, it's the zmq_ctx_new() call. You should create and use exactly one context in

your process. Technically, the context is the container for all sockets in a single process, and

acts as the transport for inproc sockets, which are the fastest way to connect threads in one

process. If at runtime a process has two contexts, these are like separate ZeroMQ instances.

If that's explicitly what you want, OK, but otherwise remember:

Call zmq_ctx_new() once at the start of a process, and zmq_ctx_destroy() once at the

end.

If you're using the fork() system call, do zmq_ctx_new() after the fork and at the beginning

of the child process code. In general, you want to do interesting (ZeroMQ) stuff in the

children, and boring process management in the parent.

Making a Clean Exit

top prev next

Classy programmers share the same motto as classy hit men: always clean-up when you

finish the job. When you use ZeroMQ in a language like Python, stuff gets automatically freed

for you. But when using C, you have to carefully free objects when you're finished with them

or else you get memory leaks, unstable applications, and generally bad karma.

Memory leaks are one thing, but ZeroMQ is quite finicky about how you exit an application.

The reasons are technical and painful, but the upshot is that if you leave any sockets open,

the zmq_ctx_destroy() function will hang forever. And even if you close all sockets,

zmq_ctx_destroy() will by default wait forever if there are pending connects or sends unless

you set the LINGER to zero on those sockets before closing them.

The ZeroMQ objects we need to worry about are messages, sockets, and contexts. Luckily it's

quite simple, at least in simple programs:

Use zmq_send() and zmq_recv() when you can, as it avoids the need to work with

zmq_msg_t objects.

If you do use zmq_msg_recv(), always release the received message as soon as you're

done with it, by calling zmq_msg_close().

If you are opening and closing a lot of sockets, that's probably a sign that you need to

redesign your application. In some cases socket handles won't be freed until you

destroy the context.

When you exit the program, close your sockets and then call zmq_ctx_destroy(). This

destroys the context.

This is at least the case for C development. In a language with automatic object destruction,

sockets and contexts will be destroyed as you leave the scope. If you use exceptions you'll

have to do the clean-up in something like a "final" block, the same as for any resource.

If you're doing multithreaded work, it gets rather more complex than this. We'll get to

multithreading in the next chapter, but because some of you will, despite warnings, try to run

before you can safely walk, below is the quick and dirty guide to making a clean exit in a

multithreaded ZeroMQ application.

First, do not try to use the same socket from multiple threads. Please don't explain why you

think this would be excellent fun, just please don't do it. Next, you need to shut down each

socket that has ongoing requests. The proper way is to set a low LINGER value (1 second),

剩余410页未读，继续阅读

naocwz

粉丝: 0
资源: 3

ØMQ指南：Python实现的零MQ入门教程

ØMQ - The Guide

[m,n]=size(f); R=f(:,:,1); G=f(:,:,2); B=f(:,:,3); %½øÐÐ¸³Öµ I=(R+G+B)/3; S=1-(3.*min(min(R,G),B))./(R+B+G+eps); theta=acos((((R-G)+(R-B))*0.5)./(sqrt((R-G).*(R-G)+(R-B).*(G-B))+eps)); H=theta; H(B>G)=2*pi-H(B>G); % H=H/(2*pi); %·µ»Ø hsi=cat(3,H,S,I);

验证邮箱地址的有效性,要求如下。 Ø 使用prompt( )方法输入邮箱地址,默认值是susan@sohu .com,如图1.43所示. Ø 正确的邮箱地址必须包含“@”和“.”,输入不正确提示如图1.45所示。

mysql数据库控制台乱码_mysql¿ØÖÆÌ¨µ¼Èëµ¼³öÊý¾Ý¿âÓëÂÒÂë½â¾ö°ì·¨...

画出广义表Y=((a,(b,c)),(),d,(e,f,g))头尾链表结构图和扩展链表结构图

Ø 简述聚类算法的分类。 Ø 解释模型的评价指标。

正规表达式构造nfa

data = (data - data.mean(axis=0)) / data.std(axis=0)

1、发现类 ； Ø2、发现类的属性； Ø3、发现类的方法； Ø4、优化设计； Ø5、梳理运行过程。

在双碳目标下，油液过滤技术有哪些新的发展方向、前景或者是技术革新？对目前产业提出了哪些新的需求？写一篇相关文档，要求1000字以上，并列近几年相关文献

广义表Y=((a,(b,c)),(),d,(e,f,g))的头尾链表结构图

最新资源

[m,n]=size(f); R=f(:,:,1); G=f(:,:,2); B=f(:,:,3); %½øÐÐ¸³Öµ I=(R+G+B)/3; S=1-(3.min(min(R,G),B))./(R+B+G+eps); theta=acos((((R-G)+(R-B))0.5)./(sqrt((R-G).(R-G)+(R-B).(G-B))+eps)); H=theta; H(B>G)=2pi-H(B>G); % H=H/(2pi); %·µ»Ø hsi=cat(3,H,S,I);

1、发现类； Ø2、发现类的属性； Ø3、发现类的方法； Ø4、优化设计； Ø5、梳理运行过程。