Linux nanomsg tcp sub pub

一、安装

下载：Releases · nanomsg/nanomsg · GitHub
安装：https://github.com/nanomsg/nanomsg

tar xvf nanomsg-1.1.4.tar.gz
cd nanomsg-1.1.4
mkdir build
cd build
cmake .. -DCMAKE_INSTALL_PREFIX=/usr/local
cmake --build .
ctest .
sudo cmake --build . --target install
sudo ldconfig

二、简单sub pub

程序一

//进程一 文件pubsub.c
#include <assert.h>
#include <nanomsg/nn.h>
#include <nanomsg/pubsub.h>
#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/time.h>
#include <unistd.h>#define SOCKET_ADDRESS_A "inproc://A" //线程A的消息队列的地址
#define SOCKET_ADDRESS_B "inproc://B" //线程B的消息队列的地址
#define SOCKET_ADDRESS_IPC_A "ipc:///tmp/a.ipc" //用于进程间通讯的消息队列的地址
#define SOCKET_ADDRESS_C "inproc://c" //线程C的消息队列的地址，这里没有用到//线程A用于发布消息的线程
void* A_pub_thread(void* arg){time_t now; //获取本地时间（描数）int flag = 0; struct tm* p;char msg[100]; //发送消息的长度int rc = 0;    //int ret = 0;int sock_pub = nn_socket(AF_SP, NN_PUB); //创建一个套接字模式为发布套接字assert(sock_pub >= 0); //确保创建成功//绑定本地地址assert(nn_bind(sock_pub, SOCKET_ADDRESS_A) >= 0);//绑定本地消息队列的地址assert(nn_bind(sock_pub, SOCKET_ADDRESS_IPC_A) >= 0);//绑定用于进程间消息接收的消息队列的地址//pub使用来发送的while (1) {now = time(NULL); //获取本地时间rc = 0;ret = 0;//如果满足 10 的发送条件if ((now % 10 == 0) && (flag == 0)) {flag = 1;p = gmtime(&now);memset(&msg, 0, sizeof(msg)); //清空发送缓存sprintf(msg, "10|%04d-%02d-%02d %02d:%02d:%02d from A", 1900 + p->tm_year, 1 + p->tm_mon, p->tm_mday, p->tm_hour, p->tm_min, p->tm_sec);//将发送字符串写入缓存，注意格式必须是 类型|内容 这里的类型就是10 内容就是当前时间 +from Aret = nn_send(sock_pub, msg, sizeof(msg), NN_DONTWAIT);//发送消息//判断消息是否发送成功if (ret < 0) {printf("A pub send msg failed\n");}}//如果满足 05 的发送条件if ((now % 5 == 0) && (now % 10 != 0) && (flag == 1)) {//和上面一样，注释就不写了flag = 0;p = gmtime(&now);memset(&msg, 0, sizeof(msg));sprintf(msg, "05|%04d-%02d-%02d %02d:%02d:%02d from A", 1900 + p->tm_year, 1 + p->tm_mon, p->tm_mday, p->tm_hour, p->tm_min, p->tm_sec);ret = nn_send(sock_pub, msg, sizeof(msg), NN_DONTWAIT);if (ret < 0) {printf("A pub send msg failed\n");}}//一秒循环一次sleep(1);}
}//线程A用于订阅消息的线程
void* A_sub_thread(void* arg){time_t now; //用于获取当前时间int flag = 0;struct tm* p; //时间结构体char rcvmsg[100]; //用于接收消息int rc = 0; //判断接收长度int sock_sub = nn_socket(AF_SP, NN_SUB); //创建用于接收消息的套接字assert(sock_sub >= 0);//确保创建套接字成功//绑定本地地址assert(nn_connect(sock_sub, SOCKET_ADDRESS_A) >= 0);//连接要接收消息的地址assert(nn_connect(sock_sub, SOCKET_ADDRESS_B) >= 0);//可以同时连接多个//sub是用来订阅的nn_setsockopt(sock_sub, NN_SUB, NN_SUB_SUBSCRIBE, "05", 2);//订阅 05 的消息（包括两上面连接的容器的消息）nn_setsockopt(sock_sub, NN_SUB, NN_SUB_SUBSCRIBE, "10", 2);//订阅 10 的消息（包括两上面连接的容器的消息）nn_setsockopt(sock_sub, NN_SUB, NN_SUB_SUBSCRIBE, "03", 2);//订阅 03 的消息（包括两上面连接的容器的消息）//pub使用来发送的while (1) {// 清空缓存memset(&rcvmsg, 0, sizeof(rcvmsg));//接收消息 第四个参数为0时表示租售接收rc = nn_recv(sock_sub, &rcvmsg, sizeof(rcvmsg), 0);if (rc > 0) {printf("A sub rcv: %s\n", rcvmsg);rc = 0;memset(&rcvmsg, 0, sizeof(rcvmsg));rc = nn_recv(sock_sub, &rcvmsg, NN_MSG, NN_DONTWAIT);}}
}//线程A
void* A_thread(void* arg){pthread_t thread_A_1, thread_A_2;pthread_create(&thread_A_1, NULL, A_pub_thread, NULL);pthread_create(&thread_A_2, NULL, A_sub_thread, NULL);pthread_join(thread_A_1, NULL);pthread_join(thread_A_2, NULL);
}void* B_pub_thread(void* arg){time_t now;int flag = 0;struct tm* p;char msg[100];int rc = 0;int ret = 0;int sock_pub = nn_socket(AF_SP, NN_PUB);assert(sock_pub >= 0);//绑定本地地址assert(nn_bind(sock_pub, SOCKET_ADDRESS_B) >= 0);//pub使用来发送的while (1) {now = time(NULL);rc = 0;ret = 0;if ((now % 3 == 0) && (now % 2 != 0) && (flag == 0)) {flag = 1;p = gmtime(&now);memset(&msg, 0, sizeof(msg));sprintf(msg, "03|%04d-%02d-%02d %02d:%02d:%02d from B", 1900 + p->tm_year, 1 + p->tm_mon, p->tm_mday, p->tm_hour, p->tm_min, p->tm_sec);ret = nn_send(sock_pub, msg, sizeof(msg), NN_DONTWAIT);if (ret < 0) {printf("A pub send msg failed\n");}}if ((now % 2 == 0) && (flag == 1)){flag = 0;}sleep(1);}
}void* B_sub_thread(void* arg){time_t now;int flag = 0;struct tm* p;char msg[100];char rcvmsg[100];int rc = 0;int ret = 0;int sock_sub = nn_socket(AF_SP, NN_SUB);assert(sock_sub >= 0);//绑定本地地址assert(nn_connect(sock_sub, SOCKET_ADDRESS_A) >= 0);assert(nn_connect(sock_sub, SOCKET_ADDRESS_B) >= 0);//sub是用来订阅的nn_setsockopt(sock_sub, NN_SUB, NN_SUB_SUBSCRIBE, "03", 2);while (1) {memset(&rcvmsg, 0, sizeof(rcvmsg));rc = nn_recv(sock_sub, &rcvmsg, sizeof(rcvmsg), 0);if (rc > 0) {printf("B sub rcv: %s\n", rcvmsg);rc = 0;memset(&rcvmsg, 0, sizeof(rcvmsg));rc = nn_recv(sock_sub, &rcvmsg, NN_MSG, NN_DONTWAIT);}}
}void* B_thread(void* arg){pthread_t thread_B_1, thread_B_2;pthread_create(&thread_B_1, NULL, B_pub_thread, NULL);pthread_create(&thread_B_2, NULL, B_sub_thread, NULL);pthread_join(thread_B_1, NULL);pthread_join(thread_B_2, NULL);
}void* C_thread(void* arg){int flag = 0;struct tm* p;char rcvmsg[100];int rc = 0;int ret = 0;int sock_sub = nn_socket(AF_SP, NN_SUB);assert(sock_sub >= 0);//绑定本地地址assert(nn_connect(sock_sub, SOCKET_ADDRESS_A) >= 0);nn_setsockopt(sock_sub, NN_SUB, NN_SUB_SUBSCRIBE, "10", 2);while (1) {memset(&rcvmsg, 0, sizeof(rcvmsg));rc = nn_recv(sock_sub, &rcvmsg, sizeof(rcvmsg), 0);while (rc > 0) {printf("C sub rcv: %s\n", rcvmsg);rc = 0;memset(&rcvmsg, 0, sizeof(rcvmsg));rc = nn_recv(sock_sub, &rcvmsg, sizeof(rcvmsg), NN_DONTWAIT);}}
}int main(int argc, char** argv)
{pthread_t thread_A, thread_B, thread_C;pthread_create(&thread_A, NULL, A_thread, NULL);pthread_create(&thread_B, NULL, B_thread, NULL);pthread_create(&thread_C, NULL, C_thread, NULL);pthread_join(thread_A, NULL);pthread_join(thread_B, NULL);pthread_join(thread_C, NULL);exit(EXIT_SUCCESS);
}

程序二

//进程二
#include <stdlib.h>
#include <string.h>
#include <sys/time.h>
#include <assert.h>
#include <unistd.h>#define SOCKET_ADDRESS_IPC_A "ipc:///tmp/a.ipc"int main(){time_t now;int flag = 0;struct tm* p;char rcvmsg[100];int rc = 0;int sock_sub = nn_socket(AF_SP, NN_SUB);assert(sock_sub >= 0);//绑定本地地址assert(nn_connect(sock_sub, SOCKET_ADDRESS_IPC_DASS) >= 0);//sub是用来订阅和接收的nn_setsockopt(sock_sub, NN_SUB, NN_SUB_SUBSCRIBE, "10", 2);while (1) {memset(&rcvmsg, 0, sizeof(rcvmsg));rc = nn_recv(sock_sub, &rcvmsg, sizeof(rcvmsg), 0);while (rc > 0) {printf("2 sub rcv: %s\n", rcvmsg);rc = 0;memset(&rcvmsg, 0, sizeof(rcvmsg));rc = nn_recv(sock_sub, &rcvmsg, sizeof(rcvmsg), NN_DONTWAIT);}}return 0;
}

三、封装sub pub

nanomsg实现进程间通讯，将nanomsg进行封装

#ifndef __NANO_ROS_H
#define __NANO_ROS_H
#include <string>
#include <cstring>
#include <assert.h>
#include <nanomsg/nn.h>
#include <nanomsg/pubsub.h>
#include <stdio.h>
#include <stdlib.h>
#include <sys/time.h>
#include <unistd.h>
#include <thread> //线程头文件
#include <functional>
#include <vector>
#include <stdint.h>
using namespace std;
const int data_size=3000000;namespace nano
{using MyFunction = std::function<void(char*)>;  std::vector <MyFunction> numericalFunctions;auto& fct_save = numericalFunctions;typedef void (*pf_char)(char*);void B_sub_thread(void);int rc = 0;int ret = 0;vector <int> sock_pub;vector <int> sock_sub;int all_pub=0;int all_sub=0;std::string address_node;std::string node_name;vector <std::string> sub_topic;vector <std::string> pub_topic;vector <bool> msg_update;char Rcvmsg[10][data_size];vector <int> msg_size;std::thread t_ros{B_sub_thread};bool thread_bit=false;int sleep_time=1;void init(std::string node){}void init(int a,char **argv,std::string node){}bool ok(void){return true;}void spinOnce(){for(int i=0;i<msg_update.size();i++){if(msg_update[i])fct_save[i](Rcvmsg[i]);//将接收的数据返回到回调函数msg_update[i]=false;}}class Publisher{public:int pub_num;Publisher(){pub_num=all_pub;}template <typename T>void publish(const T &msg)//{ret = 0;     ret = nn_send(sock_pub[pub_num],&msg,msg_size[pub_num], NN_DONTWAIT);if (ret < 0) {printf("A pub send msg failed\n");}      };template <typename T>void publish(const T &msg,int num){ret = 0;     ret = nn_send(sock_pub[pub_num],&msg,num, NN_DONTWAIT);if (ret < 0) {printf("A pub send msg failed\n");}}};class Subscriber{public:int sub_num;Subscriber(){sub_num=all_sub;}};class NodeHandle{public:template <typename T>Publisher  advertise(std::string pub_topic_name,int num){msg_size.push_back(sizeof(T)/sizeof(char));Publisher PubNum;sock_pub.push_back(nn_socket(AF_SP, NN_PUB));assert( sock_pub.back()>= 0);           //绑定本地地址std::string  PubAddress;char ip_check[150];strcpy(ip_check,pub_topic_name.c_str());if(ip_check[0]>'0'&&ip_check[0]<'9'){printf("This pub is tcp,device between\n");PubAddress="tcp://"+pub_topic_name;                  }else{printf("This pub is ipc\n");PubAddress="ipc:///tmp/"+pub_topic_name+".ipc";}pub_topic.push_back(PubAddress);const char *SOCKET_ADDRESS_IPC_B=PubAddress.c_str();assert(nn_bind(sock_pub.back(), SOCKET_ADDRESS_IPC_B) >= 0);//绑定用于进程间消息接收的消息队列的地址all_pub++;printf("all_pub is %d\n",all_pub);return PubNum;}Subscriber subscribe(std::string sub_topic_name, int num,const pf_char &sub_fun){//将回调函数作为形参加载到线程中运行Subscriber SubNum;msg_update.push_back(false);sub_topic.push_back(sub_topic_name);sock_sub.push_back(nn_socket(AF_SP, NN_SUB));assert(sock_sub.back() >= 0);std::string  SubAddress;char ip_check[150];strcpy(ip_check,sub_topic_name.c_str());if(ip_check[0]>'0'&&ip_check[0]<'9'){printf("This sub is tcp,device between\n");SubAddress="tcp://"+sub_topic.back();                  }else{printf("This sub is ipc\n");SubAddress="ipc:///tmp/"+sub_topic.back()+".ipc";}const char *SOCKET_ADDRESS_IPC_A=SubAddress.c_str();//绑定本地地址assert(nn_connect(sock_sub.back(), SOCKET_ADDRESS_IPC_A) >= 0);//sub是用来订阅的nn_setsockopt(sock_sub.back(), NN_SUB, NN_SUB_SUBSCRIBE, "", 0);fct_save.push_back(sub_fun);if(!thread_bit){t_ros.detach();thread_bit=true;}all_sub++;     return SubNum;};int node_num;};class Rate{public:Rate(int time){sleep_time=time;}void sleep(void){usleep(1000*1000/sleep_time);};};void B_sub_thread(void){char rcvmsg[data_size];while (1) {       memset(&rcvmsg, 0, sizeof(rcvmsg));for(int bbb=0;bbb<sock_sub.size();bbb++){rc = nn_recv(sock_sub[bbb], &rcvmsg, sizeof(rcvmsg),NN_DONTWAIT);if (rc > 0) {rc = 0;msg_update[bbb]=true;memcpy(&Rcvmsg[bbb],&rcvmsg[0], sizeof(rcvmsg));}}usleep(1000*1000/sleep_time);}}
}
void NANO_INFO_STREAM(std::string msg)
{timeval now;char msg_s[100];gettimeofday(&now, NULL);memset(&msg_s, 0, sizeof(msg_s));sprintf(msg_s, "%ld.%06ld:%s", now.tv_sec, now.tv_usec,msg.c_str());printf("%s\n",msg_s);
}
void NANO_INFO_STREAM(std::string msg_a,float msg)
{timeval now;char msg_s[100];gettimeofday(&now, NULL);memset(&msg_s, 0, sizeof(msg_s));sprintf(msg_s, "%ld.%06ld:%s %f", now.tv_sec, now.tv_usec,msg_a.c_str(),msg);printf("%s\n",msg_s);
}
void NANO_INFO(std::string msg_a,float msg)
{timeval now;char msg_s[100];gettimeofday(&now, NULL);memset(&msg_s, 0, sizeof(msg_s));sprintf(msg_s, "%ld.%06ld:%s %f", now.tv_sec, now.tv_usec,msg_a.c_str(),msg);printf("%s\n",msg_s);
}
#endif

以camera为例，封装消息类型

#ifndef __SENSOR_MSGS_CAM_H
#define __SENSOR_MSGS_CAM_H
#include <sys/time.h>
#include <stdint.h>
#include <string>
using namespace std;
const int CAMERA_SIZE=921656;struct Header_cam{string frame_id;uint32_t seq; timeval stamp;   // 时间戳
};struct camera_data
{unsigned char data[921600];
};
struct sensor_msgs_cam{Header_cam header;camera_data Data;
};
#endif

订阅发布函数

#include "m_nano.hpp"
#include <string>
#include "sensor_msgs_cam.hpp"
#include "readconfig.hpp"
#include "package_path.h"
#include  <opencv2/core/core.hpp>
#include  <opencv2/highgui/highgui.hpp>
#include  <opencv2/imgcodecs.hpp>
using namespace std;
sensor_msgs_cam CamData;
int main(int argc, char **argv)
{nano::init(argc, argv, "node_b");    //初始化ROS，节点命名为node_b，节点名必须唯一。nano::NodeHandle nh;    //节点句柄实例化nano::Publisher pub_camera=nh.advertise<sensor_msgs_cam>("100.100.100.99:5555",10);nano::Publisher pub_camera_t=nh.advertise<sensor_msgs_cam>("camera",10);nano::Rate loop_rate(10);    //设置发送数据的频率为10Hzstd::string yaml_path=PROJECT_PATH;yaml_path+="/config/cam.yaml";int num_cam = readConfigInt(yaml_path,"cam_num",0);int show_video = readConfigInt(yaml_path,"show_video",0);static cv::VideoCapture cap;if(cap.open(num_cam) == 0){perror("Open dev error!");}cv::Mat frame(480, 640, CV_8UC3, cv::Scalar(255,0,0));timeval now;while(nano::ok()){   cap.read(frame);  memcpy(CamData.Data.data,frame.data,sizeof(camera_data)/sizeof(char));gettimeofday(&now, NULL);CamData.header.stamp=now;if(show_video){imshow("video show",frame);cv::waitKey(1);}pub_camera.publish(CamData);pub_camera_t.publish(CamData);nano::spinOnce();      //不是必须，若程序中订阅话题则必须，否则回掉函数不起作用。loop_rate.sleep();    //按前面设置的10Hz频率将程序挂起}return 0;
}

订阅发布函数2

#include "m_nano.hpp"
#include <string>
#include "sensor_msgs_cam.hpp"
#include "readconfig.hpp"
#include "package_path.h"
#include  <opencv2/core/core.hpp>
#include  <opencv2/highgui/highgui.hpp>
#include  <opencv2/imgcodecs.hpp>
using namespace std;
int show_video=0;
sensor_msgs_cam CamData,CamData_t;
cv::Mat dst_mat(480, 640, CV_8UC3, cv::Scalar(255,0,0));
cv::Mat dst_mat_t(480, 640, CV_8UC3, cv::Scalar(255,0,0));void camera_callback(char *msg)
{memcpy(&CamData,&msg[0],CAMERA_SIZE);memcpy(dst_mat.data,CamData.Data.data,sizeof(camera_data)/sizeof(char));if(show_video){imshow("video recive",dst_mat);cv::waitKey(1);}
}void camera_callback_t(char *msg)
{memcpy(&CamData_t,&msg[0],CAMERA_SIZE);memcpy(dst_mat_t.data,CamData_t.Data.data,sizeof(camera_data)/sizeof(char));if(show_video){imshow("video recive",dst_mat_t);cv::waitKey(1);}
}int main(int argc, char **argv)
{nano::init(argc, argv, "node_b");    //初始化ROS，节点命名为node_b，节点名必须唯一。nano::NodeHandle nh;    //节点句柄实例化nano::Subscriber sub = nh.subscribe("camera",50,camera_callback);nano::Subscriber sub_t = nh.subscribe("100.100.100.99:5555",50,camera_callback_t);nano::Rate loop_rate(10);    //设置发送数据的频率为10Hzstd::string yaml_path=PROJECT_PATH;yaml_path+="/config/cam.yaml";show_video = readConfigInt(yaml_path,"show_video",0);while(nano::ok()){     nano::spinOnce();    //不是必须，若程序中订阅话题则必须，否则回掉函数不起作用。loop_rate.sleep();    //按前面设置的10Hz频率将程序挂起}return 0;
}

可实现设备内进程间通讯，及设备间通讯。

Linux nanomsg tcp sub pub相关推荐

Linux Kernel TCP/IP Stack — L2 Layer — Linux Bridge（虚拟网桥）的基本操作
目录文章目录目录 Linux bridge 的基本操作创建 Bridge 将 veth pair 连上 Bridge 为 Bridge 配置 IP 地址将物理网卡接口设备挂靠 Bridge L ...
linux下TCP通信简单实例
linux下TCP通信简单实例基于TCP(面向连接)的socket编程,分为服务器端和客户端服务器端的流程如下: (1)创建套接字(socket) (2)将套接字绑定到一个本地地址和端口上(bin ...
Linux系统TCP内核参数优化总结
简介:Linux系统TCP内核参数优化总结日常运维工作中,会遇到很多TCP相关的问题,网上有很多文章介绍需要优化哪些TCP内核参数,但是并没有很详细说明优化的依据.适用的场景是什么,如果我们不了解各 ...
LINUX 下tcp 和 udp 套接字收发缓冲区的大小决定规则 .
const int udp_recvbufsize = 384 * 1024 ; int result = ::setsockopt(m_hSocket, SOL_SOCKET, SO_RCVBUF, ...
linux数据包注释,关于 linux中TCP数据包(SKB)序列号的小笔记
关于 SKB序列号的小笔记为了修改TCP协议,现在遇到了要改动tcp分组的序列号,但是只是在tcp_sendmsg函数中找到了SKB的end_seq 一直没有找到seq 不清楚在那里初始化了,就 ...
linux上TCP connection timeout的原因查找
linux上TCP connection timeout的原因查找好久没有写文章了, 今天解决了一个网络连接超时的问题, 记录以备查看. 最近在线上nginx经常出现输出connection tim ...
linux初始化TCP服务失败,深入Linux系统追踪TCP初始化
struct socket { socket_state state;shorttype; unsignedlongflags; struct socket_wq*wq; structfile *fi ...
linux满负荷运行tail,linux内核tcp调优规范与方案
1.TCP常用内核参数优化上一篇我们介绍了服务器上有大量的TIME_WAIT等待,可能造成的危害,以及给web服务器带来负担.如何解决这个问题呢,其实,解决思路很简单,就是让服务器能够快速回收和重用 ...
linux udp 端口测试,RAKsmart：Linux下TCP/UDP 端口测试及验证方法说明
RAKsmart:Linux下TCP/UDP 端口测试及验证方法说明2020-06-11 在 Linux 系统中,有时需要在系统中测试端口的连通性,以便确认系统的TCP.UDP协议栈是否可以正常运行. ...
Linux中TCP listen()的参数
原文:http://blog.csdn.net/kobejayandy/article/details/46989465 #include<sys/socket.h> int lis ...

Linux nanomsg tcp sub pub

一、安装

二、简单sub pub

三、封装sub pub

Linux nanomsg tcp sub pub相关推荐

最新文章

热门文章