llfc项目TCP服务器笔记

ChatServer

一个TCP服务器必然会有连接的接收，维持，收发数据等逻辑。那我们就要基于asio完成这个服务的搭建。主服务是这个样子的

1. #include "LogicSystem.h"
2. #include <csignal>
3. #include <thread>
4. #include <mutex>
5. #include "AsioIOServicePool.h"
6. #include "CServer.h"
7. #include "ConfigMgr.h"
8. using namespace std;
9. bool bstop = false;
10. std::condition_variable cond_quit;
11. std::mutex mutex_quit;
13. int main()
14. {
15. try {
16. auto &cfg = ConfigMgr::Inst();
17. auto pool = AsioIOServicePool::GetInstance();
18. boost::asio::io_context io_context;
19. boost::asio::signal_set signals(io_context, SIGINT, SIGTERM);
20. signals.async_wait([&io_context, pool](auto, auto) {
21. io_context.stop();
22. pool->Stop();
23. });
24. auto port_str = cfg["SelfServer"]["Port"];
25. CServer s(io_context, atoi(port_str.c_str()));
26. io_context.run();
27. }
28. catch (std::exception& e) {
29. std::cerr << "Exception: " << e.what() << endl;
30. }
32. }

CServer类的声明

1. #include <boost/asio.hpp>
2. #include "CSession.h"
3. #include <memory.h>
4. #include <map>
5. #include <mutex>
6. using namespace std;
7. using boost::asio::ip::tcp;
8. class CServer
9. {
10. public:
11. CServer(boost::asio::io_context& io_context, short port);
12. ~CServer();
13. void ClearSession(std::string);
14. private:
15. void HandleAccept(shared_ptr<CSession>, const boost::system::error_code & error);
16. void StartAccept();
17. boost::asio::io_context &_io_context;
18. short _port;
19. tcp::acceptor _acceptor;
20. std::map<std::string, shared_ptr<CSession>> _sessions;
21. std::mutex _mutex;
22. };

构造函数中监听对方连接

1. CServer::CServer(boost::asio::io_context& io_context, short port):_io_context(io_context), _port(port),
2. _acceptor(io_context, tcp::endpoint(tcp::v4(),port))
3. {
4. cout << "Server start success, listen on port : " << _port << endl;
5. StartAccept();
6. }

接受连接的函数

1. void CServer::StartAccept() {
2. auto &io_context = AsioIOServicePool::GetInstance()->GetIOService();
3. shared_ptr<CSession> new_session = make_shared<CSession>(io_context, this);
4. _acceptor.async_accept(new_session->GetSocket(), std::bind(&CServer::HandleAccept, this, new_session, placeholders::_1));
5. }

AsioIOServicePool

从AsioIOServicePool中返回一个可用的iocontext构造Session，然后将接受的新链接的socket写入这个Session保管。

AsioIOServicePool已经在前面讲解很多次了，它的声明如下

1. #include <vector>
2. #include <boost/asio.hpp>
3. #include "Singleton.h"
4. class AsioIOServicePool:public Singleton<AsioIOServicePool>
5. {
6. friend Singleton<AsioIOServicePool>;
7. public:
8. using IOService = boost::asio::io_context;
9. using Work = boost::asio::io_context::work;
10. using WorkPtr = std::unique_ptr<Work>;
11. ~AsioIOServicePool();
12. AsioIOServicePool(const AsioIOServicePool&) = delete;
13. AsioIOServicePool& operator=(const AsioIOServicePool&) = delete;
14. // 使用 round-robin 的方式返回一个 io_service
15. boost::asio::io_context& GetIOService();
16. void Stop();
17. private:
18. AsioIOServicePool(std::size_t size = std::thread::hardware_concurrency());
19. std::vector<IOService> _ioServices;
20. std::vector<WorkPtr> _works;
21. std::vector<std::thread> _threads;
22. std::size_t _nextIOService;
23. };

AsioIOServicePool具体实现

1. #include "AsioIOServicePool.h"
2. #include <iostream>
3. using namespace std;
4. AsioIOServicePool::AsioIOServicePool(std::size_t size):_ioServices(size),
5. _works(size), _nextIOService(0){
6. for (std::size_t i = 0; i < size; ++i) {
7. _works[i] = std::unique_ptr<Work>(new Work(_ioServices[i]));
8. }
10. //遍历多个ioservice，创建多个线程，每个线程内部启动ioservice
11. for (std::size_t i = 0; i < _ioServices.size(); ++i) {
12. _threads.emplace_back([this, i]() {
13. _ioServices[i].run();
14. });
15. }
16. }
18. AsioIOServicePool::~AsioIOServicePool() {
19. std::cout << "AsioIOServicePool destruct" << endl;
20. }
22. boost::asio::io_context& AsioIOServicePool::GetIOService() {
23. auto& service = _ioServices[_nextIOService++];
24. if (_nextIOService == _ioServices.size()) {
25. _nextIOService = 0;
26. }
27. return service;
28. }
30. void AsioIOServicePool::Stop(){
31. //因为仅仅执行work.reset并不能让iocontext从run的状态中退出
32. //当iocontext已经绑定了读或写的监听事件后，还需要手动stop该服务。
33. for (auto& work : _works) {
34. //把服务先停止
35. work->get_io_context().stop();
36. work.reset();
37. }
39. for (auto& t : _threads) {
40. t.join();
41. }
42. }

CServer的处理连接逻辑

1. void CServer::HandleAccept(shared_ptr<CSession> new_session, const boost::system::error_code& error){
2. if (!error) {
3. new_session->Start();
4. lock_guard<mutex> lock(_mutex);
5. _sessions.insert(ma