使用nanomsg实现多子节点时序or数据同步

2024-05-03 18:38
文章标签 数据 实现 使用 同步 节点 时序 多子 nanomsg

本文主要是介绍使用nanomsg实现多子节点时序or数据同步，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

前言

本文通过简绍nanomsg基本通信模式结构，并采用PIPELINE、PUBSUB模式相互配合的方式来实现多子节点时序or数据同步；设计中使用PIPELINE模式用来采集子节点的同步时序或同步数据，如果子节点都同步之后使用PUBSUB模式群发订阅消息发送事件来通知子节点。

设计结构图

  • PIPELINE模式

PIPELINE模式

 聚合来自多个来源的消息，并在多个目的地之间进行负载平衡。node0只能send，node1只能recv。

  • PUBSUB 模式
PUBSUB 模式

 发布者向订阅者推送消息。只有订阅该频道的订阅者才能收到该频道的消息。

  • 多子节点时序or数据同步设计图

实现代码

/*% ./test_demo tcp://127.0.0.1:5555 -s &% ./test_demo tcp://127.0.0.1:5555 Garrett% ./test_demo tcp://127.0.0.1:5555 Silent_DXX
*/#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <time.h>#include <cstdlib>
#include <cstdio>
#include <cstring>
#include <iostream>
#include <thread>#include <nanomsg/nn.h>
#include <nanomsg/pipeline.h>
#include <nanomsg/pubsub.h>void fatal(const char *func) {std::cout << func << ":" << nn_strerror(nn_errno()) << std::endl;exit(1);
}/*  The server runs forever. */
int server(const char *url)
{int fd[2];int rc;/*  Create the socket 0. */fd[0] = nn_socket (AF_SP, NN_PULL);if (fd[0] < 0) {fprintf (stderr, "nn_socket: %s\n", nn_strerror (nn_errno ()));return (-1);}if (nn_bind (fd[0], url) < 0) {fprintf (stderr, "nn_bind: %s\n", nn_strerror (nn_errno ()));nn_close (fd[0]);return (-1);}/*  Create the socket 1. */fd[1] = nn_socket (AF_SP, NN_PUB);if (fd[1] < 0) {fprintf (stderr, "nn_socket: %s\n", nn_strerror (nn_errno ()));return (-1);}char pub_url[128];memcpy(pub_url, url, strlen(url) + 1);pub_url[strlen(pub_url) - 1] += 1;if (nn_bind (fd[1], pub_url) < 0) {fprintf (stderr, "nn_bind: %s\n", nn_strerror (nn_errno ()));nn_close (fd[1]);return (-1);}printf("Wait for clinet conneted...");getchar();static int sync_flag = 0;for (;;) {uint32_t* buf = NULL;int bytes;if ((bytes = nn_recv(fd[0], &buf, NN_MSG, 0)) < 0) {fatal("nn_recv");}printf("NODE0: RECEIVED \"%d\"\n", *(uint32_t*)buf);nn_freemsg(buf);sync_flag++;uint32_t subs;uint32_t* msg;uint32_t rev_flag = 0x55aa;msg = &rev_flag;subs = (uint32_t) nn_get_statistic (fd[1], NN_STAT_CURRENT_CONNECTIONS);if (sync_flag >= subs) {sync_flag = 0;printf("Send event, subs: %d\n", subs);rc = nn_send (fd[1], msg, sizeof (uint32_t), 0);if (rc < 0) {fprintf (stderr, "nn_send: %s (ignoring)\n", nn_strerror (nn_errno ()));}}}nn_close (fd[0]);return (-1);
}/*  The client runs just once, and then returns. */
int client (const char *url, const char *username)
{int fd[2];int rc;int bytes;uint32_t *msg;uint32_t time = 0;msg = &time;/*  Create the socket 0. */if ((fd[0] = nn_socket(AF_SP, NN_PUSH)) < 0) {fatal("nn_socket");}if ((rc = nn_connect(fd[0], url)) < 0) {fatal("nn_connect");}/*  Create the socket 1. */fd[1] = nn_socket (AF_SP, NN_SUB);if (fd[1] < 0) {fprintf (stderr, "nn_socket: %s\n", nn_strerror (nn_errno ()));return (-1);}char pub_url[128];memcpy(pub_url, url, strlen(url) + 1);pub_url[strlen(pub_url) - 1] += 1;if (nn_connect (fd[1], pub_url) < 0) {fprintf (stderr, "nn_socket: %s\n", nn_strerror (nn_errno ()));nn_close (fd[1]);return (-1);}/*  We want all messages, so just subscribe to the empty value. */if (nn_setsockopt (fd[1], NN_SUB, NN_SUB_SUBSCRIBE, "", 0) < 0) {fprintf (stderr, "nn_setsockopt: %s\n", nn_strerror (nn_errno ()));nn_close (fd[1]);return (-1);}for (;;) {printf("%s : SENDING \"%d\"\n", __func__, time);if ((bytes = nn_send(fd[0], msg, sizeof(uint32_t), 0)) < 0) {fatal("nn_send");}time++;uint8_t msg[2 * sizeof (uint32_t)];rc = nn_recv (fd[1], msg, sizeof (msg), 0);printf("Event: 0x%x\n", *(uint32_t*)&msg);if (rc < 0) {fprintf (stderr, "nn_recv: %s\n", nn_strerror (nn_errno ()));break;}sleep(2);}return 0;
}int main (int argc, char **argv)
{int rc;if (argc < 3) {fprintf (stderr, "Usage: %s <url> [-s|name]\n", argv[0]);exit (EXIT_FAILURE);}if (strcmp (argv[2], "-s") == 0) {rc = server (argv[1]);} else {rc = client (argv[1], argv[2]);}exit (rc == 0 ? EXIT_SUCCESS : EXIT_FAILURE);
}

测试

  • 启动服务端
./test_demo.exe tcp://127.0.0.1:5555 -s
  • 启动一个节点
./test_demo.exe tcp://127.0.0.1:5555 Garrett1
  • 启动第二个节点
./test_demo.exe tcp://127.0.0.1:5555 Garrett2
  • 测试输出

服务器:

$ ./test_demo.exe tcp://127.0.0.1:5555 -s
Wait for clinet conneted...
NODE0: RECEIVED "0"
NODE0: RECEIVED "0"
Send event, subs: 2
NODE0: RECEIVED "1"
NODE0: RECEIVED "1"
Send event, subs: 2
NODE0: RECEIVED "2"
NODE0: RECEIVED "2"
Send event, subs: 2
NODE0: RECEIVED "3"
NODE0: RECEIVED "3"
Send event, subs: 2
NODE0: RECEIVED "4"
NODE0: RECEIVED "4"
Send event, subs: 2
NODE0: RECEIVED "5"
NODE0: RECEIVED "5"
Send event, subs: 2
NODE0: RECEIVED "6"
NODE0: RECEIVED "6"
Send event, subs: 2
NODE0: RECEIVED "7"
NODE0: RECEIVED "7"
Send event, subs: 2

节点: 

$ ./test_demo.exe tcp://127.0.0.1:5555 Garrett1
client : SENDING "0"
Event: 0x55aa
client : SENDING "1"
Event: 0x55aa
client : SENDING "2"
Event: 0x55aa
client : SENDING "3"
Event: 0x55aa
client : SENDING "4"
Event: 0x55aa
client : SENDING "5"
Event: 0x55aa
client : SENDING "6"
Event: 0x55aa
client : SENDING "7"
Event: 0x55aa

 

这篇关于使用nanomsg实现多子节点时序or数据同步的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/957384。

相关文章

使用Python和OpenCV库实现实时颜色识别系统

使用Python和OpenCV库实现实时颜色识别系统

《使用Python和OpenCV库实现实时颜色识别系统》：本文主要介绍使用Python和OpenCV库实现的实时颜色识别系统,这个系统能够通过摄像头捕捉视频流,并在视频中指定区域内识别主要颜色（红... 目录一、引言二、系统概述三、代码解析1. 导入库2. 颜色识别函数3. 主程序循环四、HSV色彩空间详解
阅读更多...
Windows下C++使用SQLitede的操作过程

Windows下C++使用SQLitede的操作过程

《Windows下C++使用SQLitede的操作过程》本文介绍了Windows下C++使用SQLite的安装配置、CppSQLite库封装优势、核心功能（如数据库连接、事务管理）、跨平台支持及性能优... 目录Windows下C++使用SQLite1、安装2、代码示例CppSQLite：C++轻松操作SQ
阅读更多...
PostgreSQL中MVCC 机制的实现

PostgreSQL中MVCC 机制的实现

《PostgreSQL中MVCC机制的实现》本文主要介绍了PostgreSQL中MVCC机制的实现,通过多版本数据存储、快照隔离和事务ID管理实现高并发读写,具有一定的参考价值,感兴趣的可以了解一下... 目录一 MVCC 基本原理python1.1 MVCC 核心概念1.2 与传统锁机制对比二 Postg
阅读更多...
SpringBoot整合Flowable实现工作流的详细流程

SpringBoot整合Flowable实现工作流的详细流程

《SpringBoot整合Flowable实现工作流的详细流程》Flowable是一个使用Java编写的轻量级业务流程引擎,Flowable流程引擎可用于部署BPMN2.0流程定义,创建这些流程定义的... 目录1、流程引擎介绍2、创建项目3、画流程图4、开发接口4.1 Java 类梳理4.2 查看流程图4
阅读更多...
SQL Server修改数据库名及物理数据文件名操作步骤

SQL Server修改数据库名及物理数据文件名操作步骤

《SQLServer修改数据库名及物理数据文件名操作步骤》在SQLServer中重命名数据库是一个常见的操作,但需要确保用户具有足够的权限来执行此操作,：本文主要介绍SQLServer修改数据... 目录一、背景介绍二、操作步骤2.1 设置为单用户模式（断开连接）2.2 修改数据库名称2.3 查找逻辑文件名
阅读更多...
C++中零拷贝的多种实现方式

C++中零拷贝的多种实现方式

《C++中零拷贝的多种实现方式》本文主要介绍了C++中零拷贝的实现示例,旨在在减少数据在内存中的不必要复制,从而提高程序性能、降低内存使用并减少CPU消耗,零拷贝技术通过多种方式实现,下面就来了解一下... 目录一、C++中零拷贝技术的核心概念二、std::string_view 简介三、std::stri
阅读更多...
Python常用命令提示符使用方法详解

Python常用命令提示符使用方法详解

《Python常用命令提示符使用方法详解》在学习python的过程中,我们需要用到命令提示符(CMD)进行环境的配置,：本文主要介绍Python常用命令提示符使用方法的相关资料,文中通过代码介绍的... 目录一、python环境基础命令【Windows】1、检查Python是否安装2、 查看Python的安
阅读更多...
C++高效内存池实现减少动态分配开销的解决方案

C++高效内存池实现减少动态分配开销的解决方案

《C++高效内存池实现减少动态分配开销的解决方案》C++动态内存分配存在系统调用开销、碎片化和锁竞争等性能问题,内存池通过预分配、分块管理和缓存复用解决这些问题,下面就来了解一下... 目录一、C++内存分配的性能挑战二、内存池技术的核心原理三、主流内存池实现：TCMalloc与Jemalloc1. TCM
阅读更多...
OpenCV实现实时颜色检测的示例

OpenCV实现实时颜色检测的示例

《OpenCV实现实时颜色检测的示例》本文主要介绍了OpenCV实现实时颜色检测的示例,通过HSV色彩空间转换和色调范围判断实现红黄绿蓝颜色检测,包含视频捕捉、区域标记、颜色分析等功能,具有一定的参考... 目录一、引言二、系统概述三、代码解析1. 导入库2. 颜色识别函数3. 主程序循环四、HSV色彩空间
阅读更多...
Python并行处理实战之如何使用ProcessPoolExecutor加速计算

Python并行处理实战之如何使用ProcessPoolExecutor加速计算

《Python并行处理实战之如何使用ProcessPoolExecutor加速计算》Python提供了多种并行处理的方式,其中concurrent.futures模块的ProcessPoolExecu... 目录简介完整代码示例代码解释1. 导入必要的模块2. 定义处理函数3. 主函数4. 生成数字列表5.
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2