c++ - 如何使用 boost::thread_group 使用多客户端服务器？

Question

我想使用 boost::asio 和 boost::thread_group 创建多客户端服务器。
我试过这个

//Listener.h 
#pragma once
#include <boost/asio/ip/tcp.hpp>
#include <boost/bind.hpp>
#include <memory>
#include <boost/asio/io_context.hpp>
#include <boost/asio/placeholders.hpp>
#include <boost/thread/thread.hpp>
#include "Session.h"
#define PORT_NUMBER 12880

class Listener
{
public:
    Listener(boost::asio::io_context& iosvc, boost::asio::ip::tcp::endpoint& ep);

    void Start_accept();


private:
    boost::asio::io_context &_iosvc;
    boost::asio::ip::tcp::acceptor _acceptor;
    boost::thread_group threadPool;
    void OnAcceptComplete(const boost::system::error_code& error, Session* session);
};

这是 Listener.cpp 文件

//Listener.cpp
#pragma once
#include <iostream>
#include "Listener.h"
#include "SessionManager.h"

Listener::Listener(boost::asio::io_context& iosvc, boost::asio::ip::tcp::endpoint& ep) :
    _acceptor(iosvc, ep), _iosvc(iosvc) {}


void Listener::Start_accept()
{
    //Create Client Session
    Session* new_session = clientsession::SessionManager::GetInstance()->GenerateSession(_iosvc);
    std::cout << "Listener thread ID : " << std::this_thread::get_id << '\n';
    std::cout << "Listening....." << "\n\n";
    _acceptor.async_accept(*(new_session->Socket()),
        boost::bind(&Listener::OnAcceptComplete, this, boost::asio::placeholders::error, new_session));
}
void Listener::OnAcceptComplete(const boost::system::error_code& error, Session* session)
{
    if (!error) 
    {
        std::cout << "accept completed" << '\n';

        threadPool.create_thread(boost::bind(&boost::asio::io_context::run, &(session->_iocontext)));
        session->_iocontext.post(boost::bind(&Session::StartSession,this)); // error '->*': Cannot convert from'T*' to'Session *'.
        
        threadPool.join_all();
    }

    else
    {
        std::cout << error.message() << '\n';
    }
    Start_accept();
}

但是 session->_iocontext.post(boost::bind(&Session::StartSession,this));
//错误 '->': 无法从 'T' 转换为 'Session *'。
Server的主要语句如下。

//Server.cpp
#include <iostream>
#include <boost/thread/thread.hpp>
#include "Listener.h"
#include "SessionManager.h"
using namespace boost::asio;
using namespace clientsession;
SessionManager* _sessionManager;

int main()
{
    try
    {
        boost::asio::io_context iocontext;
        ip::tcp::endpoint ep(ip::address::from_string("127.0.0.1"), PORT_NUMBER); //ipv4, 포트번호
        Listener* _listener = new Listener(iocontext, ep);

        _listener->Start_accept();
        iocontext.run();

    }
    catch (std::exception& e)
    {
        std::cout << e.what() << '\n';
    }
    while (true)
    {
    }
    return 0;

}

我想制作多线程服务器 [一个 io_context 监听器(包括接受)，多个 io_context session (用于客户端)]
换句话说
我想更改为每个客户端创建 session 线程的结构。
我不知道我创建的代码是否正确。
可以进行 N:1 通信，但 session 处理是同步进行的。所以我想让一个线程成为一个 session 并异步处理它。

Answer 1

In other words I want to change the structure in which session threads are created for each client.

这是 ASIO 中的反模式。你 可以 ，但请阅读: https://www.boost.org/doc/libs/1_75_0/doc/html/boost_asio/overview/core/threads.html
在 Asio 中拥有单独线程的一种自然方式是使用 io_context每个线程也是如此。但是，更自然的方法是每个客户端都有一个链(“逻辑线程”)和一个反射(reflect)系统上真实内核数的固定线程池。

I don't know if the code I created is correct.

实际上你这样做，因为它不编译。

    session->_iocontext.post(boost::bind(&Session::StartSession,this)); // error '->*': Cannot convert from'T*' to'Session *'.

this是 Listener*这里。当然不能绑定(bind) Session的方法到 Listener 的一个实例类(class)。您想要的是 session :

post(session->_iocontext, boost::bind(&Session::StartSession, session));

接下来:

threadPool.join_all();出现在 OnAcceptedComplete这意味着您甚至在考虑接受任何其他连接之前都会等待所有线程。这意味着您实际上最好使用单线程阻塞服务器。

您需要调用 std::this_thread::get_id : 添加括号 std::this_thread::get_id() (让我想起了node addon with cluster get process id returns same id for all forked process's)

你所有的原始指针和new没有 delete将导致内存泄漏或过时指针的噩梦。

为什么监听器不在堆栈上？

  Listener _listener(iocontext, ep);
  _listener.Start_accept();

为什么会Socket()不返回引用？现在你只像这样使用它

  tcp::socket _socket;
  auto* Socket() { return &_socket; }
  // ...
  _acceptor.async_accept(*(new_session->Socket()),

钱币。更简单:

  auto& Socket() { return _socket; }
  // ....
  _acceptor.async_accept(new_session->Socket(),

等等

您发布 StartSession进入 io 服务:

 post(/*some service*/, boost::bind(&Session::StartSession, session));

确实，这仅适用于该线程 的私有(private)服务。或 使用一条线。使用 strands 似乎仍然是多余的，因为这总是发生在新接受的 session 上，所以没有其他线程可能知道它。

The exception is when you rely on thread local storage?

Or maybe your implementation of GenerateSession does not result unique sessions?

最少的修复
这是我的最小修复。我建议不要使用多个线程(无论如何你都在做异步 IO)。

#define PORT_NUMBER 12880
#include <boost/asio.hpp>
#include <boost/bind/bind.hpp>
#include <iomanip>
#include <iostream>
#include <memory>
using boost::asio::ip::tcp;
using boost::system::error_code;

struct Session : std::enable_shared_from_this<Session> {
    template <typename Executor>
    Session(Executor executor) : _socket(make_strand(executor)) {}

    auto& Socket() { return _socket; }

    void StartSession() {
        std::cout << "Session: " << this << " StartSession" << std::endl;
        async_read(_socket, boost::asio::dynamic_buffer(received),
           [self = shared_from_this()](auto ec, size_t tx) { self->on_received(ec, tx); });
    };

  private:
    tcp::socket _socket;
    std::string received;
    void on_received(error_code ec, size_t /*bytes_transferred*/) {
        std::cout << "Session: " << this << " on_received " << ec.message() << " " << std::quoted(received) << std::endl;
    }
};

using SessionPtr = std::shared_ptr<Session>;

namespace clientsession {
    struct SessionManager {
        static SessionManager* GetInstance() {
            static SessionManager instance;
            return &instance;
        }
        template <typename... T> SessionPtr GenerateSession(T&&... args) {
            return std::make_shared<Session>(std::forward<T>(args)...);
        }
    };
} // namespace clientsession

class Listener {
    public:
        Listener(boost::asio::io_context& iosvc, tcp::endpoint ep)
            : _iosvc(iosvc), _acceptor(iosvc, ep) {}

        void Start_accept() {
            // Create Client Session
            auto new_session = clientsession::SessionManager::GetInstance()
                ->GenerateSession(_iosvc.get_executor());

            std::cout << "Listening.....\n\n";
            _acceptor.async_accept(new_session->Socket(),
                    boost::bind(&Listener::OnAcceptComplete, this,
                        boost::asio::placeholders::error,
                        new_session));
        }

    private:
        boost::asio::io_context& _iosvc;
        tcp::acceptor _acceptor;
        void OnAcceptComplete(error_code error, SessionPtr session) {
            if (!error) {
                std::cout << "accept completed\n";
                session->StartSession();
                Start_accept();
            } else {
                std::cout << error.message() << '\n';
            }
        }
};

int main() {
    try {
        boost::asio::io_context iocontext;
        Listener listener(iocontext, {{}, PORT_NUMBER}); // ipv4, port number
        listener.Start_accept();

        iocontext.run();
    } catch (std::exception& e) {
        std::cout << e.what() << '\n';
    }
}

当与一堆同时的客户端进行测试时:

for a in {1..10};
do
    (sleep 1.$RANDOM; echo -n "hellow world $RANDOM") |
         netcat -w 2 localhost 12880&
done; 
time wait

打印出类似的东西

Listening.....

accept completed
Session: 0x56093453c1b0 StartSession
Listening.....

accept completed
Session: 0x56093453de60 StartSession
Listening.....

accept completed
Session: 0x56093453e3c0 StartSession
Listening.....

accept completed
Session: 0x56093453e920 StartSession
Listening.....

accept completed
Session: 0x56093453ee80 StartSession
Listening.....

accept completed
Session: 0x56093453f3e0 StartSession
Listening.....

accept completed
Session: 0x56093453f940 StartSession
Listening.....

accept completed
Session: 0x56093453fea0 StartSession
Listening.....

accept completed
Session: 0x560934540400 StartSession
Listening.....

accept completed
Session: 0x560934540960 StartSession
Listening.....

Session: 0x56093453f940 on_received End of file "hellow world 10149"
Session: 0x56093453fea0 on_received End of file "hellow world 22492"
Session: 0x560934540400 on_received End of file "hellow world 29539"
Session: 0x56093453c1b0 on_received End of file "hellow world 20494"
Session: 0x56093453ee80 on_received End of file "hellow world 24735"
Session: 0x56093453de60 on_received End of file "hellow world 8071"
Session: 0x560934540960 on_received End of file "hellow world 27606"
Session: 0x56093453e920 on_received End of file "hellow world 534"
Session: 0x56093453e3c0 on_received End of file "hellow world 21676"
Session: 0x56093453f3e0 on_received End of file "hellow world 24362"