005 ZeroMQ REQ-<ROUTER-ROUTER>-REQ 路由模式
2016-04-04 15:00
561 查看
框架:
代码放在最后。
实践,为了便于理解,没有使用多线程,tcp取代ipc通信,只加入了一个客户端和一个工人:
消息路由原理:
client.cpp
proxy.cpp
worker.cpp
代码放在最后。
实践,为了便于理解,没有使用多线程,tcp取代ipc通信,只加入了一个客户端和一个工人:
消息路由原理:
client.cpp
#include<zmq.h> #include<iostream> using std::endl; using std::cout; int main() { void* pContext = zmq_ctx_new(); void* pReqSock = zmq_socket(pContext, ZMQ_REQ); zmq_setsockopt(pReqSock, ZMQ_IDENTITY, "client", 7); zmq_connect(pReqSock, "tcp://localhost:7001"); while (true) { printf("press Enter to send message\n"); getchar(); zmq_msg_t msgOut; zmq_msg_init_size(&msgOut, 6); sprintf_s((char*)zmq_msg_data(&msgOut), 6, "hello"); zmq_msg_send(&msgOut, pReqSock,0); zmq_msg_close(&msgOut); zmq_msg_t msgIn; zmq_msg_init(&msgIn); zmq_msg_recv(&msgIn, pReqSock, 0); cout << "receive:" << (char*)zmq_msg_data(&msgIn) << endl; zmq_msg_close(&msgIn); } zmq_close(pReqSock); zmq_ctx_destroy(pContext); return 0; }
proxy.cpp
#define _CRT_SECURE_NO_WARNINGS #include<zmq.h> #include<iostream> #include<cassert> using std::endl; using std::cout; int main() { void* pContext = zmq_ctx_new(); void* frontEnd = zmq_socket(pContext, ZMQ_ROUTER); void* backEnd = zmq_socket(pContext, ZMQ_ROUTER); zmq_bind(frontEnd, "tcp://*:7001"); zmq_bind(backEnd, "tcp://*:7002"); char pWorkerStr[7];//用于保存工人ID; int szWorkerID=7; while (true) { zmq_pollitem_t items[]= { {frontEnd,0,ZMQ_POLLIN,0}, {backEnd, 0,ZMQ_POLLIN,0} }; cout << "001" << endl; int rc = zmq_poll(items, 2, -1); if (rc==-1) { break; } cout << "002" << endl; //只在后端有可用工人时,才处理前端请求; if (items[1].revents&ZMQ_POLLIN) { //后端会收到5帧; //第1帧,工人的ID; zmq_msg_t frame1; zmq_msg_init(&frame1); zmq_msg_recv(&frame1, backEnd, 0); memcpy(pWorkerStr, zmq_msg_data(&frame1), 7);//保存工人ID; cout <<"f1:"<<(char*)zmq_msg_data(&frame1); zmq_msg_close(&frame1); //第2帧,空; zmq_msg_t frame2; zmq_msg_init(&frame2); int size0= zmq_msg_recv(&frame2, backEnd, 0); cout << "-f2:" << size0; zmq_msg_close(&frame2); //第3帧,客户端ID/或者ready; zmq_msg_t frame3; zmq_msg_init(&frame3); zmq_msg_recv(&frame3, backEnd, 0); cout <<"-f3:"<< (char*)zmq_msg_data(&frame3) << endl; cout << "compare:" << strcmp((char*)zmq_msg_data(&frame3), "client"); if (zmq_msg_more(&frame3)) { cout << "002-6" << endl; //第4帧,空; zmq_msg_t frame4; zmq_msg_init(&frame4); zmq_msg_recv(&frame4, backEnd, 0); //第5帧,来自工人的回答; zmq_msg_t frame5; zmq_msg_init(&frame5); zmq_msg_recv(&frame5, backEnd, 0); cout << "002-5" << endl; //将后3帧给前端; zmq_msg_send(&frame3, frontEnd, ZMQ_SNDMORE); zmq_msg_send(&frame4, frontEnd, ZMQ_SNDMORE); zmq_msg_send(&frame5, frontEnd, 0); zmq_msg_close(&frame3); zmq_msg_close(&frame4); zmq_msg_close(&frame5); } else { zmq_msg_close(&frame3); } cout << "003" << endl; } //来自前端的请求; if (items[0].revents&ZMQ_POLLIN) { //前端收会到3帧:[client身份][空][请求]; cout << "002-7" << endl; zmq_msg_t frame3; zmq_msg_init(&frame3); zmq_msg_recv(&frame3, frontEnd, 0); zmq_msg_t frame4; zmq_msg_init(&frame4); zmq_msg_recv(&frame4, frontEnd, 0); zmq_msg_t frame5; zmq_msg_init(&frame5); zmq_msg_recv(&frame5, frontEnd, 0); //再制造2帧:[worker身份][空]; zmq_msg_t frame1; zmq_msg_init_size(&frame1, 7); memcpy(zmq_msg_data(&frame1), pWorkerStr, 7); zmq_msg_t frame2; zmq_msg_init(&frame2); //将这5帧发给后端[WorkerID] [NULL] [clientID] [NULL] [msg] zmq_msg_send(&frame1, backEnd, ZMQ_SNDMORE); zmq_msg_send(&frame2, backEnd, ZMQ_SNDMORE); zmq_msg_send(&frame3, backEnd, ZMQ_SNDMORE); zmq_msg_send(&frame4, backEnd, ZMQ_SNDMORE); zmq_msg_send(&frame5, backEnd, 0); zmq_msg_close(&frame1); zmq_msg_close(&frame2); zmq_msg_close(&frame3); zmq_msg_close(&frame4); zmq_msg_close(&frame5); } } zmq_close(frontEnd); zmq_close(backEnd); zmq_ctx_destroy(pContext); return 0; }
worker.cpp
#include<zmq.h> #include<iostream> #include<cassert> using std::endl; using std::cout; int main() { void* pContext = zmq_ctx_new(); void* pReqSock = zmq_socket(pContext, ZMQ_REQ); zmq_setsockopt(pReqSock, ZMQ_IDENTITY, "worker", 7); zmq_connect(pReqSock, "tcp://localhost:7002"); //把自己的身份通过第一次消息发送给后端; cout << "001" << endl; zmq_msg_t msgOut; zmq_msg_init_data(&msgOut, "ready", 6, 0, 0); zmq_msg_send(&msgOut, pReqSock, 0); zmq_msg_close(&msgOut); cout << "002" << endl; while (true) { //工人会收到三帧;1客户端身份;2空帧;3客户端消息; //第一帧; zmq_msg_t clientID; zmq_msg_init(&clientID); int szClientID=zmq_msg_recv(&clientID, pReqSock, 0); char* pClientStr = new char[szClientID]; memset(pClientStr, 0, szClientID ); memcpy(pClientStr, zmq_msg_data(&clientID), szClientID); cout << "frame1 client ID is:" << pClientStr << endl; zmq_msg_close(&clientID); cout << "003" << endl; //第二帧; zmq_msg_t zeroFrame; zmq_msg_init(&zeroFrame); int szFrame2=zmq_msg_recv(&zeroFrame, pReqSock, 0); assert(szFrame2 == 0); cout << "frame2 is:0" << endl; zmq_msg_close(&zeroFrame); //第三帧; zmq_msg_t msg; zmq_msg_init(&msg); zmq_msg_recv(&msg, pReqSock, 0); cout << "frame3 is:" << (char*)zmq_msg_data(&msg) << endl; zmq_msg_close(&msg); cout << "004" << endl; //回复; //第一帧; zmq_msg_t frame1; zmq_msg_init_data(&frame1, pClientStr,szClientID, 0, 0);//使用之前保存的客户端的身份; zmq_msg_send(&frame1, pReqSock, ZMQ_SNDMORE); zmq_msg_close(&frame1); //第二帧; zmq_msg_t frame2; zmq_msg_init(&frame2); zmq_msg_send(&frame2, pReqSock, ZMQ_SNDMORE); zmq_msg_close(&frame2); //第三帧; static int i = 0; zmq_msg_t frame3; zmq_msg_init_size(&frame3, 7); sprintf_s((char*)zmq_msg_data(&frame3), 7, "world%d", i++); zmq_msg_send(&frame3, pReqSock, 0); zmq_msg_close(&frame3); cout << "005-final" << endl; delete []pClientStr;//销毁身份; } zmq_close(pReqSock); zmq_ctx_destroy(pContext); return 0; }
相关文章推荐
- 第216期 辞退员工面谈有妙招 离职员工情绪管理不可少
- linux 编译ffmpeg 支持x264, x265
- 《Linux内核分析》第七周: 可执行程序的装载
- 101空降师506团2营E连全体成员
- 关于#define
- HDU 4616 Game 【有限制的树型dp】
- IE版本的标准定义
- Unity VR游戏开发干货教程:VR中的交互方式
- 吃吃饭
- 吃吃饭
- [Zedboard Linux系统移植]-从MACHINE_START開始
- android 关于启动模式的进一步理解
- SIP注册过程
- 吃吃饭
- 吃吃饭
- 装饰器模式
- 【2016.04.01】第六次模拟考试考试分析
- BZOJ3163: [Heoi2013]Eden的新背包问题
- 控制算法、基本定义及其实现
- 吃吃饭