Linux--多路转接之select - IT评测·应用市场-qidao123.com

#include <sys/select.h>
#include <sys/time.h>
#include <unistd.h>
int select(int nfds, fd_set *readfds, fd_set *writefds,
fd_set *exceptfds, struct timeval *timeout);

复制代码

#include <sys/time.h>
struct timeval {
time_t tv_sec; // 秒
suseconds_t tv_usec; // 微秒
};

复制代码

typedef struct {
__fd_mask fds_bits[__FD_SETSIZE/__NFDBITS];
} fd_set;

复制代码

fd_set rfds;
FD_ZERO(&rfds);
FD_SET(_fd_array[i], &rfds);

复制代码

struct timeval timeout = {0, 0};//设置等待时间
int n=select(max_fd+1,&rfds,nullptr,nullptr,&timeout);

复制代码

if(n>0)
{
for (int i = 0; i <1024; i++)
{
if (FD_ISSET(_fd_array[i], &rfds))
{
//处理对应的就绪事件
}
}
}

复制代码

while(1)
{
//包含以上内容
}

复制代码

#pragma once
#include <iostream>
#include <sys/types.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <netinet/in.h>
class InetAddr
{
private:
void GetAddress(std::string *ip, uint16_t *port)
{
*port = ntohs(_addr.sin_port);
*ip = inet_ntoa(_addr.sin_addr);
}
public:
InetAddr(const struct sockaddr_in &addr) : _addr(addr)
{
GetAddress(&_ip, &_port);
}
InetAddr(const std::string &ip, uint16_t port) : _ip(ip), _port(port)
{
_addr.sin_family = AF_INET;
_addr.sin_port = htons(_port);
_addr.sin_addr.s_addr = inet_addr(_ip.c_str());
}
InetAddr()
{}
std::string Ip()
{
return _ip;
}
bool operator == (const InetAddr &addr)
{
// if(_ip == addr._ip)
if(_ip == addr._ip && _port == addr._port) // 方便测试
{
return true;
}
return false;
}
// bool operator = (const struct sockaddr_in &addr)
// {
// _addr = addr;
// }
struct sockaddr_in Addr()
{
return _addr;
}
uint16_t Port()
{
return _port;
}
~InetAddr()
{
}
private:
struct sockaddr_in _addr;
std::string _ip;
uint16_t _port;
};

复制代码

#pragma once
#include<iostream>
#include<fstream>
#include<ctime>
#include<cstdarg>
#include<string>
#include<sys/types.h>
#include<unistd.h>
#include<cstdio>
#include"LockGuard.hpp"
using namespace std;
bool gIsSave=false;//默认输出到屏幕
const string logname="log.txt";
//1.日志是有等级的
enum Level
{
DEBUG=0,
INFO,
WARNING,
ERROR,
FATAL
};
void SaveFile(const string& filename,const string& messages)
{
ofstream out(filename,ios::app);
if(!out.is_open())
{
return;
}
out<<messages;
out.close();
}
//等级转化为字符串
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 "Unkonwn";
}
}
//获取当前时间
string GetTimeString()
{
time_t curr_time=time(nullptr);//时间戳
struct tm* format_time=localtime(&curr_time);//转化为时间结构
if(format_time==nullptr)
return "None";
char time_buffer[1024];
snprintf(time_buffer,sizeof(time_buffer),"%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 time_buffer;
}
pthread_mutex_t lock=PTHREAD_MUTEX_INITIALIZER;
//获取日志信息
void LogMessage(string filename,int line,bool issave,int level,char* format,...)
{
string levelstr=LevelToString(level);
string timestr=GetTimeString();
pid_t selfid=getpid();
char buffer[1024];
va_list arg;
va_start(arg,format);
vsnprintf(buffer,sizeof(buffer),format,arg);
va_end(arg);
string message= "[" + timestr + "]" + "[" + levelstr + "]" +
"[" + std::to_string(selfid) + "]" +
"[" + filename + "]" + "[" + std::to_string(line) + "] " + buffer + "\n";
LockGuard lockguard(&lock);
if(!issave)
{
cout<<message;
}
else
{
SaveFile(logname,message);
}
}
#define LOG(level,format,...) \
do \
{ \
LogMessage(__FILE__,__LINE__,gIsSave,level,format,##__VA_ARGS__); \
} while (0)
#define EnableFile() \
do \
{ \
gIsSave=true; \
} while (0)
#define EnableScreen() \
do \
{ \
gIsSave=false; \
} while (0)

复制代码

#ifndef __LOCK_GUARD_HPP__
#define __LOCK_GUARD_HPP__
#include <iostream>
#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;
};
#endif

复制代码

#include <iostream>
#include <string>
#include <functional>
#include <sys/types.h> /* See NOTES */
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <unistd.h>
#include <cstring>
#include <pthread.h>
#include <sys/types.h>
#include <memory>
#include "InetAddr.hpp"
#include "Log.hpp"
namespace socket_ns
{
class Socket;
const static int gbacklog=8;
using socket_sptr=std::shared_ptr<Socket>;//套接字指针
enum
{
SOCKET_ERROR = 1,
BIND_ERROR,
LISTEN_ERROR,
USAGE_ERROR
};
//在基类创建一系列虚函数，只要派生类能用到就在这里创建
class Socket
{
public:
virtual void CreateSocketOrDie() =0; //创建套接字
virtual void BindSocketOrDie(InetAddr& addr) =0; //绑定套接字
virtual void ListenSocketOrDie()=0; //监听套接字
virtual int Accepter(InetAddr* addr) =0; //接受客户端
virtual bool Connector(InetAddr &addr) = 0; //连接客户端
virtual void SetSocketAddrReuse() = 0; // 重启指定端口
virtual int SockFd() = 0; //获取Sockfd
virtual int Recv(std::string *out) = 0; //接收对方信息
virtual int Send(const std::string &in) = 0; //发送给对方信息
public:
//创建监听套接字，将一系列操作细分化，直接引用对应函数直接创建
void BuildListenSocket(InetAddr& addr)
{
CreateSocketOrDie();
SetSocketAddrReuse();
BindSocketOrDie(addr);
ListenSocketOrDie();
}
bool BuildClientSocket(InetAddr &addr)
{
CreateSocketOrDie();
return Connector(addr);
}
};
class TcpSocket : public Socket
{
public:
TcpSocket(int sockfd=-1)
:_sockfd(sockfd)
{}
void CreateSocketOrDie() override //override明确的重写基类函数
{
_sockfd=socket(AF_INET,SOCK_STREAM,0);
if(_sockfd<0)
{
LOG(FATAL, "socket error");
exit(SOCKET_ERROR);
}
LOG(DEBUG, "socket create success, sockfd is : %d\n", _sockfd);
}
void BindSocketOrDie(InetAddr& addr) override
{
struct sockaddr_in local;
memset(&local, 0, sizeof(local));
local.sin_family = AF_INET;
local.sin_port = htons(addr.Port());
local.sin_addr.s_addr = inet_addr(addr.Ip().c_str());
int n=bind(_sockfd,(struct sockaddr*)&local,sizeof(local));
if (n < 0)
{
LOG(FATAL, "bind error\n");
exit(BIND_ERROR);
}
LOG(DEBUG, "bind success, sockfd is : %d\n", _sockfd);
}
void ListenSocketOrDie() override
{
int n=listen(_sockfd,gbacklog);
if (n < 0)
{
LOG(FATAL, "listen error\n");
exit(LISTEN_ERROR);
}
LOG(DEBUG, "listen success, sockfd is : %d\n", _sockfd);
}
int Accepter(InetAddr* addr) override
{
struct sockaddr_in peer;
socklen_t len=sizeof(peer);
int sockfd = accept(_sockfd,(struct sockaddr*)&peer,&len);
if (sockfd < 0)
{
LOG(WARNING, "accept error\n");
return -1;
}
*addr=peer;
return sockfd;
}
virtual bool Connector(InetAddr& addr)
{
struct sockaddr_in server;
memset(&server,0,sizeof(server));
server.sin_family=AF_INET;
server.sin_addr.s_addr=inet_addr(addr.Ip().c_str());
server.sin_port=htons(addr.Port());
int n=connect(_sockfd,(struct sockaddr*)&server,sizeof(server));
if (n < 0)
{
std::cerr << "connect error" << std::endl;
return false;
}
return true;
}
void SetSocketAddrReuse() override
{
int opt = 1;
setsockopt(_sockfd, SOL_SOCKET, SO_REUSEADDR | SO_REUSEPORT, &opt, sizeof(opt)); //快速重启端口
}
int Recv(std::string *out) override
{
char inbuffer[1024];
ssize_t n = recv(_sockfd,inbuffer,sizeof(inbuffer)-1,0);
if (n > 0)
{
inbuffer[n] = 0;
*out += inbuffer; // 接收次数可能不只一次，一般是多次的，
}
return n;
}
int Send(const std::string &in) override
{
int n = send(_sockfd,in.c_str(),in.size(),0);
return n;
}
int SockFd() override
{
return _sockfd;
}
~TcpSocket()
{}
private:
int _sockfd;
};
}

复制代码

#pragma once
#include <iostream>
#include <string>
#include <memory>
#include "Socket.hpp"
using namespace socket_ns;
//select服务器要正确的编写，需要借助一个第三方数组来完成，保存合法的，所有的fd到数组中，方便后期批量化统一添加
class SelectServer
{
const static int defaultfd = -1; //默认sockfd
const static int N = sizeof(fd_set) * 8; //监视文件描述符的最大值
public:
SelectServer(uint16_t port)
:_port(port),
_listensock(make_unique<TcpSocket>())
{
InetAddr addr("0", _port); //网络地址初始化
_listensock->BuildListenSocket(addr);//创建监听套接字
//初始化辅助数组
for (int i = 0; i < N; i++)
{
_fd_array[i] = defaultfd;
}
_fd_array[0] = _listensock->SockFd();
}
void AcceptClient()
{
// 我们今天只关心了读，而读有：listensock 和 normal sockfd
InetAddr clientaddr;
int sockfd = _listensock->Accepter(&clientaddr); // 这里调用accept会不会阻塞呢？？不会。因为事件已经就绪了
if (sockfd < 0)
return;
LOG(DEBUG, "Get new Link, sockfd: %d, client info %s:%d\n", sockfd, clientaddr.Ip().c_str(), clientaddr.Port());
//select托管（监视状态）：将新fd放入辅助数组中
int pos = 1;
for (; pos < N; pos++)
{
if (_fd_array[pos] == defaultfd)
break;
}//让pos到辅助数组的空缺位置
if (pos == N)//说明监视的文件描述符满了
{
::close(sockfd); // sockfd->Close();
LOG(WARNING, "server is full!\n");
return;
}
else
{
_fd_array[pos] = sockfd;
LOG(DEBUG, "%d add to select array!\n", sockfd);
}
LOG(DEBUG, "curr fd_array[] fd list : %s\n", RfdsToString().c_str());
}
void ServiceIO(int pos)
{
char buffer[1024];
ssize_t n = ::recv(_fd_array[pos], buffer, sizeof(buffer) - 1, 0); // 这里读取会不会被阻塞？不会
if (n > 0)//处理接收数据
{
buffer[n] = 0;
std::cout << "client say# " << buffer << std::endl;
std::string echo_string = "[server echo]# ";
echo_string += buffer;
::send(_fd_array[pos], echo_string.c_str(), echo_string.size(), 0);//返回给客户端
}
else if (n == 0)//说明对方已断开连接
{
LOG(DEBUG, "%d is closed\n", _fd_array[pos]);
::close(_fd_array[pos]);
_fd_array[pos] = defaultfd;
LOG(DEBUG, "curr fd_array[] fd list : %s\n", RfdsToString().c_str());
}
else//出现错误
{
LOG(DEBUG, "%d recv error\n", _fd_array[pos]);
::close(_fd_array[pos]);
_fd_array[pos] = defaultfd;
LOG(DEBUG, "curr fd_array[] fd list : %s\n", RfdsToString().c_str());
}
}
//处理准备就绪的事件
void HandlerEvent(fd_set &rfds)
{
for (int i = 0; i < N; i++)
{
if (_fd_array[i] == defaultfd)
continue;
if (FD_ISSET(_fd_array[i], &rfds))
{
if (_fd_array[i] == _listensock->SockFd())//新的连接
{
AcceptClient();
}
else
{
// 普通的sockfd读事件就绪
ServiceIO(i);
}
}
}
}
void Loop()
{
while(true)
{
//监听套接字在等待对方发送连接
//新的连接 == 读事件就绪
//要将listensock添加到select中！
fd_set rfds; //一个记录文件描述符状态的集合
FD_ZERO(&rfds);//将所有文件描述符移除集合
int max_fd = defaultfd;//最大的文件描述符值
for (int i = 0; i < N; i++)
{
if (_fd_array[i] == defaultfd)
continue;
FD_SET(_fd_array[i], &rfds); // 将所有合法的fd添加到rfds中
if (max_fd < _fd_array[i])
{
max_fd = _fd_array[i]; // 更新出最大的fd的值
}
}
struct timeval timeout = {0, 0};//设置等待时间
int n=select(max_fd+1,&rfds,nullptr,nullptr,nullptr);
//timeout 是 NULL，则 select() 会无限期地等待直到至少有一个文件描述符准备就绪
switch (n)
{
case 0://指定时间内没有任何文件描述符准备就绪
LOG(INFO, "timeout, %d.%d\n", timeout.tv_sec, timeout.tv_usec);
break;
case -1://出现错误
LOG(ERROR, "select error...\n");
break;
default://成功状态
LOG(DEBUG, "Event Happen. n : %d\n", n); // 底层有一个事件就绪，select为什么会一直通知我？因为：我们没有处理！
HandlerEvent(rfds);
break;
}
}
}
//打印出已存在的fd
std::string RfdsToString()
{
std::string fdstr;
for (int i = 0; i < N; i++)
{
if (_fd_array[i] == defaultfd)
continue;
fdstr += std::to_string(_fd_array[i]);
fdstr += " ";
}
return fdstr;
}
~SelectServer()
{
}
private:
uint16_t _port; //端口号
std::unique_ptr<Socket> _listensock;//监听socket
int _fd_array[N]; // 辅助数组
};

复制代码

#include "SelectSever.hpp"
#include <memory>
// ./selectserver port
int main(int argc, char *argv[])
{
if (argc != 2)
{
std::cout << "Usage: " << argv[0] << " port" << std::endl;
return 0;
}
uint16_t port = std::stoi(argv[1]);//获取端口号
EnableScreen();
std::unique_ptr<SelectServer> svr = std::make_unique<SelectServer>(port);
svr->Loop();
return 0;
}

复制代码