使用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。

相关文章

Java实现字节字符转bcd编码

Java实现字节字符转bcd编码

《Java实现字节字符转bcd编码》BCD是一种将十进制数字编码为二进制的表示方式,常用于数字显示和存储,本文将介绍如何在Java中实现字节字符转BCD码的过程,需要的小伙伴可以了解下... 目录前言BCD码是什么Java实现字节转bcd编码方法补充总结前言BCD码（Binary-Coded Decima
阅读更多...
SpringBoot全局域名替换的实现

SpringBoot全局域名替换的实现

《SpringBoot全局域名替换的实现》本文主要介绍了SpringBoot全局域名替换的实现,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一... 目录 项目结构⚙️ 配置文件application.yml️ 配置类AppProperties.Ja
阅读更多...
Java使用Javassist动态生成HelloWorld类

Java使用Javassist动态生成HelloWorld类

《Java使用Javassist动态生成HelloWorld类》Javassist是一个非常强大的字节码操作和定义库,它允许开发者在运行时创建新的类或者修改现有的类,本文将简单介绍如何使用Javass... 目录1. Javassist简介2. 环境准备3. 动态生成HelloWorld类3.1 创建CtC
阅读更多...
使用Python批量将.ncm格式的音频文件转换为.mp3格式的实战详解

使用Python批量将.ncm格式的音频文件转换为.mp3格式的实战详解

《使用Python批量将.ncm格式的音频文件转换为.mp3格式的实战详解》本文详细介绍了如何使用Python通过ncmdump工具批量将.ncm音频转换为.mp3的步骤,包括安装、配置ffmpeg环... 目录1. 前言2. 安装 ncmdump3. 实现 .ncm 转 .mp34. 执行过程5. 执行结
阅读更多...
Python实现批量CSV转Excel的高性能处理方案

Python实现批量CSV转Excel的高性能处理方案

《Python实现批量CSV转Excel的高性能处理方案》在日常办公中,我们经常需要将CSV格式的数据转换为Excel文件,本文将介绍一个基于Python的高性能解决方案,感兴趣的小伙伴可以跟随小编一... 目录一、场景需求二、技术方案三、核心代码四、批量处理方案五、性能优化六、使用示例完整代码七、小结一、
阅读更多...
Java实现将HTML文件与字符串转换为图片

Java实现将HTML文件与字符串转换为图片

《Java实现将HTML文件与字符串转换为图片》在Java开发中,我们经常会遇到将HTML内容转换为图片的需求,本文小编就来和大家详细讲讲如何使用FreeSpire.DocforJava库来实现这一功... 目录前言核心实现：html 转图片完整代码场景 1：转换本地 HTML 文件为图片场景 2：转换 H
阅读更多...
Java使用jar命令配置服务器端口的完整指南

Java使用jar命令配置服务器端口的完整指南

《Java使用jar命令配置服务器端口的完整指南》本文将详细介绍如何使用java-jar命令启动应用,并重点讲解如何配置服务器端口,同时提供一个实用的Web工具来简化这一过程,希望对大家有所帮助... 目录1. Java Jar文件简介1.1 什么是Jar文件1.2 创建可执行Jar文件2. 使用java
阅读更多...
C#使用Spire.Doc for .NET实现HTML转Word的高效方案

C#使用Spire.Doc for .NET实现HTML转Word的高效方案

《C#使用Spire.Docfor.NET实现HTML转Word的高效方案》在Web开发中,HTML内容的生成与处理是高频需求,然而,当用户需要将HTML页面或动态生成的HTML字符串转换为Wor... 目录引言一、html转Word的典型场景与挑战二、用 Spire.Doc 实现 HTML 转 Word1
阅读更多...
C#实现一键批量合并PDF文档

C#实现一键批量合并PDF文档

《C#实现一键批量合并PDF文档》这篇文章主要为大家详细介绍了如何使用C#实现一键批量合并PDF文档功能,文中的示例代码简洁易懂,感兴趣的小伙伴可以跟随小编一起学习一下... 目录前言效果展示功能实现1、添加文件2、文件分组（书签）3、定义页码范围4、自定义显示5、定义页面尺寸6、PDF批量合并7、其他方法
阅读更多...
SpringBoot实现不同接口指定上传文件大小的具体步骤

SpringBoot实现不同接口指定上传文件大小的具体步骤

《SpringBoot实现不同接口指定上传文件大小的具体步骤》：本文主要介绍在SpringBoot中通过自定义注解、AOP拦截和配置文件实现不同接口上传文件大小限制的方法,强调需设置全局阈值远大于... 目录一  springboot实现不同接口指定文件大小1.1 思路说明1.2 工程启动说明二 具体实施2
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2