libev+nanomsg实现多线程通信及事件轮询实例demo

谁借莪１个温暖的怀抱￠ 2022-02-26 02:10 356阅读 0赞

概述：  
       在我们刚接触程序编码的时候，我们要轮询数据有没有发过来，我们最多的可能还是使用while+sleep这样的组合，这对于处理来说是一个效率很低的方法同时还消耗cpu,那么在多线程编程中使用libev+nanomsg会不会提高效率呢。下面实例主要工作如下：A B C 三个线程通过nanomsg通信，A线程作为主线程，控制中枢，B C请求均通过A.那么在实际应用中，比如B模块是接收客户请求并解析发送控制命令的，C模块是负责干活的，A是总控，所有控制命令通过A下发到别的模块，这样维护都很方便。那看下怎么实现的吧。

实例Demo:  
\#include <stdio.h>  
\#include <stdint.h>  
\#include <unistd.h>  
\#include <string.h>  
\#include <time.h>  
\#include <ev.h>  
\#include <pthread.h>  
\#include <nanomsg/pair.h>  
\#include <nanomsg/nn.h>  
   
   
/\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*  
动态库：libev nanomsg  
概述：A B C 三个线程通过nanomsg通信，A线程作为主线程，控制中枢，B C请求均通过A.  
demo示范：  
    A为指令处理模块  
    B为指令接收模块  
    C为指令执行模块  
    B -> A   开灯  
    A -> C   开灯  
    C :      执行开灯  
    C -> A   OK  
    A -> B   OK  
总结：  
    这只是简单的测试使用例子，你可以通过在这个框架的基础上做更多的功能，对于多线程编程这将是一个不  
错的选择.  
\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*/  
   
typedef struct \{  
    int n;          //nanomsg socket  
    int s;          //nanomsg recieve fd  
\}nanomsg\_info\_t;  
   
typedef struct \{  
    nanomsg\_info\_t ab;  
    nanomsg\_info\_t ac;  
\}Aloop\_ctrl\_t;  
   
typedef struct \{  
    nanomsg\_info\_t ba;  
\}Bloop\_ctrl\_t;  
   
typedef struct \{  
    nanomsg\_info\_t ca;  
\}Cloop\_ctrl\_t;  
   
/\*获取系统时间打印\*/  
uint32\_t print\_timenow()  
\{  
    time\_t now;  
    struct tm \*tm\_now;  
    time(&now);  
    tm\_now = localtime(&now);  
    uint32\_t times = tm\_now->tm\_hour \* 3600 + tm\_now->tm\_min \* 60 + tm\_now->tm\_sec;  
    printf("\[%02d:%02d:%02d\]\\r\\n", tm\_now->tm\_hour, tm\_now->tm\_min, tm\_now->tm\_sec);  
    return times;  
\}  
   
/\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*子线程C相关\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*/  
   
static void watcher\_c\_cb (struct ev\_loop \*loop ,struct ev\_io \*w, int revents)  
\{  
    void \*user\_data = ev\_userdata(loop);  
    Cloop\_ctrl\_t \*Cloop\_ctrl = (Cloop\_ctrl\_t \*)user\_data;  
    uint8\_t \*dat = NULL;  
    uint32\_t bytes = nn\_recv(Cloop\_ctrl->ca.n, &dat, NN\_MSG, NN\_DONTWAIT);  
    if (bytes <= 0) \{  
        return;  
    \}  
    printf("C:%s (A->C)\\r\\n", (char \*)dat);  
    nn\_freemsg(dat);  
    //接收成功，发送OK  
    char \*str = "OK";  
    uint8\_t \*udata = nn\_allocmsg(3, 0);  
    if (NULL != udata) \{  
        memcpy(udata, str, 3);  
        nn\_send(Cloop\_ctrl->ca.n, &udata, NN\_MSG, NN\_DONTWAIT);  
    \}  
      
\}  
   
int C\_nanomsg\_init(Cloop\_ctrl\_t \*Cloop\_ctrl)  
\{  
    Cloop\_ctrl->ca.n = nn\_socket(AF\_SP, NN\_PAIR);  
    if (Cloop\_ctrl->ca.n < 0) \{  
        return -1;  
    \}  
    if (nn\_connect(Cloop\_ctrl->ca.n, "inproc://c2a\_loop") < 0) \{  
        return -1;  
    \}  
    size\_t size = sizeof(size\_t);  
    if (nn\_getsockopt(Cloop\_ctrl->ca.n, NN\_SOL\_SOCKET, NN\_RCVFD, (char \*)&Cloop\_ctrl->ca.s, &size) < 0) \{  
        return -1;  
    \}  
    return 0;  
\}  
   
struct ev\_loop\* C\_loop\_init(Cloop\_ctrl\_t \*Cloop\_ctrl)  
\{  
    static struct ev\_io watcher\_c;  
    struct ev\_loop \*loop = ev\_loop\_new(EVBACKEND\_EPOLL);  
    if (NULL == loop) \{  
        printf("create C loop failed\\r\\n");  
        return NULL;  
    \}  
    ev\_io\_init (&watcher\_c, watcher\_c\_cb, Cloop\_ctrl->ca.s, EV\_READ);  
    ev\_io\_start (loop, &watcher\_c);  
    return loop;  
\}  
   
void \*C\_thread(void \*arg)  
\{  
    Cloop\_ctrl\_t Cloop\_ctrl;  
    if (C\_nanomsg\_init(&Cloop\_ctrl) < 0) \{  
        printf("nanomsg init failed\\r\\n");  
        return ;  
    \}  
    struct ev\_loop\* Cloop = C\_loop\_init(&Cloop\_ctrl);  
    if (NULL == Cloop) \{  
        printf("Cloop init failed\\r\\n");  
        return ;  
    \}  
    ev\_set\_userdata(Cloop, &Cloop\_ctrl);  
    ev\_run (Cloop, 0);  
    return ;  
\}  
   
/\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*子线程B相关\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*/  
   
static void watcher\_b\_cb (struct ev\_loop \*loop ,struct ev\_io \*w, int revents)  
\{  
    void \*user\_data = ev\_userdata(loop);  
    Bloop\_ctrl\_t \*Bloop\_ctrl = (Bloop\_ctrl\_t \*)user\_data;  
    uint8\_t \*dat = NULL;  
    uint32\_t bytes = nn\_recv(Bloop\_ctrl->ba.n, &dat, NN\_MSG, NN\_DONTWAIT);  
    if (bytes <= 0) \{  
        return;  
    \}  
    printf("B:%s (A->B)\\r\\n\\r\\n", (char \*)dat);  
    nn\_freemsg(dat);  
\}  
   
static void watcher\_timer\_cb (struct ev\_loop \*loop ,struct ev\_timer \*w, int revents)  
\{  
    static int i = 1;  
    char send\_data\[128\] = \{0\};  
    void \*user\_data = ev\_userdata(loop);  
    Bloop\_ctrl\_t \*Bloop\_ctrl = (Bloop\_ctrl\_t \*)user\_data;  
    sprintf(send\_data, "Please turn on LED\[%d\]", i);  
    i ++;  
    int length = strlen(send\_data) + 1;  
    uint8\_t \*udata = nn\_allocmsg(length, 0);  
    if (NULL != udata) \{  
        memcpy(udata, send\_data, length);  
        nn\_send(Bloop\_ctrl->ba.n, &udata, NN\_MSG, NN\_DONTWAIT);  
    \}  
    //如果定时器不重设，就会默认1秒进入一次回调  
    w->repeat = 10;  
    ev\_timer\_again(loop, w);  
\}  
   
   
int B\_nanomsg\_init(Bloop\_ctrl\_t \*Bloop\_ctrl)  
\{  
    Bloop\_ctrl->ba.n = nn\_socket(AF\_SP, NN\_PAIR);  
    if (Bloop\_ctrl->ba.n < 0) \{  
        return -1;  
    \}  
    if (nn\_connect(Bloop\_ctrl->ba.n, "inproc://b2a\_loop") < 0) \{  
        return -1;  
    \}  
    size\_t size = sizeof(size\_t);  
    if (nn\_getsockopt(Bloop\_ctrl->ba.n, NN\_SOL\_SOCKET, NN\_RCVFD, (char \*)&Bloop\_ctrl->ba.s, &size) < 0) \{  
        return -1;  
    \}  
    return 0;  
\}  
   
struct ev\_loop\* B\_loop\_init(Bloop\_ctrl\_t \*Bloop\_ctrl)  
\{  
    static struct ev\_io watcher\_b;  
    static struct ev\_timer watcher\_timer;  
    struct ev\_loop \*loop = ev\_loop\_new(EVBACKEND\_EPOLL);  
    if (NULL == loop) \{  
        printf("create loop failed\\r\\n");  
        return NULL;  
    \}  
    ev\_io\_init (&watcher\_b, watcher\_b\_cb, Bloop\_ctrl->ba.s, EV\_READ);  
    ev\_timer\_init(&watcher\_timer, watcher\_timer\_cb, 5, 1);  
    ev\_io\_start (loop, &watcher\_b);  
    ev\_timer\_start (loop, &watcher\_timer);  
    return loop;  
\}  
   
void \*B\_thread(void \*arg)  
\{  
    Bloop\_ctrl\_t Bloop\_ctrl;  
    if (B\_nanomsg\_init(&Bloop\_ctrl) < 0) \{  
        printf("nanomsg init failed\\r\\n");  
        return ;  
    \}  
    struct ev\_loop\* Bloop = B\_loop\_init(&Bloop\_ctrl);  
    if (NULL == Bloop) \{  
        printf("Bloop init failed\\r\\n");  
        return ;  
    \}  
    ev\_set\_userdata(Bloop, &Bloop\_ctrl);  
    ev\_run (Bloop, 0);  
    return ;  
\}  
   
/\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*主线程A相关\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*/  
   
static void watcher\_ab\_cb (struct ev\_loop \*loop ,struct ev\_io \*w, int revents)  
\{  
    void \*user\_data = ev\_userdata(loop);  
    Aloop\_ctrl\_t \*Aloop\_ctrl = (Aloop\_ctrl\_t \*)user\_data;  
    uint8\_t \*dat = NULL;  
    uint32\_t bytes = nn\_recv(Aloop\_ctrl->ab.n, &dat, NN\_MSG, NN\_DONTWAIT);  
    if (bytes <= 0) \{  
        return;  
    \}  
    //转发到C  
    printf("A:%s (B->A)\\r\\n", (char \*)dat);  
    nn\_send(Aloop\_ctrl->ac.n, &dat, NN\_MSG, NN\_DONTWAIT);  
\}  
   
static void watcher\_ac\_cb (struct ev\_loop \*loop ,struct ev\_io \*w, int revents)  
\{  
    void \*user\_data = ev\_userdata(loop);  
    Aloop\_ctrl\_t \*Aloop\_ctrl = (Aloop\_ctrl\_t \*)user\_data;  
    uint8\_t \*dat = NULL;  
    uint32\_t bytes = nn\_recv(Aloop\_ctrl->ac.n, &dat, NN\_MSG, NN\_DONTWAIT);  
    if (bytes <= 0) \{  
        return;  
    \}  
    //转发到B  
    printf("A:%s (C->A)\\r\\n", (char \*)dat);  
    nn\_send(Aloop\_ctrl->ab.n, &dat, NN\_MSG, NN\_DONTWAIT);  
\}  
   
/\*主事件nanomsg初始化\*/  
int A\_nanomsg\_init(Aloop\_ctrl\_t \*Aloop\_ctrl)  
\{  
    //ab通信的nanomsg初始化  
    Aloop\_ctrl->ab.n = nn\_socket(AF\_SP, NN\_PAIR);  
    if (Aloop\_ctrl->ab.n < 0) \{  
        return -1;  
    \}  
    if (nn\_bind(Aloop\_ctrl->ab.n, "inproc://b2a\_loop") < 0) \{  
        return -1;  
    \}  
    //获取此端口的接收数据fd描述符  
    size\_t size = sizeof(size\_t);  
    if (nn\_getsockopt(Aloop\_ctrl->ab.n, NN\_SOL\_SOCKET, NN\_RCVFD, (char \*)&Aloop\_ctrl->ab.s, &size) < 0) \{  
        return -1;  
    \}  
    //ac通信的nanomsg初始化  
    Aloop\_ctrl->ac.n = nn\_socket(AF\_SP, NN\_PAIR);  
    if (Aloop\_ctrl->ac.n < 0) \{  
        return -1;  
    \}  
    if (nn\_bind(Aloop\_ctrl->ac.n, "inproc://c2a\_loop") < 0) \{  
        return -1;  
    \}  
    //获取此端口的接收数据fd描述符  
    if (nn\_getsockopt(Aloop\_ctrl->ac.n, NN\_SOL\_SOCKET, NN\_RCVFD, (char \*)&Aloop\_ctrl->ac.s, &size) < 0) \{  
        return -1;  
    \}  
    return 0;  
\}  
   
/\*主事件循环初始化\*/  
struct ev\_loop\* A\_loop\_init(Aloop\_ctrl\_t \*Aloop\_ctrl)  
\{  
    static struct ev\_io watcher\_ab, watcher\_ac;  
    struct ev\_loop \*loop = ev\_loop\_new(EVBACKEND\_EPOLL);  
    if (NULL == loop) \{  
        printf("create loop failed\\r\\n");  
        return NULL;  
    \}  
    //传参  
    ev\_set\_userdata(loop, Aloop\_ctrl);  
    //初始化  
    ev\_io\_init (&watcher\_ab, watcher\_ab\_cb, Aloop\_ctrl->ab.s, EV\_READ);  
    ev\_io\_init (&watcher\_ac, watcher\_ac\_cb, Aloop\_ctrl->ac.s, EV\_READ);  
    ev\_io\_start (loop, &watcher\_ab);  
    ev\_io\_start (loop, &watcher\_ac);  
    return loop;  
\}  
   
/\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*/  
   
int main()  
\{  
    pthread\_t pb,pc;  
    Aloop\_ctrl\_t Aloop\_ctrl;  
    if (A\_nanomsg\_init(&Aloop\_ctrl) < 0) \{  
        printf("nanomsg init failed\\r\\n");  
        return -1;  
    \}  
    struct ev\_loop\* Aloop = A\_loop\_init(&Aloop\_ctrl);  
    if (NULL == Aloop) \{  
        printf("Aloop init failed\\r\\n");  
        return -1;  
    \}  
    //创建线程B  
    if (0 != pthread\_create(&pb, NULL, B\_thread, NULL)) \{  
        printf("create pthread B failed\\r\\n");  
        return -1;  
    \}  
    //创建线程C  
    if (0 != pthread\_create(&pc, NULL, C\_thread, NULL)) \{  
        printf("create pthread C failed\\r\\n");  
        return -1;  
    \}  
    //运行  
    ev\_run(Aloop, 0);  
    return 0;  
\}

编译运行：  
//编译  
gcc -o ev\_nanomsg ev\_nanomsg.c -lev -lnanomsg -lpthread  
   
//运行结果  
A:Please turn on LED\[1\] (B->A)  
C:Please turn on LED\[1\] (A->C)  
A:OK (C->A)  
B:OK (A->B)  
\---------------------