进程间通信中间件---ZeroMQ

devtools/2025/2/26 19:28:58/

ZeroMQ（也称为 ØMQ 或 0MQ）是一个高性能的异步消息传递库，专为分布式或并发应用程序设计。它提供了多种通信模式（如请求-响应、发布-订阅等），并且可以在多种传输协议（如 TCP、IPC、PGM 等）上运行。ZeroMQ 的设计理念是“消息队列”，但它并不像传统的消息队列那样需要单独的中间件服务器，而是直接嵌入到应用程序中。

为什么选择ZeroMQ ？

在我使用过程中感受到最直观的一点就是“便捷高效”，我使用过NATS,RabbitMQ,Apache Kafka等中间件都需要安装一系列服务器，还需要配置服务器等等，很是麻烦，并且也会给系统带来一些负担，而zeroMQ只需要下载一个zeroMQ的包就可以直接在程序中调用  #include <zmq.h>  就可以实现编辑了，并且拥有多语言支持，不只是支持c语言，还支持python,java,GO,C++等语言，zeroMQ还有轻量级的特点，可以直接嵌入到应用程序中，而有的中间件需要java环境，有的需要很多包才可以运行，然而这对于内存敏感的嵌入式系统是致命缺点！所以这是我选择zeroMQ中间件的原因。

ZeroMQ通信模式

无论选择什么通信模式，可供选择的通信协议有：TCP , IPC , PGM(基于IP的标准组播) , EPGM(基于UDP的组播) , In-Process(线程间通信) , TIPC(集群通信)

测试用Makefile

首先给出一个Makefile编译目录下所有的.c文件，使用方法：

make zmq

 Makefile

# 指定编译器
CC = gcc# 默认编译选项
CFLAGS = -Wall -g# 获取当前目录下所有的 .c 文件并生成目标文件名
SRCS := $(wildcard *.c)
OBJS := $(SRCS:.c=)# 默认目标：编译所有程序
all: $(OBJS)# 判断是否启用了 zmq 目标
ifeq ($(findstring zmq,$(MAKECMDGOALS)),zmq)# 如果启用了 zmq 目标，添加 ZeroMQ 相关的编译和链接选项CFLAGS += -DUSE_ZMQLDFLAGS += -lzmq
endif# 编译规则：将每个 .c 文件编译为同名的可执行文件
%: %.c$(CC) $(CFLAGS) $< $(LDFLAGS) -o $@# 让 zmq 目标依赖于所有可执行文件
zmq: $(OBJS)# 清理生成的可执行文件
clean:rm -f $(OBJS).PHONY: all clean zmq

请求-响应（Request-Reply）

双向通信：客户端发送请求，服务器返回响应

client.c

#include <stdio.h>
#include <zmq.h>
#include <string.h>
#include <unistd.h>int main(void) {// 初始化 ZeroMQ 上下文void *context = zmq_ctx_new();// 创建一个 REQ 类型的套接字void *requester = zmq_socket(context, ZMQ_REQ);zmq_connect(requester, "ipc://test_ipc");int count = 0;while (1) {char buffer[256];snprintf(buffer, sizeof(buffer), "Task %d", count++);printf("Sending task: %s\n", buffer);// 发送任务zmq_send(requester, buffer, strlen(buffer), 0);// 接收回复//char buffer[1024] = {0};zmq_recv(requester, buffer, sizeof(buffer) - 1, 0);printf("Received reply: %s\n", buffer);sleep(1); // 每秒发送一次任务}// 清理资源zmq_close(requester);zmq_ctx_destroy(context);return 0;
}

 server.c

#include <stdio.h>
#include <zmq.h>
#include <string.h>int main(void) {// 初始化 ZeroMQ 上下文void *context = zmq_ctx_new();// 创建一个 REP 类型的套接字void *responder = zmq_socket(context, ZMQ_REP);int bind_result = zmq_bind(responder, "ipc://test_ipc");if (bind_result != 0) {printf("Failed to bind socket\n");return -1;}while (1) {// 接收消息char buffer[1024] = {0};zmq_recv(responder, buffer, sizeof(buffer) - 1, 0);printf("Received: %s\n", buffer);// 发送回复const char *response = "World";zmq_send(responder, response, strlen(response), 0);}// 清理资源zmq_close(responder);zmq_ctx_destroy(context);return 0;
}

发布-订阅（Publish-Subscribe）

广播通信：发布者广播消息，所有感兴趣的订阅者接收消息

publisher.c

#include <stdio.h>
#include <zmq.h>
#include <unistd.h>
#include <string.h>int main(void) {void *context = zmq_ctx_new();void *publisher = zmq_socket(context, ZMQ_PUB);zmq_bind(publisher, "tcp://*:5556");int count = 0;while (1) {char buffer[256];snprintf(buffer, sizeof(buffer), "Message %d", count++);printf("Publishing: %s\n", buffer);// 发送消息zmq_send(publisher, buffer, strlen(buffer), 0);sleep(1); // 每秒发送一次消息}zmq_close(publisher);zmq_ctx_destroy(context);return 0;
}

subscriber.c

#include <stdio.h>
#include <zmq.h>int main(void) {void *context = zmq_ctx_new();void *subscriber = zmq_socket(context, ZMQ_SUB);zmq_connect(subscriber, "tcp://localhost:5556");// 设置订阅过滤器（空字符串表示接收所有消息）zmq_setsockopt(subscriber, ZMQ_SUBSCRIBE, "", 0);while (1) {char buffer[256] = {0};zmq_recv(subscriber, buffer, sizeof(buffer) - 1, 0);printf("Received: %s\n", buffer);}zmq_close(subscriber);zmq_ctx_destroy(context);return 0;
}

 消息过滤 (  指定订阅 )

在PUB-SUB模式下有个特点是可以指定订阅，达到信息分类接收的目的，关键在于main函数运行一次的这句话

zmq_setsockopt (void *s_, int option_, const void *optval_, size_t optvallen_);
不指定消息类型：
// 设置订阅过滤器（空字符串表示接收所有消息）
zmq_setsockopt(subscriber, ZMQ_SUBSCRIBE, "", 0);
指定消息类型：
//指定接收话题A的信息
zmq_setsockopt(subscriber, ZMQ_SUBSCRIBE, "TopicA.", 7); // 只接收以 "TopicA." 开头的消息

那么现在订阅者只接收话题A，所以我发布端也需要修改发布的内容：

 main函数如下

#include <stdio.h>
#include <zmq.h>
#include <unistd.h>
#include <string.h>void *publisher = NULL;
int count = 0;void TopicA_Send(void){char endbuffer[256];// 发送话题A消息snprintf(endbuffer,sizeof(endbuffer), "TopicA.Message %d", count);zmq_send(publisher, endbuffer, strlen(endbuffer), 0);printf("AAAAPublishing: %s\n", endbuffer);
}void TopicB_Send(void){char endbuffer[256];// 发送话题B消息snprintf(endbuffer,sizeof(endbuffer), "TopicB.Message %d", count);zmq_send(publisher, endbuffer, strlen(endbuffer), 0);printf("BBBBPublishing: %s\n", endbuffer);
}int main(void) {void *context = zmq_ctx_new();publisher = zmq_socket(context, ZMQ_PUB);zmq_bind(publisher, "ipc://ipc_test");while (1) {TopicB_Send();sleep(1); // 每秒发送一次消息TopicA_Send();sleep(1); // 每秒发送一次消息count++;}zmq_close(publisher);zmq_ctx_destroy(context);return 0;
}

 关键点在于，在发送的时候在前面加入了 TopicB. / TopicA. 达到分话题发布的效果：

 snprintf(endbuffer,sizeof(endbuffer), "TopicA.Message %d", count);zmq_send(publisher, endbuffer, strlen(endbuffer), 0);

再次复制一个subscriber.c文件重命名后面加一个B，只修改这一句话

 zmq_setsockopt(subscriber, ZMQ_SUBSCRIBE, "TopicB.", 7); // 只接收以 "TopicB." 开头的消息

 就可以达到分开接收话题B的效果

实际运行效果

实际运行起来是：

 可以看到订阅者接收了自己对应的话题并且打印出来接收的信息，原理是根据前几个字符过滤信息。

推送-拉取（Push-Pull）

单向通信：pusher推送任务，puller拉取任务

puller.c

#include <stdio.h>
#include <zmq.h>int main(void) {void *context = zmq_ctx_new();void *puller = zmq_socket(context, ZMQ_PULL);zmq_bind(puller, "tcp://*:5557");while (1) {char buffer[256] = {0};zmq_recv(puller, buffer, sizeof(buffer) - 1, 0);printf("Received task: %s\n", buffer);}zmq_close(puller);zmq_ctx_destroy(context);return 0;
}

pusher.c

#include <stdio.h>
#include <zmq.h>
#include <unistd.h>
#include <string.h>int main(void) {void *context = zmq_ctx_new();void *pusher = zmq_socket(context, ZMQ_PUSH);zmq_connect(pusher, "tcp://localhost:5557");int count = 0;while (1) {char buffer[256];snprintf(buffer, sizeof(buffer), "Task %d", count++);printf("Sending task: %s\n", buffer);// 发送任务zmq_send(pusher, buffer, strlen(buffer), 0);sleep(1); // 每秒发送一次任务}zmq_close(pusher);zmq_ctx_destroy(context);return 0;
}

思考

Publish-Subscribe 还是 Push-Pull ？
  • 在上面的 发布-订阅 和 推送-拉取 看起来比较相似，都是一收一发，单向通信，只是发布-订阅模式可以过滤话题，但是我pull的时候也可以通过软件过滤前几行字，那为什么还要这个发布-订阅模式呢？

他们有以下区别：

假设我们有两个消费者（Consumer），一个生产者（Producer）。

(1) Push-Pull 模式
  • 生产者生成 6 条消息：Task 1, Task 2, Task 3, Task 4, Task 5, Task 6。
  • 如果有两个 Pull 端，ZeroMQ 会按以下顺序分发消息：
    • Pull 端 1 接收：Task 1, Task 3, Task 5
    • Pull 端 2 接收：Task 2, Task 4, Task 6
  • 所以在一个push的时候多个pull的情况下信息会逐条、均衡的发给各个接收端达到均衡负载的目的！，相当于一对一的发送！
(2) Publish-Subscribe 模式
  • 生产者生成 6 条消息：Topic1 Task 1, Topic1 Task 2, Topic2 Task 3, Topic1 Task 4, Topic2 Task 5, Topic1 Task 6。
  • 如果有两个订阅者，并且订阅者 1 订阅了 Topic1，订阅者 2 订阅了 Topic2，那么：
    • 订阅者 1 接收：Topic1 Task 1, Topic1 Task 2, Topic1 Task 4, Topic1 Task 6
    • 订阅者 2 接收：Topic2 Task 3, Topic2 Task 5

Push-Pull 模式 是一种逐条分发消息的模式，适合任务分发和负载均衡。

Publish-Subscribe 模式 是一种广播分发消息的模式，适合实时数据广播和事件通知。

例子

一个pusher端 两个puller 端

// pusher.c
#include <zmq.h>
#include <stdio.h>
#include <unistd.h>
#include <string.h>int main(void) {void *context = zmq_ctx_new(); // 创建 ZeroMQ 上下文void *pusher = zmq_socket(context, ZMQ_PUSH); // 创建 PUSH 套接字zmq_bind(pusher, "tcp://*:5557"); // 绑定地址printf("Pusher started. Sending tasks...\n");for (int i = 1; i <= 10; i++) {char buffer[256];snprintf(buffer, sizeof(buffer), "Task %d", i); // 构造任务消息printf("Sending task: %s\n", buffer);zmq_send(pusher, buffer, strlen(buffer), 0); // 发送任务sleep(1); // 模拟任务生成间隔}zmq_close(pusher); // 关闭套接字zmq_ctx_destroy(context); // 销毁上下文return 0;
}
// puller.c
#include <zmq.h>
#include <stdio.h>
#include <string.h>
#include <unistd.h>
int main(int argc, char *argv[]) {if (argc != 2) {printf("Usage: puller <identity>\n");return 1;}const char *identity = argv[1]; // Pull 端的身份标识（用于区分不同的 Pull 端）void *context = zmq_ctx_new(); // 创建 ZeroMQ 上下文void *puller = zmq_socket(context, ZMQ_PULL); // 创建 PULL 套接字zmq_connect(puller, "tcp://localhost:5557"); // 连接到 Push 端printf("Puller %s started. Waiting for tasks...\n", identity);while (1) {char buffer[256];zmq_recv(puller, buffer, sizeof(buffer), 0); // 接收任务printf("Puller %s received task: %s\n", identity, buffer);// 模拟处理时间sleep(1);}zmq_close(puller);
}

结果：（注意在 ./puller X）后面更上puller的端点号

1.打开puller 等待接收

2.开始push，看到pull轮流接收信息处理

注意这里的puller分配机制不靠输入的arge[]标识，靠自己的机制

 

总的来说，

push-pull 关键在于分发任务均衡处理，同一文件不同进程运行，减少压力

Publish-Subscribe 关键在于订阅特定消息，实现特定消息特定进程处理


http://www.ppmy.cn/devtools/162868.html

相关文章

大语言模型（LLM）提示词（Prompt）高阶撰写指南

大语言模型（LLM）提示词（Prompt）高阶撰写指南

——结构化思维与工程化实践 一、LLM提示词设计的核心逻辑 1. 本质认知 LLM是「超强模式识别器概率生成器」&#xff0c;提示词的本质是构建数据分布约束&#xff0c;通过语义信号引导模型激活特定知识路径。优秀提示词需实现&#xff1a; 精准性&#xff1a;消除歧义&#…
阅读更多...
seacmsv9管理员账号+密码注入

seacmsv9管理员账号+密码注入

Seacms v9 SQL 注入漏洞分析与利用 1. 漏洞概述 Seacms&#xff08;海洋 CMS&#xff09;是一款基于 PHP5.X MySQL 架构的视频点播系统&#xff0c;被广泛用于影视站点管理。在 Seacms v9 版本中&#xff0c;./comment/api/index.php 存在 SQL 注入漏洞&#xff0c;漏洞参数…
阅读更多...
【Linux】初探信号的奥秘

【Linux】初探信号的奥秘

目录 一、引入信号&#xff1a; 1、什么是信号&#xff1a; 二、前后台进程&#xff1a; 三、信号的处理方式&#xff1a; 四、键盘数据与信号&#xff1a; 前言&#xff1a; 在Linux系统编程中&#xff0c;信号&#xff08;Signal&#xff09;是一种至关重要的进程间通信…
阅读更多...
面试之《nodejs中，网络请求时stream和json的区别》

面试之《nodejs中，网络请求时stream和json的区别》

在 Node.js 网络请求中&#xff0c;stream&#xff08;流&#xff09;和 JSON&#xff08;JavaScript 对象表示法&#xff09;是两种不同的数据处理方式&#xff0c;它们在数据传输、处理方式、适用场景等方面存在明显区别&#xff0c;以下为你详细介绍&#xff1a; 数据格式和…
阅读更多...
【愚公系列】《Python网络爬虫从入门到精通》034-DataFrame简单数据计算整理

【愚公系列】《Python网络爬虫从入门到精通》034-DataFrame简单数据计算整理

标题详情作者简介愚公搬代码头衔华为云特约编辑，华为云云享专家，华为开发者专家，华为产品云测专家，CSDN博客专家，CSDN商业化专家，阿里云专家博主，阿里云签约作者，腾讯云优秀博主，腾讯云内容共创官，掘金优秀博主，亚马逊技领云博主，51CTO博客专家等。近期荣誉2022年度…
阅读更多...
【Rust中级教程】2.8. API设计原则之灵活性(flexible) Pt.4：显式析构函数的问题及3种解决方案

【Rust中级教程】2.8. API设计原则之灵活性(flexible) Pt.4：显式析构函数的问题及3种解决方案

喜欢的话别忘了点赞、收藏加关注哦&#xff08;加关注即可阅读全文&#xff09;&#xff0c;对接下来的教程有兴趣的可以关注专栏。谢谢喵&#xff01;(&#xff65;ω&#xff65;) 说句题外话&#xff0c;这篇文章一共5721个字&#xff0c;是我截至目前写的最长的一篇文章&a…
阅读更多...
社群团购平台的愿景构建与开源链动2+1模式S2B2C商城小程序应用探索

社群团购平台的愿景构建与开源链动2+1模式S2B2C商城小程序应用探索

摘要&#xff1a;在数字经济背景下&#xff0c;社群团购作为一种新兴的商业模式&#xff0c;凭借其独特的互动性和便捷性&#xff0c;展现出巨大的市场潜力。本文旨在探讨社群团购平台愿景的构建策略&#xff0c;并结合开源链动21模式S2B2C商城小程序的应用&#xff0c;为创业者…
阅读更多...
【linux配置】 修改内核网络参数

【linux配置】 修改内核网络参数

命令解释 echo 1 > /proc/sys/net/ipv4/conf/all/arp_ignore 这个命令的具体含义是&#xff1a; echo 1&#xff1a;将值1写入文件。 /proc/sys/net/ipv4/conf/all/arp_ignore&#xff1a;将值1写入 /proc/sys/net/ipv4/conf/all/arp_ignore 文件&#xff0c;从而修改内核参…
阅读更多...
最新文章

Copyright @ 2022~2023