ZMQ简单使用

Posted 2020-12-20 zyyhxbs

ZMQ简单使用

一 什么是ZMQ

引用官方的说法： “ZMQ (以下 ZeroMQ 简称 ZMQ)是一个简单好用的传输层，像框架一样的一个 socket library，他使得 Socket 编程更加简单、简洁和性能更高。是一个消息处理队列库，可在多个线程、内核和主机盒之间弹性伸缩。ZMQ 的明确目标是“成为标准网络协议栈的一部分，之后进入 Linux 内核”。现在还未看到它们的成功。但是，它无疑是极具前景的、并且是人们更加需要的“传统”BSD 套接字之上的一层封装。ZMQ 让编写高性能网络应用程序极为简单和有趣。”

近几年有关”Message Queue”的项目层出不穷，知名的就有十几种，这主要是因为后摩尔定律时代，分布式处理逐渐成为主流，业界需要一套标准来解决分布式计算环境中节点之间的消息通信。几年的竞争下来，Apache 基金会旗下的符合 AMQP/1.0标准的 RabbitMQ 已经得到了广泛的认可，成为领先的 MQ 项目。

与 RabbitMQ 相比，ZMQ 并不像是一个传统意义上的消息队列服务器，事实上，它也根本不是一个服务器，它更像是一个底层的网络通讯库，在 Socket API 之上做了一层封装，将网络通讯、进程通讯和线程通讯抽象为统一的 API 接口。

这是个类似于 Socket 的一系列接口，他跟 Socket 的区别是：普通的 socket 是端到端的（1:1的关系），而 ZMQ 却是可以N：M 的关系，人们对 BSD 套接字的了解较多的是点对点的连接，点对点连接需要显式地建立连接、销毁连接、选择协议（TCP/UDP）和处理错误等，而 ZMQ 屏蔽了这些细节，让你的网络编程更为简单。ZMQ 用于 node 与 node 间的通信，node 可以是主机或者是进程。

二 C/S模式

多个客户端，一个服务端，多个客户端可以连接一个服务端发送消息

# -*- coding: utf-8 -*-
# server端
import zmq

context = zmq.Context()
socket = context.socket(zmq.REP)
socket.bind(‘tcp://*:8888‘)

while True:
    message = socket.recv()
    print(message)
    response = ‘server response!‘
    socket.send(response.encode())

# -*- coding: utf-8 -*-
# client端
import zmq

context = zmq.Context()
socket = context.socket(zmq.REQ)
socket.connect(‘tcp://127.0.0.1:8888‘)

while True:
    data = input(‘input your data:‘)
    socket.send(data.encode())
    response = socket.recv()
    print(response)

三 发布订阅模式

一个发布者，多个订阅者，广播所有client，没有队列缓存，断开连接数据将永远丢失。client可以进行数据过滤。

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

contest = zmq.Context()
socket = contest.socket(zmq.PUB)
socket.bind(‘tcp://*:8888‘)

while True:
    data = input(‘input your data:‘)
    socket.send(data.encode())

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

context = zmq.Context()
socket = context.socket(zmq.SUB)
socket.connect(‘tcp://127.0.0.1:8888‘)
socket.setsockopt(zmq.SUBSCRIBE, ‘‘.encode())

while True:
    data = socket.recv()
    print(data)

四 平行管道模式

消息单向的，也是有去无回的。push的任何一个消息，始终只会有一个pull端收到消息。由三部分组成，push进行数据推送，work进行数据缓存，pull进行数据竞争获取处理。区别于Publish-Subscribe存在一个数据缓存和处理负载。当连接被断开，数据不会丢失，重连后数据继续发送到对端。

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

contest = zmq.Context()
socket = contest.socket(zmq.PULL)
socket.bind(‘tcp://*:8888‘)

while True:
    print(socket.recv())

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

context = zmq.Context()
recive = context.socket(zmq.PULL)
recive.connect(‘tcp://127.0.0.1:8887‘)

sender = context.socket(zmq.PUSH)
sender.connect(‘tcp://127.0.0.1:8888‘)

while True:
    data = recive.recv()
    sender.send(data)

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

context = zmq.Context()
socket = context.socket(zmq.PUSH)
socket.bind(‘tcp://*:8887‘)

while True:
    data = input(‘input your data:‘)
    socket.send(data.encode())