阅读量:0
文章目录
TCP客户端connect断线重连
1、为什么要断线重连
客户端会面临服务器崩溃的情况，我们可以试着写一个客户端重连的代码，模拟并理解一些客户端行为，比如游戏客户端等.
考虑到下面2种情况：
- 服务器故障突然断开连接，过几秒又恢复连接，那么在这等待的几秒，客户端可以重新发起连接。
- 客户端wifi突然断了，也就是没网，然后过几秒又恢复了，客户端又可以重新发起连接了。
2、实现代码
代码主要实现细节：
枚举类型
Status:
- 用于表示连接状态，包括
NEW,CONNECTED,CONNECTING,DISCONNECTED,CLOSE。全局常量:
defaultsockfd，默认套接字文件描述符为-1。
retryinterval，重连间隔时间为1秒。
retryamxtimes，重连最大次数为5次。
Connection类:
构造函数初始化服务器IP、端口、状态、重连间隔和最大重连次数。
ConnectStatus()方法返回当前连接状态。
Connect()方法负责建立连接，并设置连接状态。
Process()方法负责发送和接收数据，处理通信过程中的各种状态变化。
ReConnect()方法处理重连逻辑，尝试在连接失败时进行多次重连。
DisConnect()方法负责断开连接并重置套接字文件描述符。
TcpClient类:
构造函数初始化连接对象。
Execute()方法根据连接状态执行相应操作，包括建立连接、处理数据、重连和断开连接。
Usage()函数:
- 用于输出使用说明。
main()函数:
- 检查命令行参数，初始化
TcpClient对象并执行连接逻辑。
TcpClient.cc文件：#include <iostream> #include <string> #include <strings.h> #include <unistd.h> #include "Comm.hpp" enum class Status { NEW, CONNECTED, CONNECTING, DISCONNECTED, CLOSE }; const int defaultsockfd = -1; const int retryinterval = 1; // 重连间隔时间 const int retryamxtimes = 5; // 重连最大次数 class Connection { public: Connection(std::string serverip, uint16_t serverport) : _sockfd(defaultsockfd), _serverip(serverip), _serverport(serverport), _status(Status::NEW), _retry_interval(retryinterval), _retry_max_times(retryamxtimes) { } Status ConnectStatus() { return _status; } void Connect() { _sockfd = socket(AF_INET, SOCK_STREAM, 0); if (_sockfd < 0) { _status = Status::DISCONNECTED; exit(CREATE_ERROR); } struct sockaddr_in server; bzero(&server, sizeof(server)); server.sin_family = AF_INET; server.sin_port = htons(_serverport); // int inet_pton(int af, const char *src, void *dst); inet_pton(AF_INET, _serverip.c_str(), &server.sin_addr.s_addr); // 也可以&server.sin_addr，因为sin_addr只有s_addr int n = ::connect(_sockfd, CONV(&server), sizeof(server)); if (n < 0) { DisConnect(); // _status = Status::DISCONNECTED; return; } std::cout << "connect success" << std::endl; _status = Status::CONNECTED; // 已连接 } void Process() { while (true) { // 发送数据 std::string message; std::cout << "Please Enter# "; std::getline(std::cin, message); int n = send(_sockfd, message.c_str(), message.size(), 0); if (n > 0) { // 接收数据 char buff[1024]; int m = recv(_sockfd, buff, sizeof(buff) - 0, 0); if (m > 0) { buff[m] = 0; std::cout << "Server Echo$ " << buff << std::endl; } else { _status = Status::DISCONNECTED; // 接收不成功就重连 std::cerr << "recv error" << std::endl; break; } } else { _status = Status::CLOSE; // 发送不成功就退出 std::cerr << "send error" << std::endl; break; } } } void ReConnect() { _status = Status::CONNECTING; int cnt = 1; while (true) { Connect(); if (_status == Status::CONNECTED) { break; } std::cout << "正在重连，重连次数 : " << cnt++ << std::endl; if (cnt > _retry_max_times) { _status = Status::CLOSE; // 重连失败 std::cout << "重连失败，请检查网络.." << std::endl; break; } sleep(_retry_interval); } } void DisConnect() { if (_sockfd > defaultsockfd) { close(_sockfd); _sockfd = defaultsockfd; } } private: int _sockfd; std::string _serverip; uint16_t _serverport; Status _status; int _retry_interval; int _retry_max_times; }; class TcpClient { public: TcpClient(std::string serverip, uint16_t serverport) : _connect(serverip, serverport) { } void Execute() { while (true) { switch (_connect.ConnectStatus()) { case Status::NEW: _connect.Connect(); break; case Status::CONNECTED: _connect.Process(); break; case Status::DISCONNECTED: _connect.ReConnect(); break; case Status::CLOSE: _connect.DisConnect(); return; // 断开连接了，重连不管用了 default: break; } } } private: Connection _connect; }; void Usage() { std::cout << "Please use format : ./tcp_client serverip serverport" << std::endl; } int main(int argc, char *argv[]) { if (argc != 3) { Usage(); exit(USAGE_ERROR); } std::string serverip = argv[1]; uint16_t serverport = std::stoi(argv[2]); TcpClient tcpclient(serverip, serverport); tcpclient.Execute(); return 0; }和UDP服务器响应程序一样，客户端发什么就回什么，只不过多了建立连接的步骤。
Comm.hpp文件#pragma once #include "InetAddr.hpp" enum errorcode { CREATE_ERROR = 1, BIND_ERROR, LISTEN_ERROR, SEND_ERROR, RECV_ERROR, CONNECT_ERROR, FORK_ERROR, USAGE_ERROR }; #define CONV(ADDR) ((struct sockaddr *)ADDR) std::string CombineIpAndPort(InetAddr addr) { return "[" + addr.Ip() + ":" + std::to_string(addr.Port()) + "] "; }
InetAddr.hpp文件#pragma once #include <iostream> #include <sys/types.h> /* See NOTES */ #include <sys/socket.h> #include <arpa/inet.h> #include <netinet/in.h> #include <string> class InetAddr { void GetAddress(std::string *ip, uint16_t *port) { // char *inet_ntoa(struct in_addr in); *ip = inet_ntoa(_addr.sin_addr); *port = ntohs(_addr.sin_port); } public: InetAddr(const struct sockaddr_in &addr) : _addr(addr) { GetAddress(&_ip, &_port); } std::string Ip() { return _ip; } uint16_t Port() { return _port; } bool operator==(InetAddr &addr) { return _ip == addr.Ip() && _port == addr.Port(); } const struct sockaddr_in& GetAddr() { return _addr; } ~InetAddr() {} private: struct sockaddr_in _addr; std::string _ip; uint16_t _port; };
LockGuard.hpp文件# pragma once #include <pthread.h> class LockGuard { public: LockGuard(pthread_mutex_t *mutex) : _mutex(mutex) { pthread_mutex_lock(_mutex); // 构造加锁 } ~LockGuard() { pthread_mutex_unlock(_mutex); // 析构解锁 } private: pthread_mutex_t *_mutex; };
Log.hpp文件#pragma once #include <string> #include <iostream> #include <fstream> #include <unistd.h> #include <stdarg.h> #include <sys/types.h> #include "LockGuard.hpp" using namespace std; bool isSave = false; // 默认向显示器打印 pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER; #define FILEPATH "./log.txt" enum level { DEBUG = 0, INFO, WARNING, ERROR, FATAL }; void SaveToFile(const string &message) { ofstream out(FILEPATH, ios_base::app); if (!out.is_open()) return; out << message; out.close(); } std::string LevelToString(int level) { switch (level) { case DEBUG: return "Debug"; case INFO: return "Info"; case WARNING: return "Warning"; case ERROR: return "Error"; case FATAL: return "Fatal"; default: return "Unknow"; } } std::string GetTimeString() { time_t curr_time = time(nullptr); struct tm *format_time = localtime(&curr_time); if (format_time == nullptr) return "None"; char buff[1024]; snprintf(buff, sizeof(buff), "%d-%d-%d %d:%d:%d", format_time->tm_year + 1900, format_time->tm_mon + 1, format_time->tm_mday, format_time->tm_hour, format_time->tm_min, format_time->tm_sec); return buff; } void LogMessage(const std::string filename, int line, bool issave, int level, const char *format, ...) { std::string levelstr = LevelToString(level); std::string timestr = GetTimeString(); pid_t pid = getpid(); char buff[1024]; va_list arg; // int vsnprintf(char *str, size_t size, const char *format, va_list ap); // 使用可变参数 va_start(arg, format); vsnprintf(buff, sizeof(buff), format, arg); va_end(arg); LockGuard lock(&mutex); std::string message = "[" + timestr + "]" + "[" + levelstr + "]" + "[pid:" + std::to_string(pid) + "]" + "[" + filename + "]" + "[" + std::to_string(line) + "] " + buff + '\n'; if (issave == false) std::cout << message; else SaveToFile(message); } // 固定文件名和行数 #define LOG(level, format, ...) \ do \ { \ LogMessage(__FILE__, __LINE__, isSave, level, format, ##__VA_ARGS__); \ } while (0) #define EnableScreen() \ do \ { \ isSave = false; \ } while (0) #define EnableFile() \ do \ { \ isSave = true; \ } while (0) void Test(int num, ...) { va_list arg; va_start(arg, num); while (num--) { int data = va_arg(arg, int); std::cout << data << " "; } std::cout << std::endl; va_end(arg); }
Main.cc文件#include <iostream> #include <memory> #include "TcpServer.hpp" void Usage() { // printf("./udp_server serverip serverport\n"); printf("Usage : ./udp_server serverport\n"); // ip 已经设置为0 } int main(int argc, char *argv[]) { // if (argc != 3) if (argc != 2) { Usage(); exit(USAGE_ERROR); } uint16_t serverport = std::stoi(argv[1]); std::unique_ptr<TcpServer> tsvr = std::make_unique<TcpServer>(serverport); tsvr->InitServer(); tsvr->Start(); return 0; }
Makefile文件.PHONY:all all:tcp_client tcp_server tcp_client:TcpClient.cc g++ -o $@ $^ -std=c++14 -lpthread tcp_server:Main.cc g++ -o $@ $^ -std=c++14 -lpthread .PHONY:clean clean: rm -f tcp_server tcp_client
TcpServer.hpp文件#pragma once #include <sys/types.h> /* See NOTES */ #include <sys/wait.h> #include <sys/socket.h> #include <arpa/inet.h> #include <netinet/in.h> #include <error.h> #include <string.h> #include <pthread.h> #include <functional> #include "Log.hpp" #include "InetAddr.hpp" #include "Comm.hpp" #include "Threadpool.hpp" const int defaultsockfd = -1; int gbacklog = 16; // 暂时先用 using task_t = std::function<void()>; // 声明 class TcpServer; class ThreadData { public: ThreadData(int sockfd, InetAddr addr, TcpServer *self) : _sockfd(sockfd), _addr(addr), _self(self) {} ~ThreadData() = default; public: int _sockfd; InetAddr _addr; TcpServer *_self; }; class TcpServer { public: TcpServer(uint16_t port) : _port(port), _listensock(defaultsockfd), _isrunning(false) { } void InitServer() { // 创建 _listensock = socket(AF_INET, SOCK_STREAM, 0); // 这个就是文件描述符 if (_listensock < 0) { LOG(FATAL, "create sockfd error, error code : %d, error string : %s", errno, strerror(errno)); exit(CREATE_ERROR); } LOG(INFO, "create sockfd success"); struct sockaddr_in local; bzero(&local, sizeof(local)); local.sin_family = AF_INET; local.sin_port = htons(_port); local.sin_addr.s_addr = INADDR_ANY; // 绑定 int n = ::bind(_listensock, CONV(&local), sizeof(local)); if (n < 0) { LOG(FATAL, "bind sockfd error, error code : %d, error string : %s", errno, strerror(errno)); exit(BIND_ERROR); } LOG(INFO, "bind sockfd success"); } void Service(int sockfd, InetAddr client) { while (true) { // TCP是字节流（可以使用write和read接口），UDP是数据报 char buff[1024]; // 接收消息 int n = ::read(sockfd, buff, sizeof(buff)); // bug，接收数据可能收到的不完整，比如1+100，可能先收到1+1，再收到00 -- 按序到达 std::string clientAddr = CombineIpAndPort(client); if (n > 0) { buff[n] = 0; std::string message = clientAddr + buff; LOG(INFO, "get message : \n %s", message.c_str()); // 发送消息 int m = ::write(sockfd, buff, strlen(buff)); if (m < 0) { LOG(FATAL, "send message error ,error code : %d , error string : %s", errno, strerror(errno)); exit(SEND_ERROR); } } else if (n == 0) { // 发送端不发送数据了 LOG(INFO, "%s quit", clientAddr.c_str()); break; } else { LOG(FATAL, "recv message error ,error code : %d , error string : %s", errno, strerror(errno)); exit(RECV_ERROR); } } ::close(sockfd); // 服务结束，关闭文件描述符，避免文件描述符泄漏 } static void *HandlerService(void *args) { pthread_detach(pthread_self()); // 分离线程 ThreadData *td = static_cast<ThreadData *>(args); td->_self->Service(td->_sockfd, td->_addr); delete td; return nullptr; } void Start() { _isrunning = true; while (_isrunning) { // 监听 int ret = ::listen(_listensock, gbacklog); if (ret < 0) { LOG(FATAL, "listen error, error code : %d , error string : %s", errno, strerror(errno)); exit(LISTEN_ERROR); } LOG(INFO, "listen success!"); struct sockaddr_in peer; socklen_t len = sizeof(peer); // 获取新连接 int sockfd = accept(_listensock, CONV(&peer), &len); // 建立连接成功，创建新文件描述符进行通信 if (sockfd < 0) { LOG(WARNING, "accept error, error code : %d , error string : %s", errno, strerror(errno)); continue; } LOG(INFO, "accept success! new sockfd : %d", sockfd); InetAddr addr(peer); // 给后面提供传入的ip、port // 服务 -- 发送和接收数据 // V0 -- 单进程 // Service(sockfd, addr); // 这里是while死循环，没有运行完就一直运行，下一个请求来的时候得这个while退出才能执行 // v1 -- 多进程 // int id = fork(); // if (id == 0) // { // // 子进程 // ::close(_listensock); // 子进程对监听文件描述符不关心 // if (fork() > 0) // exit(0); // 子进程创建进程后退出，孙子进程被系统领养，不用等待 // Service(sockfd, addr); // 孙子进程执行任务 // exit(0); // } // else if (id > 0) // { // ::close(sockfd); // 这里每次关闭的文件描述符都是4，使得每次accept创建的文件描述符都是4，这个4是留个各个子进程（子进程再给孙子进程）的（互不影响） // // 父进程 // pid_t rid = waitpid(id, nullptr, 0); // 虽然是阻塞等待，但是子进程是刚创建就退出来了，让孙子进程（孤儿进程）执行任务，接下来继续监听和建立连接 // if (rid == id) // { // LOG(INFO, "wait child process success"); // } // } // else // { // // 异常 // LOG(FATAL, "fork error ,error code : %d , error string : %s", errno, strerror(errno)); // exit(FORK_ERROR); // } // v2 -- 多线程 // pthread_t tid; // ThreadData *td = new ThreadData(sockfd, addr, this); // 传指针 // pthread_create(&tid, nullptr, HandlerService, td); // 这里创建线程后，线程去做执行任务，主线程继续向下执行 , 并且线程不能关闭sockf，线程和进程共享文件描述符表 // v3 -- 线程池 task_t t = std::bind(&TcpServer::Service, this, sockfd, addr); Threadpool<task_t>::GetInstance()->Enqueue(t); // v4 -- 进程池 -- 不推荐，需要传递文件描述符！ } _isrunning = false; } ~TcpServer() { if (_listensock > defaultsockfd) ::close(_listensock); // 不用了关闭监听 } private: uint16_t _port; int _listensock; bool _isrunning; };
Thread.hpp文件#ifndef __THREAD_HPP__ #define __THREAD_HPP__ #include <iostream> #include <string> #include <unistd.h> #include <functional> #include <pthread.h> using namespace std; // 封装Linux线程 namespace ThreadModule { using func_t = function<void(string &)>; class Thread { public: // /* ThreadData* */Thread(func_t<T> func, T data, const string& name = "default name") : _func(func), _data(data), _threadname(name), _stop(true) {} Thread(func_t func, const string &name = "default name") : _func(func), _threadname(name), _stop(true) {} void Execute() { _func(_threadname); // _func(_data); } // 隐含this static void *threadroutine(void *arg) { Thread *self = static_cast<Thread *>(arg); self->Execute(); // static 访问不了成员变量 return nullptr; } bool Start() { int n = pthread_create(&_tid, nullptr, threadroutine, this); if (!n) { _stop = false; return true; } else { return false; } } void Detach() { if (!_stop) { pthread_detach(_tid); } } void Join() { if (!_stop) { pthread_join(_tid, nullptr); } } string name() { return _threadname; } void Stop() { _stop = true; } // ~Thread() {} private: pthread_t _tid; string _threadname; func_t _func; bool _stop; }; } // namespace ThreadModule #endif
Threadpool.hpp文件#pragma once #include <vector> #include <queue> #include <queue> #include "Thread.hpp" #include <pthread.h> #include "LockGuard.hpp" using namespace ThreadModule; const int NUM = 3; template <typename T> class Threadpool { void LockQueue(pthread_mutex_t &mutex) { pthread_mutex_lock(&mutex); } void UnLockQueue(pthread_mutex_t &mutex) { pthread_mutex_unlock(&mutex); } void SleepThread(pthread_cond_t &cond, pthread_mutex_t &mutex) { pthread_cond_wait(&cond, &mutex); } void WakeUpThread(pthread_cond_t &cond) { pthread_cond_signal(&cond); } void WakeUpAll(pthread_cond_t &cond) { pthread_cond_broadcast(&_cond); } Threadpool(const int threadnum = NUM) : _threadnum(threadnum), _waitnum(0), _isrunning(false) { pthread_mutex_init(&_mutex, nullptr); pthread_cond_init(&_cond, nullptr); LOG(INFO, "Threadpool Constructor successful ! "); } void TaskHandler(string &name) { // sleep(1); // cout << name << " : hh " << endl; // sleep(1); LOG(DEBUG, "%s is running", name.c_str()); while (true) { LockQueue(_mutex); while (_task_queue.empty() && _isrunning) { // 等待 ++_waitnum; SleepThread(_cond, _mutex); --_waitnum; } // 此时一定大于一个线程没有休眠 if (_task_queue.empty() && !_isrunning) { // 此时任务队列已经没有内容，且此时线程池已经停止 UnLockQueue(_mutex); cout << name << " quit ... " << endl; break; } LOG(DEBUG, "%s get task sucessful !", name.c_str()); // 其他情况就得处理任务 T t = _task_queue.front(); _task_queue.pop(); UnLockQueue(_mutex); // 处理任务 t(); // cout << name << " : " << t.stringResult() << endl; // LOG(DEBUG, "%s handler task sucessful ! Result is %s", name.c_str(), t.stringResult().c_str()); sleep(1); } } void InitThreadPool() { for (int i = 0; i < _threadnum; ++i) { string name = "Thread - " + to_string(i + 1); _threads.emplace_back(bind(&Threadpool::TaskHandler, this, placeholders::_1), name); } _isrunning = true; LOG(INFO, "Init Threadpool successful !"); } public: static Threadpool<T> *GetInstance(int threadnum = NUM) { if (_instance == nullptr) { LockGuard lockguard(&_lock); if (_instance == nullptr) { // pthread_mutex_lock(&_lock); // 第一次创建线程池 _instance = new Threadpool<T>(threadnum); _instance->InitThreadPool(); _instance->Start(); LOG(DEBUG, "第一次创建线程池"); // pthread_mutex_unlock(&_lock); return _instance; } } LOG(DEBUG, "获取线程池"); return _instance; } bool Enqueue(const T &in) { bool ret = false; LockQueue(_mutex); if (_isrunning) { _task_queue.push(in); if (_waitnum > 0) WakeUpThread(_cond); LOG(DEBUG, "enqueue sucessful..."); ret = true; } UnLockQueue(_mutex); return ret; } void Stop() { LockQueue(_mutex); _isrunning = false; if (_waitnum > 0) WakeUpAll(_cond); UnLockQueue(_mutex); } void Start() { for (auto &thread : _threads) { thread.Start(); LOG(INFO, "%s is start sucessful...", thread.name().c_str()); } } void Wait() { for (auto &thread : _threads) { thread.Join(); LOG(INFO, "%s is quit...", thread.name().c_str()); } } ~Threadpool() { pthread_mutex_destroy(&_mutex); pthread_cond_destroy(&_cond); LOG(INFO, "delete mutex sucessful !"); } private: vector<Thread> _threads; queue<T> _task_queue; int _threadnum; int _waitnum; pthread_mutex_t _mutex; // 互斥访问任务队列 pthread_cond_t _cond; bool _isrunning; // 懒汉模式 static Threadpool<T> *_instance; static pthread_mutex_t _lock; }; template <typename T> Threadpool<T> *Threadpool<T>::_instance = nullptr; template <typename T> pthread_mutex_t Threadpool<T>::_lock = PTHREAD_MUTEX_INITIALIZER;运行结果：
OKOK，TCP客户端connect短线重连就到这里，如果你对Linux和C++也感兴趣的话，可以看看我的主页哦。下面是我的github主页，里面记录了我的学习代码和leetcode的一些题的题解，有兴趣的可以看看。
