TCP服务器
- 锁:Lock.hpp
- 守护进程:daemonize.hpp
- 日志文件:log.hpp
- 任务处理 Task.hpp
- 线程池 ThreadPool.hpp
- 客户端 TCPClient.cc
- 服务器 TCPServer.cc
- 头文件包 util.hpp
- Makefile
锁:Lock.hpp
代码
#pragma once
#include <iostream>
#include <pthread.h>
class Mutex
{
public:
Mutex()
{
pthread_mutex_init(&lock_, nullptr);
}
void lock()
{
pthread_mutex_lock(&lock_);
}
void unlock()
{
pthread_mutex_unlock(&lock_);
}
~Mutex()
{
pthread_mutex_destroy(&lock_);
}
private:
pthread_mutex_t lock_;
};
class LockGuard
{
public:
LockGuard(Mutex *mutex) : mutex_(mutex)
{
mutex_->lock();
std::cout << "加锁成功..." << std::endl;
}
~LockGuard()
{
mutex_->unlock();
std::cout << "解锁成功...." << std::endl;
}
private:
Mutex *mutex_;
};
介绍
- Mutex 类:
- LockGuard 类:
守护进程:daemonize.hpp
代码
#pragma once
#include <cstdio>
#include <iostream>
#include <signal.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
void daemonize()
{
int fd = 0;
// 1. 忽略SIGPIPE
signal(SIGPIPE, SIG_IGN);
// 2. 更改进程的工作目录
// chdir();
// 3. 让自己不要成为进程组组长
if (fork() > 0)
exit(1);
// 4. 设置自己是一个独立的会话
setsid();
// 5. 重定向0,1,2
if ((fd = open("/dev/null", O_RDWR)) != -1) // fd == 3
{
dup2(fd, STDIN_FILENO);
dup2(fd, STDOUT_FILENO);
dup2(fd, STDERR_FILENO);
// 6. 关闭掉不需要的fd
if(fd > STDERR_FILENO) close(fd);
}
}
说明
- 具体来说,这段代码中的 daemonize() 函数实现了以下步骤:
- 忽略 SIGPIPE 信号:
- 更改进程的工作目录:
- 不要成为进程组组长:
- 设置独立会话:
- 重定向标准输入、输出和错误:
- 关闭不需要的文件描述符:
日志文件:log.hpp
代码
#pragma once
#include <cstdio>
#include <ctime>
#include <cstdarg>
#include <cassert>
#include <cassert>
#include <cstring>
#include <cerrno>
#include <stdlib.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <unistd.h>
#define DEBUG 0
#define NOTICE 1
#define WARINING 2
#define FATAL 3
const char *log_level[] = {"DEBUG", "NOTICE", "WARINING", "FATAL"};
#define LOGFILE "serverTcp.log"
// logMessage(DEBUG, "%d", 10);
void logMessage(int level, const char *format, ...)
{
assert(level >= DEBUG);
assert(level <= FATAL);
char *name = getenv("USER");
char logInfo[1024];
va_list ap; // ap -> char*
va_start(ap, format);
vsnprintf(logInfo, sizeof(logInfo) - 1, format, ap);
va_end(ap); // ap = NULL
// 每次打开太麻烦
umask(0);
int fd = open(LOGFILE, O_WRONLY | O_CREAT | O_APPEND, 0666);
assert(fd >= 0);
FILE *out = (level == FATAL) ? stderr : stdout;
dup2(fd, 1);
dup2(fd, 2);
fprintf(out, "%s | %u | %s | %s\n",
log_level[level],
(unsigned int)time(nullptr),
name == nullptr ? "unknow" : name,
logInfo);
fflush(out); // 将C缓冲区中的数据刷新到OS
fsync(fd); // 将OS中的数据尽快刷盘
close(fd);
// char *s = format;
// while(s){
// case '%':
// if(*(s+1) == 'd') int x = va_arg(ap, int);
// break;
// }
}
说明
任务处理 Task.hpp
代码
#pragma once
#include <iostream>
#include <string>
#include <functional>
#include <pthread.h>
#include "log.hpp"
class Task
{
public:
//等价于
// typedef std::function<void (int, std::string, uint16_t)> callback_t;
using callback_t = std::function<void (int, std::string, uint16_t)>;
private:
int sock_; // 给用户提供IO服务的sock
uint16_t port_; // client port
std::string ip_; // client ip
callback_t func_; // 回调方法
public:
Task():sock_(-1), port_(-1)
{}
Task(int sock, std::string ip, uint16_t port, callback_t func)
: sock_(sock), ip_(ip), port_(port), func_(func)
{}
void operator () ()
{
logMessage(DEBUG, "线程ID[%p]处理%s:%d的请求 开始啦...",\
pthread_self(), ip_.c_str(), port_);
func_(sock_, ip_, port_);
logMessage(DEBUG, "线程ID[%p]处理%s:%d的请求 结束啦...",\
pthread_self(), ip_.c_str(), port_);
}
~Task()
{}
};
说明
线程池 ThreadPool.hpp
代码
#pragma once
#include <iostream>
#include <cassert>
#include <queue>
#include <memory>
#include <cstdlib>
#include <pthread.h>
#include <unistd.h>
#include <sys/prctl.h>
#include "Lock.hpp"
using namespace std;
int gThreadNum = 15;
template <class T>
class ThreadPool
{
private:
ThreadPool(int threadNum = gThreadNum) : threadNum_(threadNum), isStart_(false)
{
assert(threadNum_ > 0);
pthread_mutex_init(&mutex_, nullptr);
pthread_cond_init(&cond_, nullptr);
}
ThreadPool(const ThreadPool<T> &) = delete;
void operator=(const ThreadPool<T>&) = delete;
public:
static ThreadPool<T> *getInstance()
{
static Mutex mutex;
if (nullptr == instance) //仅仅是过滤重复的判断
{
LockGuard lockguard(&mutex); //进入代码块,加锁。退出代码块,自动解锁
if (nullptr == instance)
{
instance = new ThreadPool<T>();
}
}
return instance;
}
//类内成员, 成员函数,都有默认参数this
static void *threadRoutine(void *args)
{
pthread_detach(pthread_self());
ThreadPool<T> *tp = static_cast<ThreadPool<T> *>(args);
// prctl(PR_SET_NAME, "follower"); // 更改线程名称
while (1)
{
tp->lockQueue();
while (!tp->haveTask())
{
tp->waitForTask();
}
//这个任务就被拿到了线程的上下文中
T t = tp->pop();
tp->unlockQueue();
t(); // 让指定的先处理这个任务
}
}
void start()
{
assert(!isStart_);
for (int i = 0; i < threadNum_; i++)
{
pthread_t temp;
pthread_create(&temp, nullptr, threadRoutine, this);
}
isStart_ = true;
}
void push(const T &in)
{
lockQueue();
taskQueue_.push(in);
choiceThreadForHandler();
unlockQueue();
}
~ThreadPool()
{
pthread_mutex_destroy(&mutex_);
pthread_cond_destroy(&cond_);
}
int threadNum()
{
return threadNum_;
}
private:
void lockQueue() { pthread_mutex_lock(&mutex_); }
void unlockQueue() { pthread_mutex_unlock(&mutex_); }
bool haveTask() { return !taskQueue_.empty(); }
void waitForTask() { pthread_cond_wait(&cond_, &mutex_); }
void choiceThreadForHandler() { pthread_cond_signal(&cond_); }
T pop()
{
T temp = taskQueue_.front();
taskQueue_.pop();
return temp;
}
private:
bool isStart_;
int threadNum_;
queue<T> taskQueue_;
pthread_mutex_t mutex_;
pthread_cond_t cond_;
static ThreadPool<T> *instance;
// const static int a = 100;
};
template <class T>
ThreadPool<T> *ThreadPool<T>::instance = nullptr;
// #pragma once
// #include <iostream>
// #include <cassert>
// #include <queue>
// #include <memory>
// #include <cstdlib>
// #include <pthread.h>
// #include <unistd.h>
// #include <sys/prctl.h>
// #include "Lock.hpp"
// int gThreadNum=10;//线程数目
// template<class T>
// class TreadPool
// {
// private:
// TreadPool(int threadNum=gThreadNum ):threadNum_(threadNum),isStart_(false)
// {
// assert(threadNum_>0);
// pthread_mutexattr_init(&mutex_,nullptr);//互斥锁
// pthread_cond_init(&cond_,nullptr);//条件变量
// }
// ThreadPool(const ThreadPool<T> &) = delete;
// void operator=(const ThreadPool<T>&) = delete;
// public:
// static ThreadPool<T>*getinstance() //单例
// {
// static Mutex mutex;
// if(nullptr==instance)
// {
// LockGuard lockguard(&mutex); //进入代码块 加锁 退出 解锁
// if(nullptr==instance)
// {
// instance=new TreadPool<T>();
// }
// }
// return instance;
// }
// 类内成员 成员函数 都有默认参数 this
// static void *threadRoutine(void *args)
// {
// pthread_detach(pthread_self());
// ThreadPool<T>*tp =static_cast<ThreadPool<T>*>(args);
// while(1)
// {
// tp->lockQueue();
// while(!tp->havaTask())
// {
// tp->waitForTask();
// }
// 运行到这里任务就进入线程
// T t=tp->pop();
// tp->unlockQueue();
// t();//让指定的先处理这个任务
// }
// }
// void start()
// {
// asser(!isStart_);
// for(int i=0;i<threadNum_;i++)
// {
// pthread_t temp;
// pthread_create(&temp,nullptr,threadRoutine,this);
// }
// isStart_=true;
// }
// void push(const T&in)
// {
// lockQueue();
// taskQueue_.push(in);
// choiceT
// }
// ~TreadPool()
// {
// pthread_mutex_destroy(&mutex_);
// pthread_cond_destroy(&cond_);
// }
// private:
// void lockQueue() { pthread_mutex_lock(&mutex_); }
// void unlockQueue() { pthread_mutex_unlock(&mutex_); }
// bool haveTask() { return !taskQueue_.empty(); }
// void waitForTask() { pthread_cond_wait(&cond_, &mutex_); }
// void choiceThreadForHandler() { pthread_cond_signal(&cond_); }
// T pop()
// {
// T temp = taskQueue_.front();
// taskQueue_.pop();
// return temp;
// }
// bool isStart_;
// int threadNum_;
// queue<T> taskQueue_;
// pthread_mutex_t mutex_;
// pthread_cond_t cond_;
// static ThreadPool<T> *instance;
// const static int a = 100;
// };
说明
客户端 TCPClient.cc
代码
#include "util.hpp"
// 2. 需要bind吗??需要,但是不需要自己显示的bind! 不要自己bind!!!!
// 3. 需要listen吗?不需要的!
// 4. 需要accept吗?不需要的!
volatile bool quit = false;
static void Usage(std::string proc)
{
std::cerr << "Usage:\n\t" << proc << " serverIp serverPort" << std::endl;
std::cerr << "Example:\n\t" << proc << " 127.0.0.1 8081\n"
<< std::endl;
}
// ./clientTcp serverIp serverPort
int main(int argc, char *argv[])
{
if (argc != 3)
{
Usage(argv[0]);
exit(USAGE_ERR);
}
std::string serverIp = argv[1];
uint16_t serverPort = atoi(argv[2]);
// 1. 创建socket SOCK_STREAM
int sock = socket(AF_INET, SOCK_STREAM, 0);
if (sock < 0)
{
std::cerr << "socket: " << strerror(errno) << std::endl;
exit(SOCKET_ERR);
}
// 2. connect,发起链接请求,你想谁发起请求呢??当然是向服务器发起请求喽
// 2.1 先填充需要连接的远端主机的基本信息
struct sockaddr_in server;
memset(&server, 0, sizeof(server));
server.sin_family = AF_INET;
server.sin_port = htons(serverPort);
inet_aton(serverIp.c_str(), &server.sin_addr);
// 2.2 发起请求,connect 会自动帮我们进行bind!
if (connect(sock, (const struct sockaddr *)&server, sizeof(server)) != 0)
{
std::cerr << "connect: " << strerror(errno) << std::endl;
exit(CONN_ERR);
}
std::cout << "info : connect success: " << sock << std::endl;
std::string message;
while (!quit)
{
message.clear();
std::cout << "请输入你的消息>>> ";
std::getline(std::cin, message); // 结尾不会有\n
if (strcasecmp(message.c_str(), "quit") == 0)
quit = true;
ssize_t s = write(sock, message.c_str(), message.size());
if (s > 0)
{
message.resize(1024);
ssize_t s = read(sock, (char *)(message.c_str()), 1024);
if (s > 0)
message[s] = 0;
std::cout << "Server Echo>>> " << message << std::endl;
}
else if (s <= 0)
{
break;
}
}
close(sock);
return 0;
}
说明
服务器 TCPServer.cc
代码
#include "util.hpp"
#include "Task.hpp"
#include "Threadpool.hpp"
#include "daemonize.hpp"
#include <signal.h>
#include <sys/types.h>
#include <sys/wait.h>
#include <pthread.h>
class ServerTcp; // 申明一下ServerTcp
// 大小写转化服务
// TCP && UDP: 支持全双工
void transService(int sock, const std::string &clientIp, uint16_t clientPort)
{
assert(sock >= 0);
assert(!clientIp.empty());
assert(clientPort >= 1024);
char inbuffer[BUFFER_SIZE];
while (true)
{
ssize_t s = read(sock, inbuffer, sizeof(inbuffer) - 1); //我们认为我们读到的都是字符串
if (s > 0)
{
// read success
inbuffer[s] = '\0';
if (strcasecmp(inbuffer, "quit") == 0)
{
logMessage(DEBUG, "client quit -- %s[%d]", clientIp.c_str(), clientPort);
break;
}
logMessage(DEBUG, "trans before: %s[%d]>>> %s", clientIp.c_str(), clientPort, inbuffer);
// 可以进行大小写转化了
for (int i = 0; i < s; i++)
{
if (isalpha(inbuffer[i]) && islower(inbuffer[i]))
inbuffer[i] = toupper(inbuffer[i]);
}
logMessage(DEBUG, "trans after: %s[%d]>>> %s", clientIp.c_str(), clientPort, inbuffer);
write(sock, inbuffer, strlen(inbuffer));
}
else if (s == 0)
{
// pipe: 读端一直在读,写端不写了,并且关闭了写端,读端会如何?s == 0,代表对端关闭
// s == 0: 代表对方关闭,client 退出
logMessage(DEBUG, "client quit -- %s[%d]", clientIp.c_str(), clientPort);
break;
}
else
{
logMessage(DEBUG, "%s[%d] - read: %s", clientIp.c_str(), clientPort, strerror(errno));
break;
}
}
// 只要走到这里,一定是client退出了,服务到此结束
close(sock); // 如果一个进程对应的文件fd,打开了没有被归还,文件描述符泄漏!
logMessage(DEBUG, "server close %d done", sock);
}
//系统命令行
void execCommand(int sock, const std::string &clientIp, uint16_t clientPort)
{
assert(sock >= 0);
assert(!clientIp.empty());
assert(clientPort >= 1024);
char command[BUFFER_SIZE];
while (true)
{
ssize_t s = read(sock, command, sizeof(command) - 1); //我们认为我们读到的都是字符串
if (s > 0)
{
command[s] = '\0';
logMessage(DEBUG, "[%s:%d] exec [%s]", clientIp.c_str(), clientPort, command);
// 考虑安全
std::string safe = command;
if((std::string::npos != safe.find("rm")) || (std::string::npos != safe.find("unlink")))
{
break;
}
// 我们是以r方式打开的文件,没有写入
// 所以我们无法通过dup的方式得到对应的结果
FILE *fp = popen(command, "r");
if(fp == nullptr)
{
logMessage(WARINING, "exec %s failed, beacuse: %s", command, strerror(errno));
break;
}
char line[1024];
while(fgets(line, sizeof(line)-1, fp) != nullptr)
{
write(sock, line, strlen(line));
}
// dup2(fd, 1);
// dup2(sock, fp->_fileno);
// fflush(fp);
pclose(fp);
logMessage(DEBUG, "[%s:%d] exec [%s] ... done", clientIp.c_str(), clientPort, command);
}
else if (s == 0)
{
// pipe: 读端一直在读,写端不写了,并且关闭了写端,读端会如何?s == 0,代表对端关闭
// s == 0: 代表对方关闭,client 退出
logMessage(DEBUG, "client quit -- %s[%d]", clientIp.c_str(), clientPort);
break;
}
else
{
logMessage(DEBUG, "%s[%d] - read: %s", clientIp.c_str(), clientPort, strerror(errno));
break;
}
}
// 只要走到这里,一定是client退出了,服务到此结束
close(sock); // 如果一个进程对应的文件fd,打开了没有被归还,文件描述符泄漏!
logMessage(DEBUG, "server close %d done", sock);
}
class ThreadData
{
public:
uint16_t clientPort_;
std::string clinetIp_;
int sock_;
ServerTcp *this_;
public:
ThreadData(uint16_t port, std::string ip, int sock, ServerTcp *ts)
: clientPort_(port), clinetIp_(ip), sock_(sock), this_(ts)
{
}
};
class ServerTcp
{
public:
ServerTcp(uint16_t port, const std::string &ip = "")
: port_(port),
ip_(ip),
listenSock_(-1),
tp_(nullptr)
{
}
~ServerTcp()
{
}
public:
void init()
{
// 1. 创建socket
listenSock_ = socket(PF_INET, SOCK_STREAM, 0);
if (listenSock_ < 0)
{
logMessage(FATAL, "socket: %s", strerror(errno));
exit(SOCKET_ERR);
}
logMessage(DEBUG, "socket: %s, %d", strerror(errno), listenSock_);
// 2. bind绑定
// 2.1 填充服务器信息
struct sockaddr_in local; // 用户栈
memset(&local, 0, sizeof local);
local.sin_family = PF_INET;
local.sin_port = htons(port_);
ip_.empty() ? (local.sin_addr.s_addr = INADDR_ANY) : (inet_aton(ip_.c_str(), &local.sin_addr));
// 2.2 本地socket信息,写入sock_对应的内核区域
if (bind(listenSock_, (const struct sockaddr *)&local, sizeof local) < 0)
{
logMessage(FATAL, "bind: %s", strerror(errno));
exit(BIND_ERR);
}
logMessage(DEBUG, "bind: %s, %d", strerror(errno), listenSock_);
// 3. 监听socket,为何要监听呢?tcp是面向连接的!
if (listen(listenSock_, 5 /*后面再说*/) < 0)
{
logMessage(FATAL, "listen: %s", strerror(errno));
exit(LISTEN_ERR);
}
logMessage(DEBUG, "listen: %s, %d", strerror(errno), listenSock_);
// 运行别人来连接你了
// 4. 加载线程池
tp_ = ThreadPool<Task>::getInstance();
}
// static void *threadRoutine(void *args)
// {
// pthread_detach(pthread_self()); //设置线程分离
// ThreadData *td = static_cast<ThreadData *>(args);
// td->this_->transService(td->sock_, td->clinetIp_, td->clientPort_);
// delete td;
// return nullptr;
// }
void loop()
{
// signal(SIGCHLD, SIG_IGN); // only Linux
tp_->start();
logMessage(DEBUG, "thread pool start success, thread num: %d", tp_->threadNum());
while (true)
{
struct sockaddr_in peer;
socklen_t len = sizeof(peer);
// 4. 获取连接, accept 的返回值是一个新的socket fd ??
// 4.1 listenSock_: 监听 && 获取新的链接-> sock
// 4.2 serviceSock: 给用户提供新的socket服务
int serviceSock = accept(listenSock_, (struct sockaddr *)&peer, &len);
if (serviceSock < 0)
{
// 获取链接失败
logMessage(WARINING, "accept: %s[%d]", strerror(errno), serviceSock);
continue;
}
// 4.1 获取客户端基本信息
uint16_t peerPort = ntohs(peer.sin_port);
std::string peerIp = inet_ntoa(peer.sin_addr);
logMessage(DEBUG, "accept: %s | %s[%d], socket fd: %d",
strerror(errno), peerIp.c_str(), peerPort, serviceSock);
// 5 提供服务, echo -> 小写 -> 大写
// 5.0 v0 版本 -- 单进程 -- 一旦进入transService,主执行流,就无法进行向后执行,只能提供完毕服务之后才能进行accept
// transService(serviceSock, peerIp, peerPort);
// 5.1 v1 版本 -- 多进程版本 -- 父进程打开的文件会被子进程继承吗?会的
// pid_t id = fork();
// assert(id != -1);
// if(id == 0)
// {
// close(listenSock_); //建议
// //子进程
// transService(serviceSock, peerIp, peerPort);
// exit(0); // 进入僵尸
// }
// // 父进程
// close(serviceSock); //这一步是一定要做的!
// 5.1 v1.1 版本 -- 多进程版本 -- 也是可以的
// 爷爷进程
// pid_t id = fork();
// if(id == 0)
// {
// // 爸爸进程
// close(listenSock_);//建议
// // 又进行了一次fork,让 爸爸进程
// if(fork() > 0) exit(0);
// // 孙子进程 -- 就没有爸爸 -- 孤儿进程 -- 被系统领养 -- 回收问题就交给了系统来回收
// transService(serviceSock, peerIp, peerPort);
// exit(0);
// }
// // 父进程
// close(serviceSock); //这一步是一定要做的!
// // 爸爸进程直接终止,立马得到退出码,释放僵尸进程状态
// pid_t ret = waitpid(id, nullptr, 0); //就用阻塞式
// assert(ret > 0);
// (void)ret;
// 5.2 v2 版本 -- 多线程
// 这里不需要进行关闭文件描述符吗??不需要啦
// 多线程是会共享文件描述符表的!
// ThreadData *td = new ThreadData(peerPort, peerIp, serviceSock, this);
// pthread_t tid;
// pthread_create(&tid, nullptr, threadRoutine, (void*)td);
// 5.3 v3 版本 --- 线程池版本
// 5.3.1 构建任务
// 5.3 v3.1
// Task t(serviceSock, peerIp, peerPort, std::bind(&ServerTcp::transService, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
// tp_->push(t);
// 5.3 v3.2
// Task t(serviceSock, peerIp, peerPort, transService);
// tp_->push(t);
// 5.3 v3.3
Task t(serviceSock, peerIp, peerPort, execCommand);
tp_->push(t);
// waitpid(); 默认是阻塞等待!WNOHANG
// 方案1
// logMessage(DEBUG, "server 提供 service start ...");
// sleep(1);
}
}
private:
// sock
int listenSock_;
// port
uint16_t port_;
// ip
std::string ip_;
// 引入线程池
ThreadPool<Task> *tp_;
};
static void Usage(std::string proc)
{
std::cerr << "Usage:\n\t" << proc << " port ip" << std::endl;
std::cerr << "example:\n\t" << proc << " 8080 127.0.0.1\n"
<< std::endl;
}
// ./ServerTcp local_port local_ip
int main(int argc, char *argv[])
{
if (argc != 2 && argc != 3)
{
Usage(argv[0]);
exit(USAGE_ERR);
}
uint16_t port = atoi(argv[1]);
std::string ip;
if (argc == 3)
ip = argv[2];
daemonize(); // 我们的进程就会成为守护进程
ServerTcp svr(port, ip);
svr.init();
svr.loop();
return 0;
}
说明
头文件包 util.hpp
代码
#pragma once
#include <iostream>
#include <string>
#include <cstring>
#include <cstdlib>
#include <cassert>
#include <ctype.h>
#include <unistd.h>
#include <strings.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include "log.hpp"
#define SOCKET_ERR 1
#define BIND_ERR 2
#define LISTEN_ERR 3
#define USAGE_ERR 4
#define CONN_ERR 5
#define BUFFER_SIZE 1024
Makefile
.PHONY:all
all:TCPClient TCPServer
TCPClient:TCPClient.cc
g++ -o $@ $^ -std=c++11 -lpthread
TCPServer:TCPServer.cc
g++ -o $@ $^ -std=c++11 -lpthread
.PHONY:clean
clean:
rm -f TCPClient TCPServer serverTcp.log