c++ - ZeroMQ 在使用 std::thread 创建工作者时崩溃
问题描述
我正在将 ZeroMQ Guide Book 中的lbbroker.c 示例移植 到更高级别的 API czmq
。
出于某种原因,使用总是崩溃std::thread
的创建并连接到端点的 a 。zsock
这是使用 C++ 编写的负载平衡代理的代码:
// Load-balancing broker
// Demonstrates use of the CZMQ API
#include <czmq.h>
#include <cstdio>
#include <thread>
#include <iostream>
#include <vector>
#define NBR_CLIENTS 10
#define NBR_WORKERS 3
#define WORKER_READY "READY" // Signals worker is ready
// Basic request-reply client using REQ socket
//
void client_task()
{
std::cout << "Start client...\n";
zsock_t *client = zsock_new_req("tcp://*:5672");
// zsock_t *client = zsock_new_req("ipc://frontend.ipc");
std::cout << "Client connect...\n";
// zsock_connect(client, "ipc://frontend.ipc");
// Send request, get reply
std::cout << "Client send HELLO\n";
zstr_send(client, "HELLO");
char *reply = zstr_recv(client);
if (reply) {
printf("Client: %s\n", reply);
free(reply);
}
}
// Worker using REQ socket to do load-balancing
//
void worker_task()
{
std::cout << "Start work task...\n";
zsock_t *worker = zsock_new(ZMQ_REQ);
#if (defined (WIN32))
// worker = zsock_new_req("tcp://localhost:5673"); // backend
#else
// worker = zsock_new_req("ipc://backend.ipc");
#endif
std::cout << "Worker connect\n";
zsock_connect(worker, "tcp::/localhost:5673");
// zsock_connect(worker, "ipc://backend.ipc");
// Tell broker we're ready for work
zframe_t *frame = zframe_new(WORKER_READY, strlen(WORKER_READY));
zframe_send(&frame, worker, 0);
// Process messages as they arrive
while (true) {
zmsg_t *msg = zmsg_recv(worker);
if (!msg)
break; // Interrupted
zframe_print(zmsg_last(msg), "Worker: ");
zframe_reset(zmsg_last(msg), "OK", 2);
zmsg_send(&msg, worker);
}
}
// .split main task
// Now we come to the main task. This has the identical functionality to
// the previous {{lbbroker}} broker example, but uses CZMQ to start child
// threads, to hold the list of workers, and to read and send messages:
int main(void)
{
zsock_t *frontend = zsock_new_router("tcp://*:5672");
zsock_t *backend = zsock_new_router ("tcp://*:5673");
// zsock_t *frontend = zsock_new_router("ipc://frontend.ipc");
// zsock_t *backend = zsock_new_router ("ipc://backend.ipc");
// printf("Create server threads");
for (int worker_nbr = 0; worker_nbr < NBR_WORKERS; worker_nbr++) {
std::thread t(worker_task);
std::cout << "Created worker...\n";
// pthread_t worker;
// pthread_create(&worker, NULL, worker_task, (void *)(intptr_t)worker_nbr);
}
int client_nbr;
std::cout << "Create client threads\n";
for (client_nbr = 0; client_nbr < NBR_CLIENTS; client_nbr++) {
std::thread t(client_task);
// pthread_t client;
// printf("Create client thread %d", client_nbr);
// pthread_create(&client, NULL, client_task, (void *)(intptr_t)client_nbr);
}
// Queue of available workers
zlist_t *workers = zlist_new();
// .split main load-balancer loop
// Here is the main loop for the load balancer. It works the same way
// as the previous example, but is a lot shorter because CZMQ gives
// us an API that does more with fewer calls:
while (true) {
zmq_pollitem_t items[] = {
{ backend, 0, ZMQ_POLLIN, 0 },
{ frontend, 0, ZMQ_POLLIN, 0 }
};
std::cout << "Running\n";
// Poll frontend only if we have available workers
int rc = zmq_poll(items, zlist_size(workers) ? 2 : 1, -1);
if (rc == -1)
break; // Interrupted
// Handle worker activity on backend
if (items[0].revents & ZMQ_POLLIN) {
// Use worker identity for load-balancing
zmsg_t *msg = zmsg_recv(backend);
if (!msg)
break; // Interrupted
#if 0
// zmsg_unwrap is DEPRECATED as over-engineered, poor style
zframe_t *identity = zmsg_unwrap(msg);
#else
zframe_t *identity = zmsg_pop(msg);
zframe_t *delimiter = zmsg_pop(msg);
zframe_destroy(&delimiter);
#endif
zlist_append(workers, identity);
// Forward message to client if it's not a READY
zframe_t *frame = zmsg_first(msg);
if (memcmp(zframe_data(frame), WORKER_READY, strlen(WORKER_READY)) == 0) {
zmsg_destroy(&msg);
} else {
zmsg_send(&msg, frontend);
if (--client_nbr == 0)
break; // Exit after N messages
}
}
if (items[1].revents & ZMQ_POLLIN) {
// Get client request, route to first available worker
zmsg_t *msg = zmsg_recv(frontend);
if (msg) {
#if 0
// zmsg_wrap is DEPRECATED as unsafe
zmsg_wrap(msg, (zframe_t *)zlist_pop(workers));
#else
zmsg_pushmem(msg, NULL, 0); // delimiter
zmsg_push(msg, (zframe_t *)zlist_pop(workers));
#endif
zmsg_send(&msg, backend);
}
}
}
// When we're done, clean up properly
while (zlist_size(workers)) {
zframe_t *frame = (zframe_t *)zlist_pop(workers);
zframe_destroy(&frame);
}
zlist_destroy(&workers);
zsock_destroy(&frontend);
zsock_destroy(&backend);
return 0;
}
你可以用g++ lbbroker.cpp -o lbbroker -lzmq -lpthread
. 它编译成功,但是,在运行时,我遇到了一个异常并转储了一个核心:
Created worker...
terminate called without an active exception
Start work task...
aborted (core dumped)
worker_task
它在函数中崩溃了。根据 GDB,似乎成功创建了 1 个线程,然后程序崩溃了。C 版本运行良好。
解决方案
C++ 有析构函数,它们在声明变量的作用域的末尾被调用。
在 的情况下std::thread
,如果使用可连接线程调用析构函数,std::terminate
则将调用。您应该在到达析构函数之前决定是否要join
使用线程。detach
因此,要么分离线程(因此它们将自由运行,无需简单的方法等待它们完成),要么增加对象的生命周期并稍后加入它们。
推荐阅读
- python - 如何在不退出字典的情况下处理 python 中的 KeyError 异常?
- javascript - 如何使用 Vanilla JS 取消静音?
- jquery - 无法在页面加载时加载模式。标准引导方法不起作用。不知道为什么
- javascript - React Hooks 问题:未处理的拒绝(TypeError):无法读取未定义的属性“注册”
- tomcat - 尝试使用 JNDI 获取 OpenMQ 连接工厂时出现 NameNotFoundException
- express - NextJs 与快速会话和护照
- javascript - 状态没有使用 React 钩子更新
- javascript - 数据表不显示数据
- c# - 至1日、2日、3日、4日(月日)
- indy10 - Indy 400 错误请求异常