python zeromq 介绍
简介:
ZeroMQ并不是一个对socket的封装,不能用它去实现已有的网络协议。
它有自己的模式,不同于更底层的点对点通讯模式。
它有比tcp协议更高一级的协议。(当然ZeroMQ不一定基于TCP协议,它也可以用于进程间和进程内通讯)
它改变了通讯都基于一对一的连接这个假设。
ZeroMQ把通讯的需要看成4类,其中一类是一对一结对通讯,用来支持传统的TCP socke模型,但并不推荐使用。
常用的通讯模式只有三类:
1.请求回应模型。由请求端发起请求,并等待回应端回应请求。从请求端来看,一定是一对对收发配对的;
反之,在回应端一定是发收对。请求端和回应端都可以是1:N的模型。通常把1认为是server,N认为是Client。
0MQ可以很好的支持路由功能(实现路由功能的组件叫做Device),把1:N扩展为N:M(只需要加入若干路由节点)。
从这个模型看,更底层的端点地址是对上层隐藏的。每个请求都隐含回应地址,而应用则不关心它。
2.发布订阅模型。这个模型里,发布端是单向只发送数据的,且不关心是否把全部的信息都发送给订阅者。
如果发布端开始发布信息的时候,订阅端尚未连接上,这些信息直接丢弃。
不过一旦订阅端连接上来,中间会保证没有信息丢失。
同样,订阅端则只负责接收,而不能反馈。
如果发布端和订阅端需要交互(比如要确认订阅者是否已经连接上),则使用额外的socket采用请求回应模型满足这个需求。
3.管道模型。这个模型里,管道是单向的,从PUSH端单向的向PULL端单向的推送数据流。
任何分布式,并行的需求,都可以用这三种模型组合起来解决问题。
问题1:安装zeromq和python binding?
答:
在安装zeromq之前需要安装
sudo apt-get install libtool, autoconf, automake, uuid-dev(or equivalent)
获得zeromq的源码包
wget http://download.zeromq.org/zeromq-2.0.10.tar.gz
如果想获得最新的源码包http://download.zeromq.org/
安装zeromq
tar zxvf zeromq-2.0.10.tar.gz
cd zeromq-2.0.10
(查看帮助 ./configure --help)
./configure
make
sudo make install
sudo ldconfig
安装python binding
获得pyzmq的源代码
https://github.com/zeromq/pyzmq/downloads/
pip install pyzmq
注意:pyzmq版本要与zeromq版本一致
问题2:zeromq的hello world?
答:
python examples https://github.com/imatix/zguide/tree/master/examples/Python
hwserver.py
# # Hello World server in Python # Binds REP socket to tcp://*:5555 # Expects "Hello" from client, replies with "World" # import zmq import time context = zmq.Context() socket = context.socket(zmq.REP) socket.bind("tcp://*:5555") while True: # Wait for next request from client message = socket.recv() print "Received request: ", message # Do some 'work' time.sleep (1) # Do some 'work' # Send reply back to client socket.send("World")
hwclient.py
# # Hello World client in Python # Connects REQ socket to tcp://localhost:5555 # Sends "Hello" to server, expects "World" back # import zmq context = zmq.Context() # Socket to talk to server print "Connecting to hello world server..." socket = context.socket(zmq.REQ) socket.connect ("tcp://localhost:5555") # Do 10 requests, waiting each time for a response for request in range (1,10): print "Sending request ", request,"..." socket.send ("Hello") # Get the reply. message = socket.recv() print "Received reply ", request, "[", message, "]"
问题3:zeroMQ实现一个消息层?
答:
实现一个ZeroMQ消息层需要三个步骤:
1.选择传输协议
0MQ提供了4种不同的传输协议
INPROC an In-Process communication model
IPC an Inter-Process communication model
MULTICAST multicast via PGM, possibly encapsulated in UDP
TCP a network based transport
2.建立基础
由于在网络中两个端点是相对动态的,很难有一个稳定的单一连接点。
如果是这种情况,可以使用由0MQ提供的转发设备。
转发设备可以绑定2个不同端口,并且转发消息从一个端点到另一个端点。
这样做的话,在网络中转发设备能够变成一个稳定的点,其它组件都可以去连接。
0MQ提供了3种类型的设备
QUEUE, a forwarder for the request/response messaging pattern
FORWARDER, a forwarder for the publish/subscribe messaging pattern
STREAMER, a forwarder for the pipelined messaging pattern
3.选择通讯模式
0MQ支持4种模式
REQUEST/REPLY, bidirectional, load balanced and state based
PUBLISH/SUBSCRIBE, publish to multiple recipients at once
UPSTREAM / DOWNSTREAM, distribute data to nodes arranged in a pipeline
PAIR, communication exclusively between peers
Req/Rep
均衡负载请求:
server 1
import zmq context = zmq.Context() socket = context.socket(zmq.REP) socket.bind("tcp://127.0.0.1:5000") while True: msg = socket.recv() print "Got", msg socket.send(msg)
server 2
import zmq context = zmq.Context() socket = context.socket(zmq.REP) socket.bind("tcp://127.0.0.1:6000") while True: msg = socket.recv() print "Got", msg socket.send(msg)
client
import zmq context = zmq.Context() socket = context.socket(zmq.REQ) socket.connect("tcp://127.0.0.1:5000") socket.connect("tcp://127.0.0.1:6000") for i in range(10): msg = "msg %s" % i socket.send(msg) print "Sending", msg msg_in = socket.recv()
会发现client的请求会被均衡的分配给两个server
Example client output:
Sending msg 0
Sendingmsg1
Sendingmsg2
Sendingmsg3
Sendingmsg4
Sendingmsg5
Sendingmsg6
Sendingmsg7
Sendingmsg8
Sending msg 9
Example output server 1 at port 5000:
Got msg 0
Gotmsg2
Gotmsg4
Gotmsg6
Got msg 8
Example output server 2 at port 6000:
Got msg 1
Gotmsg3
Gotmsg5
Gotmsg7
Got msg 9
现在,如果我们要加入一个额外的server去管理我们的请求,我们将不得不修改我们的代码。
这是非常麻烦的,我们需要让每个client都知道有一个额外的server可以均衡请求。
为了解决这个问题,替代client直接去连接多个server的方式,client去连接转发设备,再由转发设备路由全部的消息给server。
Pub/Sub
在pub/sub模式下组件是松耦合的。类似于广播电台。
一个广播server为现场足球赛
import zmq from random import choice context = zmq.Context() socket = context.socket(zmq.PUB) socket.bind("tcp://127.0.0.1:5000") countries = ['netherlands','brazil','germany','portugal'] events = ['yellow card', 'red card', 'goal', 'corner', 'foul'] while True: msg = choice( countries ) +" "+ choice( events ) print "->",msg socket.send( msg )
输出
-> portugal corner
->portugalyellowcard
->portugalgoal
->netherlandsyellowcard
->germanyyellowcard
->brazilyellowcard
->portugalgoal
->germanycorner
…一个客户端去收听特定的消息
import zmq context = zmq.Context() socket = context.socket(zmq.SUB) socket.connect("tcp://127.0.0.1:5000") socket.setsockopt(zmq.SUBSCRIBE, "netherlands") socket.setsockopt(zmq.SUBSCRIBE, "germany") while True: print socket.recv()
输出
netherlands red card
netherlandsgoal
netherlandsredcard
germanyfoul
netherlandsyellowcard
germanyfoul
netherlandsgoal
netherlandscorner
germanyfoul
netherlandscorner
…Pipelining
并发处理数据,其工作模式
一个工作者得到来自上游socket的消息,一旦处理完成后发送消息到下游。
Paired socket
服务器监听某个端口,客户端连接到这个端口,消息可以双向流动。
server
import zmq context = zmq.Context() socket = context.socket(zmq.PAIR) socket.bind("tcp://127.0.0.1:5555")
client
import zmq context = zmq.Context() socket = context.socket(zmq.PAIR) socket.connect("tcp://127.0.0.1:5555")
ps:
推荐