/**************************************************************技术博客http://www.cnblogs.com/itdef/技术交流群群号码：324164944欢迎c c++ windows驱动爱好者 服务器程序员沟通交流**************************************************************/ROUTER 与 REQ通讯
#include "stdafx.h"
#include "zhelpers.hpp"
#include <thread> void worker_thread(void *arg) {
zmq::context_t context();
zmq::socket_t worker(context, ZMQ_REQ); s_set_id(worker, (intptr_t)arg);
worker.connect("tcp://localhost:5671"); // "ipc" doesn't yet work on windows. int total = ;
while () {
// Tell the broker we're ready for work
s_send(worker, "Hi Boss"); // Get workload from broker, until finished
std::string workload = s_recv(worker);
if ("Fired!" == workload) {
std::cout << "Processed: " << total << " tasks" << std::endl;
break;
}
total++; // Do some random work
s_sleep(within() + );
}
return;
} int main() {
zmq::context_t context();
zmq::socket_t broker(context, ZMQ_ROUTER); broker.bind("tcp://*:5671"); // "ipc" doesn't yet work on windows. const int NBR_WORKERS = ;
std::thread workers[NBR_WORKERS];
for (int worker_nbr = ; worker_nbr < NBR_WORKERS; worker_nbr++) {
workers[worker_nbr]= std::thread( worker_thread, (void *)(intptr_t)worker_nbr);
} // Run for five seconds and then tell workers to end
int64_t end_time = s_clock() + ;
int workers_fired = ;
while () {
// Next message gives us least recently used worker
std::string identity = s_recv(broker);
s_recv(broker); // Envelope delimiter
s_recv(broker); // Response from worker s_sendmore(broker, identity);
s_sendmore(broker, "");
// Encourage workers until it's time to fire them
if (s_clock() < end_time)
s_send(broker, "Work harder");
else {
s_send(broker, "Fired!");
if (++workers_fired == NBR_WORKERS)
break;
}
} for (int worker_nbr = ; worker_nbr < NBR_WORKERS; worker_nbr++) {
workers[worker_nbr].join();
}
return ;
}
ROUTER 与 DEALER通讯
// rtdealer_cpp.cpp : 定义控制台应用程序的入口点。
// #include "stdafx.h"
//
// Custom routing Router to Dealer
//
// Olivier Chamoux <olivier.chamoux@fr.thalesgroup.com> #include "zhelpers.hpp"
#include <thread> static void *
worker_task(void *args)
{
zmq::context_t context();
zmq::socket_t worker(context, ZMQ_DEALER); #if (defined (WIN32))
s_set_id(worker, (intptr_t)args);
#else
s_set_id(worker); // Set a printable identity
#endif worker.connect("tcp://localhost:5671"); int total = ;
while () {
// Tell the broker we're ready for work
s_sendmore(worker, "");
s_send(worker, "Hi Boss"); // Get workload from broker, until finished
s_recv(worker); // Envelope delimiter
std::string workload = s_recv(worker);
// .skip
if ("Fired!" == workload) {
std::cout << "Completed: " << total << " tasks" << std::endl;
break;
}
total++; // Do some random work
s_sleep(within() + );
} return NULL;
} // .split main task
// While this example runs in a single process, that is just to make
// it easier to start and stop the example. Each thread has its own
// context and conceptually acts as a separate process.
int main() {
zmq::context_t context();
zmq::socket_t broker(context, ZMQ_ROUTER); broker.bind("tcp://*:5671");
srandom((unsigned)time(NULL)); const int NBR_WORKERS = ;
std::thread workers[NBR_WORKERS];
int worker_nbr = ;
for (; worker_nbr < NBR_WORKERS; ++worker_nbr) {
workers[worker_nbr] = std::thread( worker_task, (void *)(intptr_t)worker_nbr);
} // Run for five seconds and then tell workers to end
int64_t end_time = s_clock() + ;
int workers_fired = ;
while () {
// Next message gives us least recently used worker
std::string identity = s_recv(broker);
{
s_recv(broker); // Envelope delimiter
s_recv(broker); // Response from worker
} s_sendmore(broker, identity);
s_sendmore(broker, ""); // Encourage workers until it's time to fire them
if (s_clock() < end_time)
s_send(broker, "Work harder");
else {
s_send(broker, "Fired!");
if (++workers_fired == NBR_WORKERS)
break;
}
} for (int worker_nbr = ; worker_nbr < NBR_WORKERS; ++worker_nbr) {
workers[worker_nbr].join();
} return ;
}
DEALER REQ区别在于 REQ模式socket发送信息时候会自动在信息前添加“”空白信息作为隔离
DEALER模式socket需要手动添加 s_sendmore(socket, "");
而ROUTER 在接受信息前会受到来源socket的地址信息 而发送时会将地址信息添加到socket的信息之前 同时还要在在信息前添加“”空白信息作为隔离
s_sendmore(broker, identity);
s_sendmore(broker, "");
均衡负载 此段代码比较复杂 理解较难
我将代码设置为单线程 然后调试 纠正了一些我理解的router模式
稍后画图
// lbbroker_cpp.cpp : 定义控制台应用程序的入口点。
// #include "stdafx.h"
#include "zhelpers.hpp"
#include <thread>
#include <queue> void client_thread(void* arg) {
zmq::context_t context();
zmq::socket_t client(context, ZMQ_REQ); s_set_id(client, (intptr_t));
client.connect("tcp://localhost:5672"); s_send(client, "HELLO");
std::string reply = s_recv(client);
std::cout << "Client: " << reply << std::endl;
return;
} void worker_thread(void* arg){
zmq::context_t context();
zmq::socket_t worker(context, ZMQ_REQ); s_set_id(worker, (intptr_t));
worker.connect("tcp://localhost:5673"); s_send(worker, "READY");
while () {
std::string address = s_recv(worker);
{
std::string empty = s_recv(worker);
assert(empty.size() == );
}
std::string request = s_recv(worker);
std::cout << "Worker: " << request << std::endl; s_sendmore(worker, address);
s_sendmore(worker, "");
s_send(worker, "OK");
}
return;
} int main()
{
zmq::context_t context();
zmq::socket_t frontend(context, ZMQ_ROUTER);
zmq::socket_t backend(context, ZMQ_ROUTER); frontend.bind("tcp://*:5672");
backend.bind("tcp://*:5673"); int client_nbr;
std::thread client[];
for (client_nbr = ; client_nbr < ; client_nbr++)
{
client[client_nbr] = std::thread(client_thread,(void *)(intptr_t)client_nbr);
} int worker_nbr;
std::thread worker[];
for (worker_nbr = ; worker_nbr < ; worker_nbr++)
{
worker[worker_nbr] = std::thread(worker_thread, (void *)(intptr_t)worker_nbr);
} std::queue<std::string> worker_queue; while ()
{
zmq::pollitem_t items[] = {
{backend,,ZMQ_POLLIN,},
{frontend,,ZMQ_POLLIN,}
};
if (worker_queue.size())
zmq::poll(&items[], , -);
else
zmq::poll(&items[], , -); if (items[].revents & ZMQ_POLLIN) { // Queue worker address for LRU routing
worker_queue.push(s_recv(backend)); {
// Second frame is empty
std::string empty = s_recv(backend);
assert(empty.size() == );
} // Third frame is READY or else a client reply address
std::string client_addr = s_recv(backend); // If client reply, send rest back to frontend
if (client_addr.compare("READY") != ) { {
std::string empty = s_recv(backend);
assert(empty.size() == );
} std::string reply = s_recv(backend);
s_sendmore(frontend, client_addr);
s_sendmore(frontend, "");
s_send(frontend, reply); if (--client_nbr == )
break;
}
}
if (items[].revents & ZMQ_POLLIN) { // Now get next client request, route to LRU worker
// Client request is [address][empty][request]
std::string client_addr = s_recv(frontend); {
std::string empty = s_recv(frontend);
assert(empty.size() == );
} std::string request = s_recv(frontend); std::string worker_addr = worker_queue.front();//worker_queue [0];
worker_queue.pop(); s_sendmore(backend, worker_addr);
s_sendmore(backend, "");
s_sendmore(backend, client_addr);
s_sendmore(backend, "");
s_send(backend, request);
}
} for (client_nbr = ; client_nbr < ; client_nbr++)
{
client[client_nbr].join();
} for (worker_nbr = ; worker_nbr < ; worker_nbr++)
{
worker[worker_nbr].join();
} return ;
}
此处图仅为个人领悟 不能保证完全正确 请谨慎参考
开始看官方的图示 我以为上述代码是如官方所示
但是对于router来说
connect到router的客户端或者工作者(client worker)
与router直接send到router的信息处理上是有区别的
下图用直线和箭头来区分
我将代码设置为单线程 然后调试 得出流程如图:
由于新建线程的随机性
步骤1 2 是的先后次序是随机的
1 client 发送信息到frontend router
信息格式为
"0001"
""
"HELLO"
2 worker发送信息到backend router
信息格式为
"0009"
""
"READY"
3函数主体进入POLL循环
根据代码
if (worker_queue.size())
zmq::poll(&items[0], 2, -1);
else
zmq::poll(&items[0], 1, -1);
由于空闲worker队列中暂时无记录
所以仅仅对backend 进行poll轮询
首先接受信息记录的通讯ID 为9 并push进worker队列
worker_queue.push(s_recv(backend));
然后接受空字节分隔符 在接受发送的正文“READY”
接收信息格式如下:
"0009"
""
"READY"
由于此次接受的正文是“READY” 根据代码不进入发送流程
if (client_addr.compare("READY") != 0)
步骤3结束
4 主体代码进入第二次POLL轮询
frontend 接收信息如下
信息格式为
"0001"
""
"HELLO"
弹出之前接收空闲worker的ID
std::string worker_addr = worker_queue.front();
然后向backend发送信息格式如下:
"0009"
""
"0001"
""
"HELLO"
5 backend接收信息后直接路由到ID为0009的worker
6 worker接收到信息格式为
"0001"
""
"HELLO"
然后发送信息格式:
"0001"
""
"OK"
7 backend接收到步骤6的信息：
将步骤6的worker的id 0009 push仅worker队列
worker_queue.push(s_recv(backend));
此次接受的信息格式为
"0009"
""
"001"
""
"OK"
由于此次接受正文不是"READY"
进入发送模式 发送信息到frontend
发送信息格式为
"0001"
""
"OK"
8 frontend直接路由此信息到ID 0001的client
最后client接收到的信息格式为
"OK"