当前位置： 首页 > article >正文

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

article 2025/2/23 5:49:35

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_ZMQ
    LDFLAGS += -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.kler.cn/a/557097.html

相关文章：

  • HarmonyOS 开发套件 介绍——下篇
  • std::lock_guard、std::unique_lock、std::shared_lock
  • 青少年软件编程（C语言）等级三级考试试题（2）
  • DeepSeek 到底是什么类型的应用，其核心功能是什么？
  • 工业机器人中用于3D碰撞检测的算法库有哪些
  • ubuntu ffmpeg 安装踩坑
  • 【Python项目】基于Django的网站验证码的生成与识别系统
  • 基于Java+SpringBoot+Vue的前后端分离的汽车租赁系统
  • 正则表达式效验邮箱格式, 手机号格式, 密码长度
  • Python 学习之旅：高级阶段（十四）Web 开发框架 Flask
  • 邮件安全之发件人伪造
  • [漏洞篇]文件上传漏洞详解
  • matlab 车辆进出检测算法设计GUI界面-论文
  • 蓝桥杯刷题2.21|笔记
  • 爬虫学习第八篇-学习小总结
  • MYSQL查询优化器选错了索引怎么办
  • matlab 轮边驱动系统汽车垂向动力学分析
  • python学opencv|读取图像（七十四）人脸识别：EigenFaces算法
  • vue中将el-table导出为excel文件
  • 使用LangChain构建第一个ReAct Agent