boost asio 异步io

布满荆棘的人生 2022-06-05 04:40 174阅读 0赞

boost::asio::io_service bIOService;//定义一个io_service对象
    	boost::asio::deadline_timer timer(bIOService,boost::posix_time::seconds(5));//定义一个deadline_timer对象,绑定io_service服务 5秒后执行
    	timer.async_wait(boost_handler);//5秒后处理的函数
    	bIOService.run();//io_service运行

void boost_handler(const boost::system::error_code& ec){
    	TRACE("5 s.\n");
    }
    
    void boost_handler2(const boost::system::error_code& ec){
    	TRACE("10 s.\n");
    }
    
    void CMFC08Dlg::OnBnClickedButton2()
    {
    	TRACE("当前线程ID,%d\n",GetCurrentThreadId());
    	boost::asio::io_service bIOService;//定义一个io_service对象
    	boost::asio::deadline_timer timer(bIOService,boost::posix_time::seconds(5));//定义一个deadline_timer对象,绑定io_service服务 5秒后执行
    	timer.async_wait(boost_handler);//5秒后处理的函数
    
    	boost::asio::deadline_timer timer2(bIOService,boost::posix_time::seconds(10));//定义一个deadline_timer对象,绑定io_service服务 10秒后执行
    	timer2.async_wait(boost_handler2);
    
    	bIOService.run();//io_service运行,这里会等待boost_handler 和 boost_handler2 调用完毕后再返回
    }

//由于有两个线程，所以 handler1() 和 handler2() 可以同时执行
    //1.如果boost_handler2触发时,boost_handler仍在执行，则boost_handler2就会在第二个线程中执行。 
    //2.如果boost_handler已经终止，则 I/O 服务可以自由选择任一线程执行boost_handler2。
    void boost_handler(const boost::system::error_code& ec){
    	TRACE("5 s.\n");
    }
    
    void boost_handler2(const boost::system::error_code& ec){
    	TRACE("5 s.\n");
    }
    
    boost::asio::io_service bIOService;//定义一个io_service对象
    
    void run(){
    	bIOService.run();
    }
    
    void CMFC08Dlg::OnBnClickedButton2()
    {
    	TRACE("当前线程ID,%d\n",GetCurrentThreadId());
    	
    	boost::asio::deadline_timer timer(bIOService,boost::posix_time::seconds(5));//定义一个deadline_timer对象,绑定io_service服务 5秒后执行
    	timer.async_wait(boost_handler);//5秒后处理的函数
    
    	boost::asio::deadline_timer timer2(bIOService,boost::posix_time::seconds(5));//定义一个deadline_timer对象,绑定io_service服务 10秒后执行
    	timer2.async_wait(boost_handler2);
    
    	boost::thread t1(run);
    	boost::thread t2(run);
    
    	t1.join();
    	t2.join();
    }

//每个线程处理不同的io_service对象
    
    void boost_handler(const boost::system::error_code& ec){
    	TRACE("5 s.\n");
    }
    
    void boost_handler2(const boost::system::error_code& ec){
    	TRACE("5 s.\n");
    }
    
    boost::asio::io_service bIOService_1;//定义一个io_service对象
    boost::asio::io_service bIOService_2;//定义一个io_service对象
    
    void run_1(){
    	bIOService_1.run();
    }
    
    void run_2(){
    	bIOService_2.run();
    }
    
    void CMFC08Dlg::OnBnClickedButton2()
    {
    	TRACE("当前线程ID,%d\n",GetCurrentThreadId());
    	
    	boost::asio::deadline_timer timer(bIOService_1,boost::posix_time::seconds(5));//定义一个deadline_timer对象,绑定io_service服务 5秒后执行
    	timer.async_wait(boost_handler);//5秒后处理的函数
    
    	boost::asio::deadline_timer timer2(bIOService_2,boost::posix_time::seconds(5));//定义一个deadline_timer对象,绑定io_service服务 10秒后执行
    	timer2.async_wait(boost_handler2);
    
    	boost::thread t1(run_1);
    	boost::thread t2(run_2);
    
    	t1.join();
    	t2.join();
    }

//异步访问网站
    boost::asio::io_service bIOService;
    boost::asio::ip::tcp::resolver bResolver(bIOService);
    boost::asio::ip::tcp::socket bSock(bIOService);
    boost::array<char,4096> buffer;
    
    void read_handler(const boost::system::error_code& ec,std::size_t bytes_transferred){
    	//数据读取完毕的处理器
    	if(0!=ec){
    		TRACE("无法读取数据,%s\n",ec.message());
    		return;
    	}
    	string szData(buffer.data(),bytes_transferred);
    	TRACE("读取的数据,%s",szData.c_str());
    	bSock.async_read_some(boost::asio::buffer(buffer),read_handler);//继续又读取
    }
    
    void connect_handler(const boost::system::error_code & ec){
    	//连接完成后调用的处理器
    	if(0!=ec){
    		TRACE("无法连接,%s\n",ec.message());
    		return;
    	}
    	boost::asio::write(bSock,boost::asio::buffer("GET / HTTP 1.1\r\nHost: www.hao123.com\r\n\r\n"));//执行一个发送数据的操作
    	bSock.async_read_some(boost::asio::buffer(buffer),read_handler);//tcp套接字绑定异步读取数据操作,绑定异步读取处理器函数
    }
    
    void resolve_handler(const boost::system::error_code& ec,boost::asio::ip::tcp::resolver::iterator it){
    	//解决方案初始化完毕后的处理器
    	if(0!=ec){
    		TRACE("无法连接,%s\n",ec.message());
    		return;
    	}
    	bSock.async_connect(*it, connect_handler); //tcp套接字对象异步连接绑定,tcp解决方案对象,绑定连接处理器
    }
    
    void CMFC08Dlg::OnBnClickedButton2()
    {
    	TRACE("当前线程ID,%d\n",GetCurrentThreadId());
    	boost::asio::ip::tcp::resolver::query bQuery("www.hao123.com","80");//构造一个query 对象
    	bResolver.async_resolve(bQuery,resolve_handler);//Resolver 是一个tcp 解决方案对象,绑定query对象,并且绑定解决处理器,主要用于解析域名为真实的连接IP对象
    	bIOService.run();//服务运行
    }

//异步IO服务器
    boost::asio::io_service b_io_service; 
    boost::asio::ip::tcp::endpoint b_endpoint(boost::asio::ip::tcp::v4(), 8080);//终端端口 80,注意端口不能给占有,这里用全局变量会有问题的,最好加上try
    boost::asio::ip::tcp::acceptor b_acceptor(b_io_service, b_endpoint); //接收器对象,绑定io_service,绑定终端端口
    boost::asio::ip::tcp::socket b_sock(b_io_service); //套接字,绑定io_service
    std::string data = "HTTP/1.1 200 OK\r\nContent-Length: 13\r\n\r\nHello, world!"; //反馈数据
    
    void write_handler(const boost::system::error_code &ec, std::size_t bytes_transferred) 
    { 
    	//发送完毕的处理器
    } 
    
    void accept_handler(const boost::system::error_code &ec) 
    { 
    	//接收新的连接处理器
    	if (0!=ec) 
    	{ 
    		TRACE("接收处理器,%s\n",ec.message());
    		return ;
    	} 
    	boost::asio::async_write(b_sock, boost::asio::buffer(data), write_handler);
    	//b_acceptor.async_accept(*(new boost::asio::ip::tcp::socket(b_io_service)), accept_handler); //可以再开一个允许套接字连接,但是async_write要发送对应的新一个套接字连接才行
    } 
    
    void CMFC08Dlg::OnBnClickedButton2()
    {
    	TRACE("当前线程ID,%d\n",GetCurrentThreadId());
    	b_acceptor.listen(); //监听
    	b_acceptor.async_accept(b_sock, accept_handler); //开始接受连接
    	b_io_service.run();//io_service 运行
    }

//Boost.Asio扩展
    template <typename Service>
    class basic_timer:
    	public boost::asio::basic_io_object<Service>
    {
    public:
    	explicit basic_timer(boost::asio::io_service& bIOService)
    		:boost::asio::basic_io_object<Service>(bIOService)
    	{
    
    	}
    
    	void wait(std::size_t seconds){
    		//调用的时候转接到service对象的类中调用,换句话呢说也就是Service的模板类
    		return this->service.wait(this->implementation,seconds);
    	}
    
    	template <typename Handler>
    	void async_wait(std::size_t seconds,Handler handler){
    		//调用的时候转接到service对象的类中调用,换句话呢说也就是Service的模板类
    		this->service.async_wait(this->implementation,seconds,handler);
    	}
    };
    //时间控制器接入对象
    class timer_impl
    {
    public:
    	timer_impl()
    		:handle_(CreateEvent(NULL,FALSE,FALSE,NULL)){
    
    	}
    	~timer_impl(){
    		CloseHandle(handle_);
    	}
    	void destroy(){
    		SetEvent(handle_);
    	}
    
    	void wait(std::size_t seconds,boost::system::error_code& ec){
    		DWORD res=WaitForSingleObject(handle_,seconds*1000);
    		if(WAIT_OBJECT_0 == res){
    			ec=boost::asio::error::operation_aborted;
    		}else
    			ec=boost::system::error_code(); 
    	}
    protected:
    private:
    	HANDLE handle_;//事件句柄
    };
    template <typename TimerImplementation = timer_impl>
    class basic_timer_service
    	:public boost::asio::io_service::service{
    public :
    	static boost::asio::io_service::id id;
    	explicit basic_timer_service(boost::asio::io_service& io_service)
    		:boost::asio::io_service::service(io_service),
    		async_work_(new boost::asio::io_service::work(async_io_service_)),//boost::scoped_ptr<boost::asio::io_service::work> async_work_,作用域指针初始化
    		async_thread_(boost::bind(&boost::asio::io_service::run, &async_io_service_))//boost::thread async_thread_,通过boost::bind,线程初始化运行地址以及参数
    	{
    
    	}
    	~basic_timer_service(){
    		async_work_.reset();//释放资源
    		async_io_service_.stop();//服务终止
    		async_thread_.join();//线程停止,个人觉得应该线程应该先终止再释放,async_work_.reset();,async_io_service_.stop(),否则线程在运行时刻,这里资源释放了则会造成影响
    	}
    	//定义共享指针,implementation_type
    	typedef boost::shared_ptr<TimerImplementation> implementation_type;
    
    	void construct(implementation_type &impl){
    		impl.reset(new TimerImplementation());//共享指针资源初始化
    	}
    
    	void destroy(implementation_type& impl){
    		impl->destroy();//调用共享指针对象的destroy函数,释放其资源(这里有一个疑问,指针析构的时候可以由于对象自己来析构)
    		impl.reset();//因为这里调用了reset函数
    	}
    
    	void wait(implementation_type &impl,std::size_t seconds){
    		boost::system::error_code ec;
    		impl->wait(seconds,ec);//调用共享指针对象的等待方法
    		boost::asio::detail::throw_error(ec);//最后还抛出异常?
    	}
    
    	template <typename Handler>
    	class wait_operation{
    	public:
    		wait_operation(implementation_type &impl,boost::asio::io_service& io_service,std::size_t seconds,Handler handler)
    			:impl_(impl),
    			  io_service_(io_service),
    			  work_(io_service),
    			  seconds_(seconds),
    			  handler_(handler)
    		{
    
    		}
    		  void operator()() const {
    			  implementation_type impl=impl_.lock();//尝试锁定获取共享对象
    			  if(impl){
    				  boost::system::error_code ec;
    				  impl->wait(seconds_,ec);//等待指定的时间
    				  this->io_service_.post(boost::asio::detail::bind_handler(handler_,ec));//post一个异步处理函数,然后io_service在run的时候才会执行handler(wait_handler)
    			  }else{
    				  this->io_service_.post(boost::asio::detail::bind_handler(handler_,boost::asio::error::operation_aborted));//post一个异步处理函数,包含错误信息,,然后io_service在run的时候才会执行handler(wait_handler)
    			  }
    		  }
    	private:
    		boost::weak_ptr<TimerImplementation> impl_;//弱指针对象
    		boost::asio::io_service &io_service_;//运行的io_service对象
    		boost::asio::io_service::work work_;//io_service工作对象
    		std::size_t seconds_;//等待时间
    		Handler handler_;//处理器函数
    	};
    
    	template <typename Handler>
    	void async_wait(implementation_type& impl,std::size_t seconds,Handler handler ){
    		//异步调用时候
    		this->async_io_service_.post(wait_operation<Handler>(impl,this->get_io_service(),seconds,handler));
    	}
    private:
    	void shutdown_service(){
    
    	}
    	boost::asio::io_service async_io_service_; //io_service 对象,内部的一个io_service对象
    	boost::scoped_ptr<boost::asio::io_service::work> async_work_; //服务工作对象
    	boost::thread async_thread_; //线程对象
    };
    
    template <typename TimerImplementation> 
    boost::asio::io_service::id basic_timer_service<TimerImplementation>::id; 
    
    void wait_handler(const boost::system::error_code &ec) 
    { 
    	TRACE("5 s后调用\n");
    } 
    
    typedef basic_timer<basic_timer_service<> > timer;
    
    void CMFC08Dlg::OnBnClickedButton2()
    {
    	boost::asio::io_service io_service; //创建一个io_service基类
    	timer t(io_service); //构造timer(basic_timer_service) 对象,绑定io_service对象
    	t.async_wait(5, wait_handler); //异步调用,绑定wait_handler处理函数,这里会创建新的线程去等待5秒
    	io_service.run(); //服务对象运行
    }

**运行流程图**

![20171119195303975][]

![20171119195406666][]

[20171119195303975]: /images/20220605/3c4ed312aceb470a934ed02ab7edd4e4.png
[20171119195406666]: /images/20220605/8f36a07004774372a6c4d8db2331b52d.png