详解Python 实现 ZeroMQ 的三种基本工作模式

 更新时间:2020-03-25 02:16:09   作者:佚名   我要评论(0)

简介


引用官方说法:ZMQ(以下 ZeroMQ 简称 ZMQ)是一个简单好用的传输层,像框架一样的一个 socket library,他使得 Socket 编程更加简单、简洁和性能更高。

简介

引用官方说法:ZMQ(以下 ZeroMQ 简称 ZMQ)是一个简单好用的传输层,像框架一样的一个 socket library,他使得 Socket 编程更加简单、简洁和性能更高。

是一个消息处理队列库,可在多个线程、内核和主机盒之间弹性伸缩。

ZMQ 的明确目标是“成为标准网络协议栈的一部分,之后进入 Linux 内核”。现在还未看到它们的成功。但是,它无疑是极具前景的、并且是人们更加需要的“传统” BSD 套接字之上的一 层封装。ZMQ 让编写高性能网络应用程序极为简单和有趣。

它跟 RabbitMQ,ActiveMQ 之类有着相当本质的区别,ZeroMQ 根本就不是一个消息队列服务器,更像是一组底层网络通讯库,对原有的 Socket API 加上一层封装,使我们操作更简便。

三种工作模式

Request-Reply 模式:

说到“请求-应答”模式,不得不说的就是它的消息流动模型。消息流动模型指的是该模式下,必须严格遵守“一问一答”的方式。

发出消息后,若没有收到回复,再发出第二条消息时就会抛出异常。同样的,对于 Rep 也是,在没有接收到消息前,不允许发出消息。

基于此构成“一问一答”的响应模式。

server:

# -*- coding=utf-8 -*-

import zmq

context = zmq.Context()
socket = context.socket(zmq.REP)
socket.bind("tcp://*:5555")

while True:
 message = socket.recv()
 print("Received: %s" % message)
 socket.send("I am OK!")

client:

# -*- coding=utf-8 -*-

import zmq
context = zmq.Context()
socket = context.socket(zmq.REQ)
socket.connect("tcp://localhost:5555")

socket.send('Are you OK?')
response = socket.recv()
print("response: %s" % response)

Publish-Subscribe 模式:

“发布-订阅”模式下,“发布者”绑定一个指定的地址,例如“192.168.10.1:5500”,“订阅者”连接到该地址。该模式下消息流是单向的,只允许从“发布者”流向“订阅者”。且“发布者”只管发消息,不理会是否存在“订阅者”。一个“发布者”可以拥有多个订阅者,同样的,一个“订阅者”也可订阅多个发布者。

虽然我们知道“发布者”在发送消息时是不关心“订阅者”的存在于否,所以先启动“发布者”,再启动“订阅者”是很容易导致部分消息丢失的。那么可能会提出一个说法“我先启动‘订阅者',再启动‘发布者',就能解决这个问题了?”

对于 ZeroMQ 而言,这种做法也并不能保证 100% 的可靠性。在 ZeroMQ 领域中,有一个叫做“慢木匠”的术语,就是说即使我是先启动了“订阅者”,再启动“发布者”,“订阅者”总是会丢失第一批数据。因为在“订阅者”与端点建立 TCP 连接时,会包含几毫秒的握手时间,虽然时间短,但是是存在的。再加上 ZeroMQ 后台 IO 是以一部方式执行的,所以若不在双方之间施加同步策略,消息丢失是不可避免的。

关于“发布-订阅”模式在 ZeroMQ 中的一些其他特点:

  • 公平排队,一个“订阅者”连接到多个发布者时,会均衡的从每个“发布者”读取消息,不会出现一个“发布者”淹没其他“发布者”的情况。
  • ZMQ3.0 以上的版本,过滤规则发生在“发布方”。 ZMQ3.0 以下的版本,过滤规则发生在“订阅方”。其实也就是处理消息的位置。

server:

# -*- coding=utf-8 -*-

import zmq
import time

context = zmq.Context()
socket = context.socket(zmq.PUB)
socket.bind("tcp://*:5555")

for i in range(10):
 print('send message...' + str(i))
 socket.send('message' + str(i))
 time.sleep(1)

client:

# -*- coding=utf-8 -*-

import zmq

context = zmq.Context()
socket = context.socket(zmq.SUB)
socket.connect("tcp://localhost:5555")
socket.setsockopt(zmq.SUBSCRIBE, '')
while True:
 response = socket.recv()
 print("response: %s" % response)

Parallel Pipeline 模式:

在说明“管道模式”前,需要明确的是在 ZeroMQ 中并没有绝对的服务端与客户端之分,所有的数据接收与发送都是以连接为单位的,只区分 ZeroMQ 定义的类型。就像套接字绑定地址时,可以使用 bind ,也可以使用 connect ,只是通常我们将理解中的服务端 bind 到一个地址,而理解中的客户端 connec 到该地址。

“管道模式”一般用于任务分发与结果收集,由一个任务发生器来产生任务,“公平”的派发到其管辖下的所有 worker,完成后再由结果收集器来回收任务的执行结果。

整体流程比较好理解,worker 连接到任务发生器上,等待任务的产生,完成后将结果发送至结果收集器。如果要以客户端服务端的概念来区分,这里的任务发生器与结果收集器是服务端,而 worker 是客户端。

前面说到了这里任务的派发是“公平的”,因为内部采用了 LRU 的算法来找到最近最久未工作的闲置 worker。但是公平在这里是相对的,当任务发生器启动后,第一个连接到它的 worker 会在一瞬间承受整个任务发生器产生的 tasks。

总结来说由三部分组成,push 进行数据推送,work 进行数据缓存,pull 进行数据竞争获取处理。区别于 Publish-Subscribe 存在一个数据缓存和处理负载。

当连接被断开,数据不会丢失,重连后数据继续发送到对端。

server:

# -*- coding=utf-8 -*-

import zmq
import time

context = zmq.Context()
socket = context.socket(zmq.PUSH)
socket.bind("tcp://*:5557")

for i in range(10):
 socket.send('message' + str(i))
 # 没启 worker 时不会发消息
 print('send message...' + str(i))
 time.sleep(1)

work:

# -*- coding=utf-8 -*-
import zmq
context = zmq.Context()
receive = context.socket(zmq.PULL)
receive.connect('tcp://127.0.0.1:5557')

sender = context.socket(zmq.PUSH)
sender.connect('tcp://127.0.0.1:5558')

while True:
 data = receive.recv()
 print('transform...' + data)
 sender.send(data)

client:

# -*- coding=utf-8 -*-
import zmq

context = zmq.Context()
socket = context.socket(zmq.PULL)
socket.bind("tcp://*:5558")

while True:
 response = socket.recv()
 print("response: %s" % response)

以上。

参考文档:

https://www.jb51.net/article/177043.htm

总结

到此这篇关于详解Python 实现 ZeroMQ 的三种基本工作模式的文章就介绍到这了,更多相关python ZeroMQ工作模式内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家!

您可能感兴趣的文章:

  • 关于ZeroMQ 三种模式python3实现方式

相关文章

  • 详解Python 实现 ZeroMQ 的三种基本工作模式

    详解Python 实现 ZeroMQ 的三种基本工作模式

    简介 引用官方说法:ZMQ(以下 ZeroMQ 简称 ZMQ)是一个简单好用的传输层,像框架一样的一个 socket library,他使得 Socket 编程更加简单、简洁和性能更高。 是
    2020-03-25
  • 微信小程序 flexbox layout快速实现基本布局的解决方案

    微信小程序 flexbox layout快速实现基本布局的解决方案

    问题描述 flexbox layout ——弹性盒子布局。弹性盒子可以快速的对小程序进行布局。一般传统的小程序布局方法对内容量少的页面而言很方便,但对页面比较复杂的来讲
    2020-03-25
  • 解决Android Studio Gradle Metadata特别慢的问题

    解决Android Studio Gradle Metadata特别慢的问题

    如下所示: 更改build.gradle buildscript { repositories { // jcenter() // jcenter(){ url 'http://jcenter.bintray.com/'} maven{url 'http://m
    2020-03-25
  • Java线程优先级和守护线程原理解析

    Java线程优先级和守护线程原理解析

    一、线程优先级的介绍 java 中的线程优先级的范围是1~10,默认的优先级是5。“高优先级线程”会优先于“低优先级线程”执行。 java 中有两种线程:用户线程和守护线
    2020-03-25
  • Android实现获取meta-data和build.gradle的值

    Android实现获取meta-data和build.gradle的值

    有时候会用到meta-data中的参数,比如定义的渠道号,类似友盟统计。也会用到定义在build.gradle文件中android标签下的defaultConfig标签下,添加manifestPlaceholde
    2020-03-25
  • Java如何使用interrupt()终止线程

    Java如何使用interrupt()终止线程

    一、interrupt() 说明 interrupt()的作用是中断本线程。 本线程中断自己是被允许的;其它线程调用本线程的interrupt()方法时,会通过checkAccess()检查权限。这有
    2020-03-25
  • Android获取清单文件中的meta-data,解决碰到数值为null的问题

    Android获取清单文件中的meta-data,解决碰到数值为null的问题

    1.meta-data是什么?如何获取meta-data? 在AndroidManifest.xml中,元素可以作为子元素,被包在activity、application 、service、或者receiver元素中,不同的父元
    2020-03-25
  • 安装多版本Vue-CLI的实现方法

    安装多版本Vue-CLI的实现方法

    安装多版本Vue-CLI,即在电脑上同时安装多个版本的Vue-CLI。那么为什么要安装多个版本呢?原因是Vue-cli3.0版本与之前的版本有了较大不同,当你要处理3.0之前的项
    2020-03-24
  • Dialog全屏,去掉状态栏的方式

    Dialog全屏,去掉状态栏的方式

    dialog即使设置全屏了,但还是有状态栏占用高度这; 直接将下面这行代码放到你的dialog中即可 @Override protected void onStart() { super.onStart(); int
    2020-03-24
  • ios uicollectionview实现横向滚动

    ios uicollectionview实现横向滚动

    现在使用卡片效果的app很多,之前公司让实现一种卡片效果,就写了一篇关于实现卡片的文章。文章最后附有demo 实现上我选择了使用UICollectionView ;用UICollecti
    2020-03-24

最新评论