C++ boost：如何實現每個線程的會話

Question

我想實現一個服務器中每個連接會話都有一個線程。但是，句柄接受回調在同一個線程中調用。我不太熟悉asio中的線程模型。有網站提及它嗎？

編輯：我嘗試寫一個基於boost的例子的聊天室服務器。您可以看到相同的線程用於處理連接請求。所以我不知道如何把它變成分離的線程。

輸出：

[00410190]主線程啓動
[00410860]線程啓動
[00410898]線程啓動
[00410190]開始接受
[00410898]手柄接受
[00410898]開始接受
[00410898]手柄接受
[00410898]開始接受

代碼：

#include "stdafx.h" 
#include <algorithm> 
#include <cstdlib> 
#include <deque> 
#include <iostream> 
#include <list> 
#include <set> 
#include <boost/bind.hpp> 
#include <boost/shared_ptr.hpp> 
#include <boost/enable_shared_from_this.hpp> 
#include <boost/asio.hpp> 
#include "chat_message.h" 
#include <boost/thread.hpp> 
#include <boost/thread/mutex.hpp> 

using boost::asio::ip::tcp; 

//---------------------------------------------------------------------- 

typedef std::deque<chat_message> chat_message_queue; 
boost::mutex global_stream_lock; 
//---------------------------------------------------------------------- 

class chat_participant 
{ 
public: 
    virtual ~chat_participant() {} 
    virtual void deliver(const chat_message& msg) = 0; 
}; 

typedef boost::shared_ptr<chat_participant> chat_participant_ptr; 

//---------------------------------------------------------------------- 

class chat_room 
{ 
public: 
    void join(chat_participant_ptr participant) 
    { 
    participants_.insert(participant); 
    std::for_each(recent_msgs_.begin(), recent_msgs_.end(), 
     boost::bind(&chat_participant::deliver, participant, _1)); 
    } 

    void leave(chat_participant_ptr participant) 
    { 
    participants_.erase(participant); 
    } 

    void deliver(const chat_message& msg) 
    { 
    recent_msgs_.push_back(msg); 
    while (recent_msgs_.size() > max_recent_msgs) 
     recent_msgs_.pop_front(); 

    std::for_each(participants_.begin(), participants_.end(), 
     boost::bind(&chat_participant::deliver, _1, boost::ref(msg))); 
    } 

private: 
    std::set<chat_participant_ptr> participants_; 
    enum { max_recent_msgs = 100 }; 
    chat_message_queue recent_msgs_; 
}; 

//---------------------------------------------------------------------- 

class chat_session 
    : public chat_participant, 
    public boost::enable_shared_from_this<chat_session> 
{ 
public: 
    chat_session(boost::asio::io_service& io_service, chat_room& room) 
    : socket_(io_service), 
     room_(room) 
    { 
    } 

    tcp::socket& socket() 
    { 
    return socket_; 
    } 

    void start() 
    { 
    room_.join(shared_from_this()); 
    boost::asio::async_read(socket_, 
     boost::asio::buffer(read_msg_.data(), chat_message::header_length), 
     boost::bind(
      &chat_session::handle_read_header, shared_from_this(), 
      boost::asio::placeholders::error)); 
    } 

    void deliver(const chat_message& msg) 
    { 
    bool write_in_progress = !write_msgs_.empty(); 
    write_msgs_.push_back(msg); 
    if (!write_in_progress) 
    { 
     boost::asio::async_write(socket_, 
      boost::asio::buffer(write_msgs_.front().data(), 
      write_msgs_.front().length()), 
      boost::bind(&chat_session::handle_write, shared_from_this(), 
      boost::asio::placeholders::error)); 
    } 
    } 

    void handle_read_header(const boost::system::error_code& error) 
    { 
    if (!error && read_msg_.decode_header()) 
    { 
     boost::asio::async_read(socket_, 
      boost::asio::buffer(read_msg_.body(), read_msg_.body_length()), 
      boost::bind(&chat_session::handle_read_body, shared_from_this(), 
      boost::asio::placeholders::error)); 
    } 
    else 
    { 
     room_.leave(shared_from_this()); 
    } 
    } 

    void handle_read_body(const boost::system::error_code& error) 
    { 
    if (!error) 
    { 
     room_.deliver(read_msg_); 
     std::string s(read_msg_.data(), read_msg_.length()); 
     std::cout << s<< std::endl; 
     boost::asio::async_read(socket_, 
      boost::asio::buffer(read_msg_.data(), chat_message::header_length), 
      boost::bind(&chat_session::handle_read_header, shared_from_this(), 
      boost::asio::placeholders::error)); 
    } 
    else 
    { 
     room_.leave(shared_from_this()); 
    } 
    } 

    void handle_write(const boost::system::error_code& error) 
    { 
    if (!error) 
    { 
     write_msgs_.pop_front(); 
     if (!write_msgs_.empty()) 
     { 
     boost::asio::async_write(socket_, 
      boost::asio::buffer(write_msgs_.front().data(), 
       write_msgs_.front().length()), 
      boost::bind(&chat_session::handle_write, shared_from_this(), 
       boost::asio::placeholders::error)); 
     } 
    } 
    else 
    { 
     room_.leave(shared_from_this()); 
    } 
    } 

private: 
    tcp::socket socket_; 
    chat_room& room_; 
    chat_message read_msg_; 
    chat_message_queue write_msgs_; 
}; 

typedef boost::shared_ptr<chat_session> chat_session_ptr; 

//---------------------------------------------------------------------- 

class chat_server 
{ 
public: 
    chat_server(boost::asio::io_service& io_service, 
     const tcp::endpoint& endpoint) 
    : io_service_(io_service), 
     acceptor_(io_service, endpoint) 
    { 
    start_accept(); 
    } 

    void start_accept() 
    { 
     global_stream_lock.lock(); 
     std::cout << "[" << boost::this_thread::get_id() 
       << "] start accept" << std::endl; 
     global_stream_lock.unlock(); 
    chat_session_ptr new_session(new chat_session(io_service_, room_)); 
    acceptor_.async_accept(new_session->socket(), 
     boost::bind(&chat_server::handle_accept, this, new_session, 
      boost::asio::placeholders::error)); 
    } 

    void handle_accept(chat_session_ptr session, 
     const boost::system::error_code& error) 
    { 
     global_stream_lock.lock(); 
     std::cout << "[" << boost::this_thread::get_id() 
       << "] handle accept" << std::endl; 
     global_stream_lock.unlock(); 

    if (!error) 
    { 
     session->start(); 
    } 

    start_accept(); 
    } 

private: 
    boost::asio::io_service& io_service_; 
    tcp::acceptor acceptor_; 
    chat_room room_; 
}; 

typedef boost::shared_ptr<chat_server> chat_server_ptr; 
typedef std::list<chat_server_ptr> chat_server_list; 


void WorkerThread(boost::shared_ptr<boost::asio::io_service> io_service) 
{ 
     global_stream_lock.lock(); 
     std::cout << "[" << boost::this_thread::get_id() 
       << "] Thread Start" << std::endl; 
     global_stream_lock.unlock(); 

     io_service->run(); 

     global_stream_lock.lock(); 
     std::cout << "[" << boost::this_thread::get_id() 
       << "] Thread Finish" << std::endl; 
     global_stream_lock.unlock(); 
} 

//---------------------------------------------------------------------- 

int _tmain(int argc, char* argv[]) 
{ 
    try 
    { 
    if (argc < 2) 
    { 
     std::cerr << "Usage: chat_server <port> [<port> ...]\n"; 
     return 1; 
    } 

    boost::shared_ptr<boost::asio::io_service> io_service( 
       new boost::asio::io_service 
     ); 

    boost::shared_ptr<boost::asio::io_service::work> work(
       new boost::asio::io_service::work(*io_service) 
     ); 

    //boost::asio::io_service::strand strand(*io_service); 

    global_stream_lock.lock(); 
    std::cout << "[" << boost::this_thread::get_id() 
      << "] main Thread Start" << std::endl; 
    global_stream_lock.unlock(); 

    boost::thread_group worker_threads; 
    for(int x = 0; x < 2; ++x) 
    { 
     worker_threads.create_thread(boost::bind(&WorkerThread, 
                io_service)); 
    } 

    chat_server_list servers; 
    for (int i = 1; i < argc; ++i) 
     std::cout<<argv[i] << std::endl; 

    for (int i = 1; i < argc; ++i) 
    { 
     using namespace std; // For atoi. 
     tcp::endpoint endpoint(tcp::v4(), atoi(argv[i])); 
     chat_server_ptr server(new chat_server(*io_service, endpoint)); 
     servers.push_back(server); 
    } 

    work.reset(); 

    worker_threads.join_all(); 
    } 
    catch (std::exception& e) 
    { 
    std::cerr << "Exception: " << e.what() << "\n"; 
    } 

    return 0; 
} 

Answer 1

This boost asio example顯示了使用單一io_service對象和線程池來處理傳入會話的HTTP服務器。它可能有幫助。