重头戏!ZeroMQ的独家对模式详解:ZMQ_PAIR

2024-02-16 03:30

本文主要是介绍重头戏!ZeroMQ的独家对模式详解:ZMQ_PAIR,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

一、ØMQ模式总览

  • ØMQ支持多种模式,具体可以参阅:https://blog.csdn.net/qq_41453285/article/details/106865539
  • 本文介绍ØMQ的独家对模式

二、独家对模式

  • 在前面的文章中我们介绍过如何编写ØMQ多线程程序:https://blog.csdn.net/qq_41453285/article/details/106882216
  • 独家对模式(Exclusive pair)用于将一个对等点精确地连接到另一个对等点。此模式用于跨inproc传输的线程间通信
  • 互斥对模式由http://rfc.zeromq.org/spec:31正式定义
  • 独家对模式支持的套接字类型只有1种:
    • ZMQ_PAIR

三、“PAIR”套接字类型

  • ZMQ_PAIR类型的套接字只能一次连接到单个对等方。对通过ZMQ_PAIR套接字发送的消息不执行消息路由或筛选
  • 当ZMQ_PAIR套接字由于已达到连接对等方的高水位线而进入静音状态时,或者如果没有连接任何对等方,则套接字上的任何zmq_send()操作都应阻塞,直到对等方可用于发送;消息不会被丢弃
  • 适用协议:
    • ZMQ_PAIR套接字设计用于通过nproc传输进行线程间通信,并且不实现自动重新连接等功能
    • 尽管ZMQ_PAIR套接字可用于inproc以外的其他传输协议,但是它们无法自动重新连接,并且当以前存在任何连接(包括关闭状态的连接)时,新的传入连接将被终止,这使得它们在大多数情况下不适合TCP
                                                                                                       ZMQ_PAIR特性摘要 
兼容的对等套接字ZMQ_PAIR
方向双向
发送/接收模式无限制
入网路由策略不适用(N/A)

外发路由策略

不适用(N/A)
静音状态下的操作阻塞

四、PAIR套接字应用场景:协调线程

  • 在编写多线程应用程序时,会遇到如何“协调线程”的问题,例如一个线程状态发生改变时同时另一个线程:
    • 如果使用以往的多线程程序,你可能会使用信号量或互斥等技术
    • 但是在ØMQ中,你可以使用ZMQ_PAIR套接字来进行线程间的通信
  • 下面是一个演示案例,下面创建三个PAIR套接字:
    • PAIR3:主线程中的PAIR套接字,等待PAIR2发来通知消息
    • PAIR2:在主线程中调用pthread_create()创建线程,在线程的回调函数中创建PAIR2套接字,该套接字等待PAIR1发来通知消息
    • PAIR1:PAIR2所在的线程再调用一次pthread_create(),在线程的回调函数中创建PAIR1套接字,PAIR1会向PAIR2发送消息
  • 整体的流程就是:PAIR1发送消息给PAIR2,PAIR2接收到PAIR1的消息之后再发送消息给PAIR3,PAIR3接收到PAIR2的消息之后退出程序

  • 代码如下:
// mtrelay.c
// 源码链接: https://github.com/dongyusheng/csdn-code/blob/master/ZeroMQ/mtrelay.c
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <assert.h>
#include <unistd.h>
#include <pthread.h>
#include <zmq.h>// PAIR2套接字执行函数, 参数为主进程的上下文对象指针
static void *step2(void *arg);// PAIR1套接字执行函数, 参数为主进程的上下文对象指针
static void *step1(void *arg);// 从socket套接字上接收消息
static char *s_recv(void *socket);// 向socket套接字发送消息string
static int s_send(void *socket, char *string);int main()
{int rc;// 1.创建新的上下文对象void *context = zmq_ctx_new();assert(context != NULL);// 2.创建、绑定PAIR3套接字, PAIR2会连接该套接字并向该套接字发送消息void *receiver = zmq_socket(context, ZMQ_PAIR);assert(receiver != NULL);rc = zmq_bind(receiver, "inproc://step3");assert(rc != -1);// 3.创建线程, 线程中创建PAIR2套接字, PAIR2套接字会向PAIR3发送消息pthread_t thread_id;pthread_create(&thread_id, NULL, step2, context);// 4.阻塞等待接收PAIR2发来消息char *string = s_recv(receiver);assert(string != NULL);free(string);// 5.打印消息printf("Test successful!\n");// 6.关闭套接字、销毁上下文zmq_close(receiver);zmq_ctx_destroy(context);return 0;
}static void *step2(void *arg)
{// 参数为上下文对象指针int rc;// 1.创建、绑定PAIR2套接字, PAIR1会连接该套接字并向该套接字发送消息void *receiver = zmq_socket(arg, ZMQ_PAIR);assert(receiver != NULL);rc = zmq_bind(receiver, "inproc://step2");assert(rc != -1);// 2.创建线程, 线程中创建PAIR1套接字, PAIR1套接字会向PAIR2发送消息pthread_t thread_id;pthread_create(&thread_id, NULL, step1, arg);// 3.等待接收PAIR1发来消息char *string = s_recv(receiver);assert(string != NULL);free(string);// 关闭PAIR2套接字zmq_close(receiver);// 接收到PAIR1发来消息之后, 再开始向PAIR3发送通知// 4.重新创建一个PAIR2套接字, 该套接连接PAIR3void *xmitter = zmq_socket(arg, ZMQ_PAIR);assert(xmitter != NULL);rc = zmq_connect(xmitter, "inproc://step3");assert(rc != -1);// 5.向PAIR3发送通知printf("Step 2 ready, signaling step 3\n");rc = s_send(xmitter, "READY");assert(rc != -1);zmq_close(xmitter);return NULL;
}static void *step1(void *arg)
{// 参数为上下文对象指针int rc;// 创建一个PAIR1套接字, 然后连接PAIR2套接字void *xmitter = zmq_socket(arg, ZMQ_PAIR);assert(xmitter != NULL);rc = zmq_connect(xmitter, "inproc://step2");assert(rc != -1);// 5.向PAIR2发送通知printf("Step 1 ready, signaling step 2\n");rc = s_send(xmitter, "READY");assert(rc != -1);zmq_close(xmitter);return NULL;
}static char *s_recv(void *socket)
{int size;zmq_msg_t msg;zmq_msg_init(&msg);size = zmq_msg_recv(&msg, socket, 0);if(size == -1)return NULL;char *string = (char*)malloc(size + 1);if(string == NULL)return NULL;memcpy(string, zmq_msg_data(&msg), size);zmq_msg_close(&msg);string[size] = 0;return string;
}static int s_send(void *socket, char *string)
{int rc;zmq_msg_t msg;zmq_msg_init_size(&msg, strlen(string));memcpy(zmq_msg_data(&msg), string, strlen(string));rc = zmq_msg_send(&msg, socket, 0);zmq_msg_close(&msg);return rc;
}
  • 编译运行如下:
gcc -o mtrelay mtrelay.c -lzmq

案例分析

  • 这是使用ØMQ进行多线程编程的一个经典模式:
    • 1.两个线程通过inproc通信,使用的是共享的上下文
    • 2.父线程创建一个套接字,将其绑定到一个inproc端点,然后启动子线程,将上下文传递给它
    • 3.子线程创建第二个套接字,将它连接到该inproc端点,然后发信号告诉父线程,它已准备就绪
  • 使用这种模式的多线程代码是不可扩展到进程的。如果你使用inproc和套接字对,你就正在构建一个紧耦合的应用程序,也就是说,其中你的线程在结构是相互依存的,只有在低延迟真的很重要的时候才这样做。另一种设计模式是一个松耦合的应用程序,其中的线程有自己的上下文并通过ipc或tcp通信。你可以轻松地将松耦合的线程分解为单独的进程

为什么选择的是PAIR?

  • 此处使用的是PAIR套接字,其他套接字组合也能够完成上面相同的工作,但是其他台套接字都有副作用,可能会干扰信令:
    • 你可以让发送者使用PUSH并让接收者使用PULL,但是PUSH会把消息发送给所有存在的接收者,假设你启动了2个接收者,那么就会“丢失”一半的信号。PAIR具有拒绝多个连接,两个连接的组成的对是独占的
    • 你可以让发送者使用DEALER并让接收者使用ROUTER,但是ROUTER将你的信息包装在一个“封包”中,这意味着你的大小为0的信号变成了一个多部分消息。如果你不关心数据并把任何东西都当做一个有效的信号,并且如果你不会不止一次地从套接字上读取,这并不重要。但是,如果你决定要发送实际数据时,你会突然发现ROUTER为你提供了“错误”的消息。DEALER也分发传出消息,这带来与PUSH相同的风险
    • 你可以让发送者使用PUB,而让接收者使用SUB,这将完全按照你发送它们的原样正确地传递你的消息,而且PUB不像PUSH或DEALER那样分发消息。但是,你需要用空订阅配置订阅者,这比较麻烦,更糟的是,PUB-SUB链接的可靠性是与实践相关的,并且如果在PUB套接字发送消息时,SUB套接字正在连接,信息就有可能会丢失
  • 综合以上的原因,使得PAIR成为线程对之间协调的最佳选择

五、节点协调

  • 当你想协调节点时,PAIR套接字就无法正常工作了,这是线程和节点的策略不同的少数地方之一。详情参阅下一篇文章:https://blog.csdn.net/qq_41453285/article/details/106949903。

  • 我是小董,V公众点击"笔记白嫖"解锁更多【ZeroMQ】资料内容。

这篇关于重头戏!ZeroMQ的独家对模式详解:ZMQ_PAIR的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/713363

相关文章

MySQL数据库双机热备的配置方法详解

《MySQL数据库双机热备的配置方法详解》在企业级应用中,数据库的高可用性和数据的安全性是至关重要的,MySQL作为最流行的开源关系型数据库管理系统之一,提供了多种方式来实现高可用性,其中双机热备(M... 目录1. 环境准备1.1 安装mysql1.2 配置MySQL1.2.1 主服务器配置1.2.2 从

Linux kill正在执行的后台任务 kill进程组使用详解

《Linuxkill正在执行的后台任务kill进程组使用详解》文章介绍了两个脚本的功能和区别,以及执行这些脚本时遇到的进程管理问题,通过查看进程树、使用`kill`命令和`lsof`命令,分析了子... 目录零. 用到的命令一. 待执行的脚本二. 执行含子进程的脚本,并kill2.1 进程查看2.2 遇到的

MyBatis常用XML语法详解

《MyBatis常用XML语法详解》文章介绍了MyBatis常用XML语法,包括结果映射、查询语句、插入语句、更新语句、删除语句、动态SQL标签以及ehcache.xml文件的使用,感兴趣的朋友跟随小... 目录1、定义结果映射2、查询语句3、插入语句4、更新语句5、删除语句6、动态 SQL 标签7、ehc

详解SpringBoot+Ehcache使用示例

《详解SpringBoot+Ehcache使用示例》本文介绍了SpringBoot中配置Ehcache、自定义get/set方式,并实际使用缓存的过程,文中通过示例代码介绍的非常详细,对大家的学习或者... 目录摘要概念内存与磁盘持久化存储:配置灵活性:编码示例引入依赖:配置ehcache.XML文件:配置

从基础到高级详解Go语言中错误处理的实践指南

《从基础到高级详解Go语言中错误处理的实践指南》Go语言采用了一种独特而明确的错误处理哲学,与其他主流编程语言形成鲜明对比,本文将为大家详细介绍Go语言中错误处理详细方法,希望对大家有所帮助... 目录1 Go 错误处理哲学与核心机制1.1 错误接口设计1.2 错误与异常的区别2 错误创建与检查2.1 基础

k8s按需创建PV和使用PVC详解

《k8s按需创建PV和使用PVC详解》Kubernetes中,PV和PVC用于管理持久存储,StorageClass实现动态PV分配,PVC声明存储需求并绑定PV,通过kubectl验证状态,注意回收... 目录1.按需创建 PV(使用 StorageClass)创建 StorageClass2.创建 PV

Python版本信息获取方法详解与实战

《Python版本信息获取方法详解与实战》在Python开发中,获取Python版本号是调试、兼容性检查和版本控制的重要基础操作,本文详细介绍了如何使用sys和platform模块获取Python的主... 目录1. python版本号获取基础2. 使用sys模块获取版本信息2.1 sys模块概述2.1.1

一文详解Python如何开发游戏

《一文详解Python如何开发游戏》Python是一种非常流行的编程语言,也可以用来开发游戏模组,:本文主要介绍Python如何开发游戏的相关资料,文中通过代码介绍的非常详细,需要的朋友可以参考下... 目录一、python简介二、Python 开发 2D 游戏的优劣势优势缺点三、Python 开发 3D

Redis 基本数据类型和使用详解

《Redis基本数据类型和使用详解》String是Redis最基本的数据类型,一个键对应一个值,它的功能十分强大,可以存储字符串、整数、浮点数等多种数据格式,本文给大家介绍Redis基本数据类型和... 目录一、Redis 入门介绍二、Redis 的五大基本数据类型2.1 String 类型2.2 Hash

Java中的.close()举例详解

《Java中的.close()举例详解》.close()方法只适用于通过window.open()打开的弹出窗口,对于浏览器的主窗口,如果没有得到用户允许是不能关闭的,:本文主要介绍Java中的.... 目录当你遇到以下三种情况时,一定要记得使用 .close():用法作用举例如何判断代码中的 input