NanoMsg框架C++的相关函数介绍

1. mongos、nanomsg简述

来自：Go的nanomsg实现 mangos - 云+社区 - 腾讯云

nanomsg是一个消息协议SP ("Scalable Protocols"可扩展协议)的c语言实现，而mangos用golang实现了SP (“Scalable Protocols”)。

消息协议不同于通常我们说的消息队列，是指一个简单的传输会话协议。

mangos重点也是替代直接手写TCP，实现各种场合的通讯范式。

推荐：Fast, Scalable Networking in Go with Mangos – Brave New Geek

那么mangos、nanomsg有何优点么？

主要是：简单、抽象合理、兼容多种语言、轻量级、学习成本低、比自己造的轮子好用很多。

理解的误区：mangos/nanomsg并不是消息队列，也不是RPC框架。

2. zeroMQ、nanomsg和可扩展协议

A Look at Nanomsg and Scalability Protocols (Why ZeroMQ Shouldn’t Be Your First Choice) – Brave New Geek

可以简单理解这些网络框架和协议是对TCP、PGM、IPC、ITC等协议的封装，提供新的接口便于分布式环境下的通信，而zeroMQ较差的可扩展性将其局限于某一些协议，为了解决扩展性等一些其它的问题出现了Nanomsg，Nanomsg 通过为传输和消息传递协议提供可插入的接口来解决这个问题。这意味着支持超出标准集 PUB/SUB、REQ/REP 等的新传输（例如 WebSockets）和新消息模式。

也许最有趣的是 nanomsg 与 ZeroMQ 的哲学背离。nanomsg 不是作为一个通用的网络库，而是打算通过实现所谓的“可扩展性协议”来提供用于构建可扩展和高性能分布式系统的“乐高积木”。这些可扩展协议是通信模式，它们是网络堆栈传输层之上的抽象。这些协议彼此完全分离，因此每个协议都可以体现明确定义的分布式算法。正如 nanomsg 的作者 Martin Sustrik 所说，其目的是通过IETF标准化协议规范。（zeroMQ可以理解成网络库，而NanoMsg更像按照应用场景定义新的可扩展协议，具体选择那一种则需要看应用场景，就分布式场景下NanoMsg的可扩展性可能更适合分布式复杂多变场景）

Nanomsg 目前定义了六种不同的可扩展性协议：PAIR、REQREP、PIPELINE、BUS、PUBSUB 和 SURVEY。

PAIR（双向通信）

PAIR 在两个端点之间实现简单的一对一、双向通信。两个节点可以相互发送消息。

REQREP（客户端请求、服务器回复）

REQREP 协议定义了一种用于构建无状态服务来处理用户请求的模式。客户端发送请求，服务器接收请求，进行一些处理，然后返回响应。

PIPELINE（单向数据流）

PIPELINE 提供单向数据流，这对于创建负载平衡的处理管道非常有用。生产者节点提交分布在消费者节点之间的工作。

BUS（多对多通信）

BUS 允许从每个对等点发送的消息传递到组中的每个其他对等点。

PUBSUB（主题广播）

PUBSUB 允许发布者向零个或多个订阅者多播消息。订阅者可以连接到多个发布者，可以订阅特定的主题，允许他们只接收与他们相关的消息。

SURVEY（向小组提问）

最后一个可扩展性协议是 SURVEY。SURVEY 模式与 PUBSUB 的相似之处在于来自一个节点的消息被广播到整个组，但不同的是组中的每个节点都响应该消息。这开辟了各种各样的应用程序，因为它使您可以快速轻松地一次性查询大量系统的状态。调查受访者必须在调查员配置的时间窗口内做出回应。

3.C++的相关函数

NanoMsg相关函数

微卡智享

下面我们就列一下几个常用的函数，基本的通讯也就是用这几个函数进行处理的。

函数	简单介绍
nn_socket	创建一个套接字
nn_setsockopt	设置套接字的选项
nn_bind	绑定地址
nn_connect	连接另一个套接字
nn_send	发送数据
nn_recv	接收数据

nn_socket

所在头文件：#include <nanomsg/nn.h>，作用为创建一个套接字。

int nn_socket (int domain, int protocol);

参数：

domain：这个参数有两个格式，AF_SP和AF_SP_RAW，AF_SP表示标准的格式，AF_SP_RAW表示一种原始的格式，一般开发使用AF_SP即可。

protocol：设置通讯协议类型。如NN_PAIR等。

返回值：返回套接字。

nn_setsockopt

所在头文件：#include <nanomsg/nn.h> ，作用和原始socket开发中的setsockopt类似，用来设置套接字的选项。

int nn_setsockopt (int s, int level, int option, const void *optval, size_t optvallen);

参数：

s：上面的函数nn_socket的返回值；

level：默认正常0即可（即NN_SOL_SOCKET）；

option：需要改变的选项，一般我们来说主要就是设置超时这块的设置（即NN_SNDTIMEO和NN_RCVTIMEO）；

*optval：上一个option参数对应的值；

optvallen：上一个*optval参数的长度；

返回值：返回不为零表示失败。

nn_bind

所在头文件：#include <nanomsg/nn.h> ，作用绑定地址。

int nn_bind (int s, const char *addr);

参数：

s：上面的函数nn_socket的返回值；

*addr：地址；

返回值：返回小于零表示失败。

nn_connect

所在头文件：#include <nanomsg/nn.h> ，作用连接另一个套接字。

int nn_connect(int s, const char *addr);

参数：

s：上面的函数nn_socket的返回值；

*addr：连接的地址；

返回值：返回小于零表示失败。

nn_send

所在头文件：#include <nanomsg/nn.h> ，作用发送数据。

int nn_send (int s, const void *buf, size_t len, int flags);

参数：

s：上面的函数nn_socket的返回值；

*buf：发送的数据；

len：发送数据的长度；

flags：0表示在阻塞模式下执行，1（NN_DONTWAIT），表示在非阻塞模式下执行。

返回值：返回发送数据的长度，小于零表示错误。

nn_recv

所在头文件：#include <nanomsg/nn.h> ，作用接收数据。

int nn_recv（int s，void * buf，size_t len，int flags）;

参数：

s：上面的函数nn_socket的返回值；

*buf：接收的数据；

len：接收数据的最大长度；

flags：0表示在阻塞模式下执行，1（NN_DONTWAIT），表示在非阻塞模式下执行。

返回值：返回接收到的数据长度。

实例 nnclient.cpp

/** @Copyright <YEAR> <COPYRIGHT HOLDER>: * @Author: lis* @Date: 2022-04-28 00:35:35* @LastEditTime: 2022-04-28 00:48:00* @LastEditors: lis* @Description: nanomsg clinet 的测试* @FilePath: /nanomsg/src/nnclient.cpp*/
#include <stdio.h>
#include <string>
#include <iostream>
#include "nanomsg/reqrep.h"
#include "nanomsg/nn.h"


int main(int argc, char* argv[]) {int m_sock_ctl;if ((m_sock_ctl = nn_socket(AF_SP, NN_REQ)) < 0) {printf("=======control sock error: nn_socket\n");m_sock_ctl = 0;return -1;}if (nn_connect(m_sock_ctl, "tcp://0.0.0.0:10000") < 0) {printf("=======control sock error: nn_connect\n");nn_close(m_sock_ctl);m_sock_ctl = 0;return -2;}
std::string cmd = "Open 192.168.1.44";nn_send(m_sock_ctl, cmd.c_str(), cmd.length(), 0);char* buffer = NULL;int buffer_len = 0;if ((buffer_len = nn_recv(m_sock_ctl, &buffer, NN_MSG, 0)) <= 0) {printf("=======control sock error: nn_recv\n");return -1;}std::string msg = std::string(buffer, buffer_len);printf("%s\n", msg.c_str());nn_freemsg(buffer);return 0;
}

server.cpp

/** @Copyright <YEAR> <COPYRIGHT HOLDER>: * @Author: lis* @Date: 2022-04-28 00:00:37* @LastEditTime: 2022-04-28 00:45:24* @LastEditors: lis* @Description: nanomsg server 的测试* @FilePath: /nanomsg/src/server.cpp*/
#include <stdio.h>
#include <string>
#include <iostream>
#include "nanomsg/reqrep.h"
#include "nanomsg/nn.h"

#define CTL_URL "tcp://0.0.0.0:10000"

int main(int argc, char* argv[]) {int m_sock_ctl;if ((m_sock_ctl = nn_socket(AF_SP, NN_REP)) < 0) {printf("=======control sock error: nn_socket\n");return -1;}if (nn_bind(m_sock_ctl, CTL_URL) < 0) {printf("=======control sock error: nn_bind\n");nn_close(m_sock_ctl);m_sock_ctl = 0;return -2;}
printf("=======control sock OK: %s\n", CTL_URL);
// int timeout = 1000;// if (nn_setsockopt(m_sock_ctl,//                 NN_SOL_SOCKET,//                 NN_SNDTIMEO,//                 &timeout,//                 sizeof(int)) < 0) {//     printf("=======channel sock error: nn_setsockopt\n");//     nn_close(m_sock_ctl);//     return -1;// }// if (nn_setsockopt(m_sock_ctl,//                 NN_SOL_SOCKET,//                 NN_RCVTIMEO,//                 &timeout,//                 sizeof(int)) < 0) {//     printf("=======channel sock error: nn_setsockopt\n");//     nn_close(m_sock_ctl);//     return -1;// }

while (1) {char* buffer = NULL;int buffer_len = 0;// 获取请求if ((buffer_len = nn_recv(m_sock_ctl, &buffer, NN_MSG, 0)) <= 0) {printf("=======control sock error: nn_recv\n");return -1;}std::string msg = std::string(buffer, buffer_len);nn_freemsg(buffer);
printf("=========recv msg: %s\n", msg.c_str());std::string result = "test";
printf("=========send msg: %s\n", result.c_str());// 返回相应nn_send(m_sock_ctl, result.c_str(), result.length(), 0);}
nn_close(m_sock_ctl);


}