asio 多线程高性能异步服务器

Posted 2022-06-03 qianbo_insist

asio 多线程理解

理解asio的多线程，一个线程一个iocontext，这样各个socket是在不同的service里面进行的，我们使用第一个线程来接收，用不同的线程来处理客户端

#ifndef c_thread_pool_h
#define c_thread_pool_h

#include <boost/asio.hpp>
#include <memory>
#include <thread>
#include <vector>

class c_thread_pool

public:
   c_thread_pool(int threadNum = std::thread::hardware_concurrency())
       : work_(new boost::asio::io_service::work(service_))
   
       for (int i = 0; i < threadNum; ++i)
       
           threads_.emplace_back([this] ()  service_.run(); );
       
   

   c_thread_pool(const c_thread_pool &) = delete;
   c_thread_pool &operator=(const c_thread_pool &) = delete;

   boost::asio::io_service &getioservice()
   
       return service_;
   

   void stop()
   
       work_.reset();
       for (auto &t: threads_)
       
           t.join();            
               
   
private:
   boost::asio::io_context service_;
   std::unique_ptr<boost::asio::io_service::work> work_;
   std::vector<std::thread> threads_;
;

#endif 

寥寥数句，把线程池勾勒出来

session 和 server

#include "asio_thread_pool.hpp"

#include <boost/asio.hpp>
#include <array>
#include <iostream>

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

class TCPSession : public std::enable_shared_from_this<TCPSession>

public:
	TCPSession(boost::asio::io_service &io_service)
        : socket_(io_service)
            
    
    
    tcp::socket &socket()
    
        return socket_;
    

    void start()
    
        doRead();
    
    
private:
    void doRead()
    
        auto self = shared_from_this();
		//socket_.async_receive()
		//boost::asio::async_read(socket_,
		//	boost::asio::buffer(buffer_, chat_message::header_length),
		//	[this, self](boost::system::error_code ec, std::size_t /*length*/)
		//
		//	if (!ec && read_msg_.decode_header())
		//	
		//		do_read_body();
		//	
		//	else
		//	
		//		room_.leave(shared_from_this());
		//	
		//);
        socket_.async_read_some(boost::asio::buffer(buffer_, 1024),
                                [this, self](boost::system::error_code ec,
                                                          std::size_t bytes_transferred)
                                
                                    if (!ec)
                                    
										buffer_[bytes_transferred] = '\\0';
										std::cout << buffer_;
										doWrite(bytes_transferred);
                                    
                                );        
    

    void doWrite(std::size_t length)
    
		auto self = shared_from_this();
		boost::asio::async_write(socket_, boost::asio::buffer(buffer_, length),
			[this, self](boost::system::error_code ec,
				std::size_t /* bytes_transferred */)
		
			if (!ec)
			
				doRead();
			
		);
    

private:
    tcp::socket socket_;
	char buffer_[1024];
;

class c_echo_server

public:
	c_echo_server(boost::asio::io_service &io_service, unsigned short port, c_thread_pool * pool)
        : io_service_(io_service), pool_(pool),
          acceptor_(io_service, tcp::endpoint(tcp::v4(), port))
    
        doAccept();
    

    void doAccept()
    
		
        auto conn = std::make_shared<TCPSession>(pool_->getIOService());
        acceptor_.async_accept(conn->socket(),
                               [this, conn](boost::system::error_code ec)
                               
                                   if (!ec)
                                   
                                       conn->start();
                                                                      
                                   this->doAccept();
                               );    
    
    
private: 
    boost::asio::io_service &io_service_;
    tcp::acceptor acceptor_;
	c_thread_pool * pool_;
;

int main(int argc, char *argv[])

	c_thread_pool pool(4);

    unsigned short port = 5800;
	c_echo_server server(pool.getIOService(), port,&pool);

    pool.stop();
    
    return 0;

client

使用测试工具先来测试

服务端做回显数据，这里没有做协议，后面再做协议，并且加上并发测试客户端

开发者涨薪指南 48位大咖的思考法则、工作方式、逻辑体系