libgo & zeromq —— C++下的协程与消息队列的配合

89 阅读 0 评论 59 点赞

我是靠谱客的博主无情口红，这篇文章主要介绍libgo & zeromq —— C++下的协程与消息队列的配合，现在分享给大家，希望可以做个参考。

libgo通过hook了网络阻塞函数来实现协程的调度，但是zeromq只需要原生的函数，如果hook则会报错，Assertion failed: pfd.revents & POLLIN (src/signaler.cpp:261)，这时候我们只想要libgo的线程调度功能怎么办？

只要重新编译一次libgo，把disable_hook打开就可以,如果使用hook版本的，则会出现这个问题。

于是我们重新编译

编译完成后，重新编译一次server & client 即可完成。

服务端

复制代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
/*
    ZMQ订阅服务器

*/
#include<iostream>
#include<sstream>
#include<zmq.h>

using namespace std;
int main(){
    int port;
    cout<<"请输入你想发布订阅的端口:";
    cin>>port;
    stringstream os;
    os<<"tcp://*:"<<port;

    //开启zmq服务器
    void *ctx=zmq_ctx_new();
    void *publisher=zmq_socket(ctx,ZMQ_PUB);
    int rc=zmq_bind(publisher,os.str().c_str());
    while(1){
        char buffer[10];
        cout<<"发布>>";
        cin>>buffer;
        zmq_send(publisher,buffer,sizeof(buffer),0);
    }
    return 0;
}

客户端

复制代码

/*
    ZQM+libgo订阅客户端，可以协程实现对多个服务器的订阅
*/
#include<iostream>
#include<zmq.h>
#include<libgo/coroutine.h>
#include<sstream>

using namespace std;

int main(){
    stringstream os;
    go [&os](){
        int port;
        while(1){
            cout<<"请输入要订阅的服务器端口:";
            cin>>port;
            os.str("");
            os<<"tcp://127.0.0.1:"<<port;
            string addr=os.str();
            go [addr](){        //开启一个新的协程来处理一个服务器的订阅
                void *ctx=zmq_ctx_new();
                void *subscriber=zmq_socket(ctx,ZMQ_SUB);
                int rc=zmq_connect(subscriber,addr.c_str());//连接到服务器
                rc=zmq_setsockopt(subscriber,ZMQ_SUBSCRIBE,":",1);//设定订阅的过滤选项：只订阅那些以:开头的消息
                while(1){
                    char buffer[10];
                    int len=zmq_recv(subscriber,buffer,sizeof(buffer),ZMQ_NOBLOCK); //不阻塞模式
                    if(len==-1){
                        //如果没有收到消息则跳过，因为设置了不阻塞，这样才能方便协程调度
                        co_sleep(1);//睡眠1ms,避免CPU跑飞
                        co_yield;//让出协程
                        continue;
                    }
                    cout<<"收到订阅:"<<buffer<<endl;
                }
            };
        }
    };
    co_sched.Start(2);//开始调度协程,设置为两个调度线程，一个用来调度主要的while输入订阅端口号，一个用来处理各个订阅者
    return 0;
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
/*
    ZQM+libgo订阅客户端，可以协程实现对多个服务器的订阅
*/
#include<iostream>
#include<zmq.h>
#include<libgo/coroutine.h>
#include<sstream>

using namespace std;

int main(){
    stringstream os;
    go [&os](){
        int port;
        while(1){
            cout<<"请输入要订阅的服务器端口:";
            cin>>port;
            os.str("");
            os<<"tcp://127.0.0.1:"<<port;
            string addr=os.str();
            go [addr](){        //开启一个新的协程来处理一个服务器的订阅
                void *ctx=zmq_ctx_new();
                void *subscriber=zmq_socket(ctx,ZMQ_SUB);
                int rc=zmq_connect(subscriber,addr.c_str());//连接到服务器
                rc=zmq_setsockopt(subscriber,ZMQ_SUBSCRIBE,":",1);//设定订阅的过滤选项：只订阅那些以:开头的消息
                while(1){
                    char buffer[10];
                    int len=zmq_recv(subscriber,buffer,sizeof(buffer),ZMQ_NOBLOCK); //不阻塞模式
                    if(len==-1){
                        //如果没有收到消息则跳过，因为设置了不阻塞，这样才能方便协程调度
                        co_sleep(1);//睡眠1ms,避免CPU跑飞
                        co_yield;//让出协程
                        continue;
                    }
                    cout<<"收到订阅:"<<buffer<<endl;
                }
            };
        }
    };
    co_sched.Start(2);//开始调度协程,设置为两个调度线程，一个用来调度主要的while输入订阅端口号，一个用来处理各个订阅者
    return 0;
}