ZeroMQ 在多线程应用程序中处理中断
ZeroMQ handling interrupt in multithreaded application
ZeroMQ 在多线程环境中的优雅退出
规格:ubuntu 16.04 与 c++11,libzmq:4.2.3
示例代码
static int s_interrupted = 0;
static void s_signal_handler (int signal_value)
{
s_interrupted = 1;
//some code which will tell main thread to exit
}
static void s_catch_signals (void)
{
struct sigaction action;
action.sa_handler = s_signal_handler;
action.sa_flags = 0;
sigemptyset (&action.sa_mask);
sigaction (SIGINT, &action, NULL);
sigaction (SIGTERM, &action, NULL);
}
static void Thread((zsock_t *pipe, void *)
{
zmq::context_t context(1);
zmq::socket_t requester1(context,ZMQ_DEALER);
zmq::socket_t requester2(context,ZMQ_DEALER);
requester1.connect(address1);
requester2.connect(address2);
zmq_pollitem_t items []=
{{requester1,0,ZMQ_POLLIN,0},
{requester2,0,ZMQ_POLLIN,0}};
while(true)
{
zmq::message_t message;
zmq::poll (items, 2, -1);
if (items [0].revents & ZMQ_POLLIN)
{
requester1.recv(&message);
}
if (items [1].revents & ZMQ_POLLIN)
{
requester2.recv(&message);
}
}
}
int main()
{
.
//some code
.
zactor_t *actor = zactor_new (Threaded, nullptr);
s_catch_signals();
.
//continue
.
//wait till thread finishes to exit
return 0;
}
现在当中断发生时,它会从主线程调用信号处理程序。我不知何故需要告诉线程(轮询器)退出信号处理程序。有什么想法可以实现吗?
根据 ZMQ 文档,您有 2 "idiomatic" 种处理方式:
经过测试,似乎zmq::poll 没有在SIGINT 上抛出异常。
因此,解决方案似乎是使用专用于关闭的套接字。
解决方案如下所示:
#include <iostream>
#include <thread>
#include <signal.h>
#include <zmq.hpp>
zmq::context_t* ctx;
static void s_signal_handler (int signal_value)
{
std::cout << "Signal received" << std::endl;
zmq::socket_t stop_socket(*ctx, ZMQ_PAIR);
stop_socket.connect("inproc://stop_address");
zmq::message_t msg("0", 1);
stop_socket.send(msg);
std::cout << "end sighandler" << std::endl;
}
static void s_catch_signals (void)
{
struct sigaction action;
action.sa_handler = s_signal_handler;
action.sa_flags = 0;
sigemptyset (&action.sa_mask);
sigaction (SIGINT, &action, NULL);
sigaction (SIGTERM, &action, NULL);
}
void thread(void)
{
std::cout << "Thread Begin" << std::endl;
zmq::context_t context (1);
ctx = &context;
zmq::socket_t requester1(context,ZMQ_DEALER);
zmq::socket_t requester2(context,ZMQ_DEALER);
zmq::socket_t stop_socket(context, ZMQ_PAIR);
requester1.connect("tcp://127.0.0.1:36483");
requester2.connect("tcp://127.0.0.1:36483");
stop_socket.bind("inproc://stop_address");
zmq_pollitem_t items []=
{
{requester1,0,ZMQ_POLLIN,0},
{requester2,0,ZMQ_POLLIN,0},
{stop_socket,0,ZMQ_POLLIN,0}
};
while ( true )
{
// Blocking read will throw on a signal
int rc = 0;
std::cout << "Polling" << std::endl;
rc = zmq::poll (items, 3, -1);
zmq::message_t message;
if(rc > 0)
{
if (items [0].revents & ZMQ_POLLIN)
{
requester1.recv(&message);
}
if (items [1].revents & ZMQ_POLLIN)
{
requester2.recv(&message);
}
if(items [2].revents & ZMQ_POLLIN)
{
std::cout << "message stop received " << std::endl;
break;
}
}
}
requester1.setsockopt(ZMQ_LINGER, 0);
requester2.setsockopt(ZMQ_LINGER, 0);
stop_socket.setsockopt(ZMQ_LINGER, 0);
requester1.close();
requester2.close();
stop_socket.close();
std::cout << "Thread end" << std::endl;
}
int main(void)
{
std::cout << "Begin" << std::endl;
s_catch_signals ();
zmq::context_t context (1);
zmq::socket_t router(context,ZMQ_ROUTER);
router.bind("tcp://127.0.0.1:36483");
std::thread t(&thread);
t.join();
std::cout << "end join" << std::endl;
}
请注意,如果您不想将上下文共享给信号处理程序,您可以使用 "ipc://..."。
如果您希望在处理信号时保留 ZMQ 的 Actor 模型的感觉,您可以在 Linux: signalfd manpage 上使用 signalfd 接口。这样你就可以使用 zmq poll 来等待信号被传递,而不是使用信号处理程序。
它还有一个额外的好处,即在处理通过文件描述符传递的信号时,您可以调用任何您喜欢的函数,因为您是在同步处理它,而不是异步处理它。
ZeroMQ 在多线程环境中的优雅退出
规格:ubuntu 16.04 与 c++11,libzmq:4.2.3
示例代码
static int s_interrupted = 0;
static void s_signal_handler (int signal_value)
{
s_interrupted = 1;
//some code which will tell main thread to exit
}
static void s_catch_signals (void)
{
struct sigaction action;
action.sa_handler = s_signal_handler;
action.sa_flags = 0;
sigemptyset (&action.sa_mask);
sigaction (SIGINT, &action, NULL);
sigaction (SIGTERM, &action, NULL);
}
static void Thread((zsock_t *pipe, void *)
{
zmq::context_t context(1);
zmq::socket_t requester1(context,ZMQ_DEALER);
zmq::socket_t requester2(context,ZMQ_DEALER);
requester1.connect(address1);
requester2.connect(address2);
zmq_pollitem_t items []=
{{requester1,0,ZMQ_POLLIN,0},
{requester2,0,ZMQ_POLLIN,0}};
while(true)
{
zmq::message_t message;
zmq::poll (items, 2, -1);
if (items [0].revents & ZMQ_POLLIN)
{
requester1.recv(&message);
}
if (items [1].revents & ZMQ_POLLIN)
{
requester2.recv(&message);
}
}
}
int main()
{
.
//some code
.
zactor_t *actor = zactor_new (Threaded, nullptr);
s_catch_signals();
.
//continue
.
//wait till thread finishes to exit
return 0;
}
现在当中断发生时,它会从主线程调用信号处理程序。我不知何故需要告诉线程(轮询器)退出信号处理程序。有什么想法可以实现吗?
根据 ZMQ 文档,您有 2 "idiomatic" 种处理方式:
经过测试,似乎zmq::poll 没有在SIGINT 上抛出异常。 因此,解决方案似乎是使用专用于关闭的套接字。 解决方案如下所示:
#include <iostream>
#include <thread>
#include <signal.h>
#include <zmq.hpp>
zmq::context_t* ctx;
static void s_signal_handler (int signal_value)
{
std::cout << "Signal received" << std::endl;
zmq::socket_t stop_socket(*ctx, ZMQ_PAIR);
stop_socket.connect("inproc://stop_address");
zmq::message_t msg("0", 1);
stop_socket.send(msg);
std::cout << "end sighandler" << std::endl;
}
static void s_catch_signals (void)
{
struct sigaction action;
action.sa_handler = s_signal_handler;
action.sa_flags = 0;
sigemptyset (&action.sa_mask);
sigaction (SIGINT, &action, NULL);
sigaction (SIGTERM, &action, NULL);
}
void thread(void)
{
std::cout << "Thread Begin" << std::endl;
zmq::context_t context (1);
ctx = &context;
zmq::socket_t requester1(context,ZMQ_DEALER);
zmq::socket_t requester2(context,ZMQ_DEALER);
zmq::socket_t stop_socket(context, ZMQ_PAIR);
requester1.connect("tcp://127.0.0.1:36483");
requester2.connect("tcp://127.0.0.1:36483");
stop_socket.bind("inproc://stop_address");
zmq_pollitem_t items []=
{
{requester1,0,ZMQ_POLLIN,0},
{requester2,0,ZMQ_POLLIN,0},
{stop_socket,0,ZMQ_POLLIN,0}
};
while ( true )
{
// Blocking read will throw on a signal
int rc = 0;
std::cout << "Polling" << std::endl;
rc = zmq::poll (items, 3, -1);
zmq::message_t message;
if(rc > 0)
{
if (items [0].revents & ZMQ_POLLIN)
{
requester1.recv(&message);
}
if (items [1].revents & ZMQ_POLLIN)
{
requester2.recv(&message);
}
if(items [2].revents & ZMQ_POLLIN)
{
std::cout << "message stop received " << std::endl;
break;
}
}
}
requester1.setsockopt(ZMQ_LINGER, 0);
requester2.setsockopt(ZMQ_LINGER, 0);
stop_socket.setsockopt(ZMQ_LINGER, 0);
requester1.close();
requester2.close();
stop_socket.close();
std::cout << "Thread end" << std::endl;
}
int main(void)
{
std::cout << "Begin" << std::endl;
s_catch_signals ();
zmq::context_t context (1);
zmq::socket_t router(context,ZMQ_ROUTER);
router.bind("tcp://127.0.0.1:36483");
std::thread t(&thread);
t.join();
std::cout << "end join" << std::endl;
}
请注意,如果您不想将上下文共享给信号处理程序,您可以使用 "ipc://..."。
如果您希望在处理信号时保留 ZMQ 的 Actor 模型的感觉,您可以在 Linux: signalfd manpage 上使用 signalfd 接口。这样你就可以使用 zmq poll 来等待信号被传递,而不是使用信号处理程序。
它还有一个额外的好处,即在处理通过文件描述符传递的信号时,您可以调用任何您喜欢的函数,因为您是在同步处理它,而不是异步处理它。