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

server/2025/2/26 19:06:23/

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/server/170823.html

相关文章

加油站小程序实战教程02数据源设计

加油站小程序实战教程02数据源设计

目录 一、引言二、需求分析三、表结构设计思路四、关键设计要点五、总结 一、引言 在移动互联网时代&#xff0c;小程序已成为连接用户与服务的重要桥梁。以加油小程序为例&#xff0c;其核心功能涉及地图定位、加油站展示、加油下单、钱包管理、优惠券、订单管理以及发票、车…
阅读更多...
体验用ai做了个python小游戏

体验用ai做了个python小游戏

体验用ai做了个python小游戏 写在前面使用的工具2.增加功能1.要求增加视频作为背景。2.我让增加了一个欢迎页面。3.我发现中文显示有问题。4.我提出了背景修改意见&#xff0c;欢迎页面和结束页面背景是视频&#xff0c;游戏页面背景是静态图片。5.提出增加更多游戏元素。 总结…
阅读更多...
Nginx的安装和部署以及Nginx的反向代理与负载均衡

Nginx的安装和部署以及Nginx的反向代理与负载均衡

Nginx的安装和部署以及Nginx的反向代理与负载均衡 1. 本文内容 Nginx的安装Nginx的静态网站部署Nginx的反向代理与负载均衡&#xff0c;配置反向代理与负载均衡 2. Nginx的安装与启动 2.1 什么是Nginx Nginx是一款高性能的http服务器/反向代理服务器及电子邮件&#xff08…
阅读更多...
【多模态处理篇二】【深度揭秘：DeepSeek视频理解之时空注意力机制解析】

【多模态处理篇二】【深度揭秘：DeepSeek视频理解之时空注意力机制解析】

一、为啥要搞视频理解这事儿 咱先唠唠为啥视频理解这么重要哈。现在这互联网时代，视频那可是铺天盖地的。你刷短视频平台，看在线电影，玩游戏直播，到处都是视频。但是计算机它一开始可不懂视频里到底是啥意思，它看到的就是一堆像素点和声音信号。 视频理解呢，就是要让计…
阅读更多...
《论软件的可靠性评价》审题技巧 - 系统架构设计师

《论软件的可靠性评价》审题技巧 - 系统架构设计师

论软件的可靠性评价写作框架 一、考点概述 软件可靠性评价作为软件可靠性活动的关键环节&#xff0c;是确保软件质量、提升用户体验的重要手段。本题主要考察以下几个方面的内容&#xff1a; 首先&#xff0c;本题要求考生理解并掌握软件可靠性评价的基本概念及其在软件开发…
阅读更多...
AI自动化爬虫项目对比报告

AI自动化爬虫项目对比报告

摘要 本报告旨在深入研究AI自动化爬虫项目，对比分析其在实现方式、效率提升、自托管能力等方面的差异。 随着大数据和人工智能技术的快速发展，传统网络爬虫技术面临着越来越多的挑战，如网站反爬虫机制的加强、网页结构复杂多变等。AI自动化爬虫技术应运而生，利用机器学习、…
阅读更多...
AI 与光学的融合：开启科技变革新征程

AI 与光学的融合：开启科技变革新征程

在当今科技发展的浪潮中&#xff0c;人工智能&#xff08;AI&#xff09;与光学这两个看似独立的领域&#xff0c;正逐渐紧密交织&#xff0c;开启了一场影响深远的科技变革。AI 技术凭借强大的数据分析、模式识别和智能决策能力&#xff0c;与光学在信息传输、成像、传感等方面…
阅读更多...
简单说一下什么是RPC

简单说一下什么是RPC

部分内容来源&#xff1a;JavaGuide RPC是什么 RPC是远程调用 RPC的原理 RPC的五个部分 为了能够帮助小伙伴们理解 RPC 原理&#xff0c;我们可以将整个 RPC 的核心功能看作是下面 5 个部分实现的&#xff1a; 客户端&#xff08;服务消费端&#xff09;&#xff1a;调用…
阅读更多...
最新文章

Copyright @ 2022~2023