我正在使用Nanomsg在使用C ++的系统中进行IPC。我想创建一个后台线程来处理发送和接收消息。我使用对范例，并使用nn_poll检查套接字fd是否可写或可读，如果可读然后读；如果可写，则从消息队列中弹出一项并发送。我的问题是，由于nn_poll循环中没有睡眠，因此我创建的backgroudn线程会占用大量CPU使用率，是否有任何方法可以减少CPU使用率，但仍会像没有睡眠那样产生延迟？以下是我的示例代码。谢谢。
Server.cpp
#include <iostream>
#include <thread>
#include <string>
#include <queue>
#include <utility>
#include <mutex>
#include <nanomsg/pair.h>
#include <nanomsg/nn.h>
class Nanomsg {
private:
bool _server;
bool _stop;
int _sock;
std::string _url;
std::thread _th;
std::queue<std::string> _queue;
std::mutex _queueMutex;
void _start() {
_sock = nn_socket(AF_SP, NN_PAIR);
if (_sock < 0) {
std::cout << "failed to create socket" << std::endl;
return;
}
int rc = 0;
if (_server) {
rc = nn_bind(_sock, _url.c_str());
} else {
rc = nn_connect(_sock, _url.c_str());
}
if (rc < 0) {
std::cout << "failed to connect/bind socket" << std::endl;
return;
}
struct nn_pollfd pfd{};
pfd.fd = _sock;
pfd.events = NN_POLLIN | NN_POLLOUT;
while (!_stop) {
std::cout << "ssasd" << std::endl;
rc = nn_poll(&pfd, 1, 2000);
if (rc == 0) {
std::cout << "timeout" << std::endl;
continue;
}
if (rc == -1) {
std::cout << "error!" << std::endl;
return;
}
if (pfd.revents & NN_POLLIN) {
char *buf = nullptr;
int rbs = nn_recv(_sock, &buf, NN_MSG, 0);
if (rbs < 0) {
continue;
}
std::string r(buf, rbs);
std::cout << "received [" << r << "]" << std::endl;
nn_freemsg(buf);
}
if (pfd.revents & NN_POLLOUT) {
std::cout << "asd" << std::endl;
if (_queue.empty()) {
continue;
}
{
std::lock_guard<std::mutex> lock(_queueMutex);
auto msg = _queue.front();
std::cout << "send [" << msg << "]" << std::endl;
rc = nn_send(_sock, msg.c_str(), msg.length(), 0);
if (rc >= 0) {
_queue.pop();
}
}
}
}
}
public:
Nanomsg() : _sock(0), _server(false), _stop(false), _url("ipc:///tmp/test.ipc") {
}
Nanomsg(std::string url, bool server) : _url(std::move(url)), _sock(0), _server(server), _stop(false) {
}
void start() {
_th = std::thread([=]() {
_start();
});
}
void stop() {
_stop = true;
if (_th.joinable()) {
_th.join();
}
}
void send(const std::string& msg) {
{
std::lock_guard<std::mutex> lock(_queueMutex);
_queue.push(msg);
}
}
};
int main() {
Nanomsg server("ipc:///tmp/test.ipc", true);
server.start();
while (true) {
server.send("test");
std::this_thread::sleep_for(std::chrono::seconds(3));
}
return 0;
}
Client.cpp
#include <iostream>
#include <thread>
#include <string>
#include <queue>
#include <utility>
#include <mutex>
#include <nanomsg/pair.h>
#include <nanomsg/nn.h>
struct nn_pollf {
int fd;
short events;
short revents;
};
class Nanomsg {
private:
bool _server;
bool _stop;
int _sock;
std::string _url;
std::thread _th;
std::queue<std::string> _queue;
std::mutex _queueMutex;
void _start() {
_sock = nn_socket(AF_SP, NN_PAIR);
if (_sock < 0) {
std::cout << "failed to create socket" << std::endl;
return;
}
int rc = 0;
if (_server) {
rc = nn_bind(_sock, _url.c_str());
} else {
rc = nn_connect(_sock, _url.c_str());
}
if (rc < 0) {
std::cout << "failed to connect/bind socket" << std::endl;
return;
}
struct nn_pollfd pfd{};
pfd.fd = _sock;
pfd.events = NN_POLLIN | NN_POLLOUT;
while (!_stop) {
std::cout << "ssasd" << std::endl;
rc = nn_poll(&pfd, 1, 2000);
if (rc == 0) {
std::cout << "timeout" << std::endl;
continue;
}
if (rc == -1) {
std::cout << "error!" << std::endl;
return;
}
if (pfd.revents & NN_POLLIN) {
char *buf = nullptr;
int rbs = nn_recv(_sock, &buf, NN_MSG, 0);
if (rbs < 0) {
continue;
}
std::string r(buf, rbs);
std::cout << "received [" << r << "]" << std::endl;
nn_freemsg(buf);
}
if (pfd.revents & NN_POLLOUT) {
std::cout << "asd" << std::endl;
if (_queue.empty()) {
continue;
}
{
std::lock_guard<std::mutex> lock(_queueMutex);
auto msg = _queue.front();
std::cout << "send [" << msg << "]" << std::endl;
rc = nn_send(_sock, msg.c_str(), msg.length(), 0);
if (rc >= 0) {
_queue.pop();
}
}
}
}
}
public:
Nanomsg() : _sock(0), _server(false), _stop(false), _url("ipc:///tmp/test.ipc") {
}
Nanomsg(std::string url, bool server) : _url(std::move(url)), _sock(0), _server(server), _stop(false) {
}
void start() {
_start();
// _th = std::thread([=]() {
// _start();
// });
}
void stop() {
_stop = true;
if (_th.joinable()) {
_th.join();
}
}
void send(const std::string& msg) {
{
std::lock_guard<std::mutex> lock(_queueMutex);
_queue.push(msg);
}
}
};
int main() {
Nanomsg client("ipc:///tmp/test.ipc", false);
client.start();
return 0;
}
最佳答案
如果没有什么可发送的，也没有任何接收的，请让您的线程进入睡眠状态
一毫秒。您在当前设计中几乎唯一可以做的事情。
如果可能，您可以使用nanomsg next generation (nng)并为其提供一个asynchronous interface机会。似乎您仍然自己在实现异步接口，那么为什么不也使用nanomsg？它们具有您操作系统的网络API的所有功能，因此应该能够提供最佳延迟，而不会浪费CPU时间。
创建一个异步I / O句柄，并使用nng_aio_alloc(3)设置回调。呼叫nng_recv_aio(3)以在收到数据时得到通知。不要自己管
发送队列，请改用nng_send_aio(3)中的void Nanomsg::send()。
不幸的是nng是一个单独的库，您正在使用经典的nanomsg。我注意到只有写作。
关于c++ - 具有多线程应用程序的Nanomsg无阻塞双向套接字，我们在Stack Overflow上找到一个类似的问题：https://stackoverflow.com/questions/57834618/