boost asio非同步讀寫網路聊天程式客戶端 例項詳解
阿新 • • 發佈:2019-01-10
// // chat_server.cpp // ~~~~~~~~~~~~~~~ // // Copyright (c) 2003-2013 Christopher M. Kohlhoff (chris at kohlhoff dot com) // // Distributed under the Boost Software License, Version 1.0. (See accompanying // file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt) // //伺服器端程式,首先開啟一個chat_server,用來初始話io_service和chat_room; 伺服器每次只開啟一個chat_room,chat_room的作用是 1//.儲存使用者資訊 2.實現使用者的join和leave 3.儲存從每個客戶端接收到的資訊 4.將接收到的訊息掛到每一個客戶端write_msgs隊尾 //每連線上一個客戶端(socket)就開啟一個新的chat_session 並加入唯一的chat_room chat_session中實現分別對每一個客戶端的非同步讀寫//操作 #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.hpp" using boost::asio::ip::tcp; //---------------------------------------------------------------------- typedef std::deque<chat_message> chat_message_queue; //---------------------------------------------------------------------- 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); } //將從某個客戶端收到的訊息掛到 每一個客戶端的write_msgs隊尾 具體見chat_participant::deliver void deliver(const chat_message& msg) { recent_msgs_.push_back(msg); while (recent_msgs_.size() > max_recent_msgs) recent_msgs_.pop_front(); //room中儲存的訊息數有限 std::for_each(participants_.begin(), participants_.end(), boost::bind(&chat_participant::deliver, _1, boost::ref(msg))); } private: std::set<chat_participant_ptr> participants_;//用set來儲存使用者資訊 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()//每生成一個新的chat_session都會呼叫 { 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); //把room中儲存的訊息掛到write_msgs隊尾 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_); 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) //全域性只有一個io_service和一個acceptor { start_accept(); } void start_accept() { 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) { if (!error) { session->start(); } start_accept(); //每連線上一個socket都會呼叫 } private: boost::asio::io_service& io_service_; tcp::acceptor acceptor_; chat_room room_; //chat_room中沒有過載建構函式 所以會直接呼叫預設建構函式 }; typedef boost::shared_ptr<chat_server> chat_server_ptr; typedef std::list<chat_server_ptr> chat_server_list; //---------------------------------------------------------------------- int main(int argc, char* argv[]) { try { if (argc < 2) { std::cerr << "Usage: chat_server <port> [<port> ...]\n"; return 1; } boost::asio::io_service io_service; chat_server_list servers; 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); } io_service.run(); } catch (std::exception& e) { std::cerr << "Exception: " << e.what() << "\n"; } return 0; }