【网络】套接字(Socket)编程——TCP版

🌈 个人主页： 小小、码农的 CSDN 博客 🔥 系列方向： linux网络 💪 学习宣言： 行动胜于空谈，实践出真知

上一篇我们已经完整实现了 UDP Socket 编程。

UDP Server 的核心流程比较简单：

socket()
↓
bind()
↓
recvfrom()
↓
业务处理
↓
sendto()

但 TCP 和 UDP 有一个非常关键的区别：

TCP 是面向连接的。

因此 TCP 在正式通信之前，必须先建立连接。

服务器和客户端的流程也就变成了：

TCP Server

socket()
↓
bind()
↓
listen()
↓
accept()
↓
read() / write()

TCP Client

socket()
↓
connect()
↓
read() / write()

这一篇就沿着这两条主线真正写出一个 TCP Server，并在此基础上一步步解决：

单客户端
↓
多客户端
↓
多进程
↓
多线程
↓
线程池
↓
守护进程

最后再回到 TCP 本身，从内核角度理解：

连接到底是什么？
accept 和三次握手是什么关系？
为什么 TCP 是全双工？
read/write 到底在干什么？
为什么 TCP 必须解决消息边界问题即粘包问题？

1. 先把最基本的 TCP 通信跑起来

1.1 程序结构

和 UDP 版本一样，我们先建立几个最基本的文件：

TcpServer
├── server.hpp
├── server.cc
├── client.hpp
├── client.cc
└── Makefile

最终实现的仍然是最简单的 Echo Server：

Client
↓
发送 hello
↓
Server
↓
收到 hello
↓
原样返回 hello
↓
Client

先准备 Makefile：

.PHONY:all
all:server client

server:server.cc
g++ -o $@ $^ -std=c++11

client:client.cc
g++ -o $@ $^ -std=c++11

.PHONY:clean
clean:
rm -rf server client

1.2 Server：socket 和 bind

首先创建服务器 Socket。

UDP 中：

socket(AF_INET, SOCK_DGRAM, 0);

TCP 中：

socket(AF_INET, SOCK_STREAM, 0);

区别就在第二个参数：

SOCK_DGRAM
↓
UDP 数据报

SOCK_STREAM
↓
TCP 字节流

因此：

listen_sock_ = socket(AF_INET, SOCK_STREAM, 0);

创建的是一个 TCP Socket。

接着仍然需要绑定服务器自己的：

IP
+
Port

struct sockaddr_in local;
memset(&local, 0, sizeof(local));

local.sin_family = AF_INET;
local.sin_addr.s_addr = inet_addr(ip_.c_str());
local.sin_port = htons(port_);

bind(listen_sock_, (const struct sockaddr*)&local, sizeof(local));

这里和 UDP Server 基本一致。

端口仍然需要：

htons();

转换为网络字节序。

到这里：

UDP Server

已经可以直接等待数据了。

但 TCP 不行。

因为 TCP 是：

面向连接的。

服务器还必须告诉操作系统：

我这个 Socket 接下来不是主动连接别人，而是专门等待别人来连接我。

这就是：

listen();

1.3 listen：把 Socket 变成监听 Socket

函数原型：

#include <sys/socket.h>

int listen(int sockfd, int backlog);

使用：

listen(listen_sock_, backlog);

第一个参数：

sockfd

就是之前：

socket();

创建并通过：

bind();

绑定好的 Socket。

listen() 做完以后，这个 Socket 就从一个普通 TCP Socket 变成了：

监听 Socket。

它以后主要负责：

等待连接
+
获取连接

而不是直接和客户端收发业务数据。

backlog 到底是什么？

我们这一篇暂时不用继续深入，只需要记住：

backlog
↓
等待 accept 的连接队列长度

另外：

listen();

本身通常不会阻塞等待客户端。

真正负责：

没有连接就一直等。

的是下面的：

accept();

1.4 accept：真正获取一个连接

函数原型：

int accept(int sockfd, struct sockaddr* addr, socklen_t* addrlen);

使用：

struct sockaddr_in client;
socklen_t len = sizeof(client);

int service_sock = accept(listen_sock_, (struct sockaddr*)&client, &len);

在默认阻塞模式下，如果当前没有已经建立好的连接：

accept()
↓
阻塞等待

一旦有客户端成功建立 TCP 连接：

accept()
↓
返回

但这里最重要的不是阻塞。

而是它的返回值：

service_sock

为什么服务器需要两个 Socket？

这是 TCP Socket 编程中最重要的地方之一。

一开始：

listen_sock_ = socket(...);

得到的是：

监听 Socket

它负责：

listen
accept

但是：

accept();

成功以后会返回：

service_sock

这个才是：

专门和当前客户端通信的 Socket。

所以服务器实际上是：

┌── service_sock A ↔ Client A
│
listen_sock ─accept─┼── service_sock B ↔ Client B
│
└── service_sock C ↔ Client C

假设有三个客户端连接服务器。

服务器一共会存在：

1 个 listen_sock
+
3 个 service_sock

监听 Socket 继续等待新连接。

每一个 accept() 返回的 Socket，则负责和对应客户端通信。

所以这两个 Socket 的职责一定不能混：

listen_sock
↓
只负责获取新连接

service_sock
↓
真正负责 read/write

accept 还能拿到客户端信息

和 UDP 的：

recvfrom();

类似，accept() 的：

addr
addrlen

也能够把客户端地址带回来。

例如：

uint16_t clientport = ntohs(client.sin_port);

char clientip[32];
inet_ntop(AF_INET, &client.sin_addr, clientip, sizeof(clientip));

于是服务器能够知道：

哪个 IP
+
哪个 Port

建立了连接。

1.5 当前阶段完整 Server

到这里，我们已经把 TCP Server 建立连接之前需要的几个核心接口学完了：

socket()
↓
创建 TCP Socket
↓
bind()
↓
绑定服务器 IP + Port
↓
listen()
↓
进入监听状态
↓
accept()
↓
获取已经建立好的 TCP 连接

此时这个 Server 暂时还不会和客户端进行数据通信，它只完成一件事：

等待客户端建立连接，并打印客户端的 IP、Port 以及 accept() 返回的通信 Socket。

当前项目只有三个文件：

TcpServer
├── server.hpp
├── server.cc
└── Makefile

server.hpp

#pragma once

#include <iostream>
#include <string>
#include <cstring>
#include <cerrno>
#include <cstdlib>
#include <unistd.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>

namespace fxs_server
{
const uint16_t default_port = 8877;
const std::string default_ip = "0.0.0.0";
const int backlog = 5;

enum
{
USAGE_ERR = 1,
SOCKET_ERR,
BIND_ERR,
LISTEN_ERR
};

class TcpServer
{
public:
TcpServer(uint16_t port = default_port, const std::string& ip = default_ip)
: ip_(ip), port_(port)
{}

~TcpServer()
{}

void InitServer()
{
// 1.创建TCP监听套接字
listen_sock_ = socket(AF_INET, SOCK_STREAM, 0);

if(listen_sock_ < 0)
{
std::cerr << "Create Socket Fail: "
<< strerror(errno) << std::endl;
exit(SOCKET_ERR);
}

std::cout << "Create Socket Success: "
<< listen_sock_ << std::endl;

// 2.绑定IP和Port
struct sockaddr_in local;
memset(&local, 0, sizeof(local));

local.sin_family = AF_INET;
local.sin_port = htons(port_);
local.sin_addr.s_addr = inet_addr(ip_.c_str());

if(bind(listen_sock_, (const struct sockaddr*)&local, sizeof(local)) < 0)
{
std::cerr << "Bind IP&&Port Fail: "
<< strerror(errno) << std::endl;
exit(BIND_ERR);
}

std::cout << "Bind IP&&Port Success" << std::endl;

// 3.进入监听状态
if(listen(listen_sock_, backlog) < 0)
{
std::cerr << "Listen Fail: "
<< strerror(errno) << std::endl;
exit(LISTEN_ERR);
}

std::cout << "Listen Success" << std::endl;
}

void StartServer()
{
while(true)
{
// 4.获取已经建立好的连接
struct sockaddr_in client;
socklen_t len = sizeof(client);

int service_sock = accept(listen_sock_, (struct sockaddr*)&client, &len);

if(service_sock < 0)
{
std::cerr << "Accept Fail: "
<< strerror(errno) << std::endl;
continue;
}

// 5.获取客户端IP和Port
char clientip[32];
inet_ntop(AF_INET, &client.sin_addr, clientip, sizeof(clientip));

uint16_t clientport = ntohs(client.sin_port);

std::cout << "Get a new connection -> "
<< "[" << clientip << ":" << clientport << "] "
<< "service_sock: " << service_sock
<< std::endl;

// 当前阶段还没有学习数据通信，
// 所以这里只验证accept能够成功获取连接
close(service_sock);
}
}

private:
int listen_sock_;
std::string ip_;
uint16_t port_;
};
}

server.cc

#include <iostream>
#include <memory>
#include "server.hpp"

using namespace std;
using namespace fxs_server;

void Usage(const char* program)
{
cout << "Usage:" << endl;
cout << "\\t" << program << " ServerPort" << endl;
}

int main(int argc, char* argv[])
{
if(argc != 2)
{
Usage(argv[0]);
return USAGE_ERR;
}

uint16_t port = stoi(argv[1]);

unique_ptr<TcpServer> server(new TcpServer(port));

server->InitServer();
server->StartServer();

return 0;
}

编译：

make

运行：

./server 8877

此时可以看到：

Create Socket Success: …
Bind IP&&Port Success
Listen Success

随后程序停在那里。

这并不是程序卡死了，而是：

accept();

在默认阻塞模式下发现当前还没有已经建立好的连接，因此正在等待客户端。

现在，我们就可以真正找一个 TCP 客户端来验证它。

1.6 使用 telnet 测试 TCP 连接

目前我们还没有编写自己的 TCP Client。

但是为了验证：

socket
bind
listen
accept

这一整套流程是否已经正常工作，可以暂时使用：

telnet

充当 TCP Client。

如果 Ubuntu 中还没有安装：

sudo apt install telnet

首先保持 Server 正在运行：

./server 8877

然后再打开一个终端：

telnet 127.0.0.1 8877

此时客户端会主动向：

127.0.0.1:8877

建立 TCP 连接。

Server 中原本阻塞的：

accept();

就会立即返回，并打印类似：

Get a new connection -> [127.0.0.1:xxxxx] service_sock: 4

其中：

127.0.0.1
↓
客户端IP

xxxxx
↓
客户端由操作系统自动分配的临时Port

4
↓
accept返回的通信Socket文件描述符

整个过程：

Server
↓
socket()
↓
bind()
↓
listen()
↓
accept()
↓
等待……

Client
↓
telnet 127.0.0.1 8877
↓
TCP建立连接
↓

Server accept()返回
↓
获得client IP + Port
↓
获得service_sock

这里 telnet 建立连接以后很快又会被 Server 断开，这是正常现象。

因为当前代码：

accept();

成功以后只做了：

打印连接信息
↓
close(service_sock)

我们目前只是为了验证：

TCP Server 已经能够成功监听并获取连接。

下一步学习：

read();
write();

以后，服务器才会真正利用这个 service_sock 和客户端进行数据通信。

1.7 read 和 write：真正开始 TCP 通信

连接建立完成以后，接下来就是通信。

Linux 中：

Socket 也是文件。

所以我们完全可以直接使用以前学习文件时的：

read();
write();

读取和发送网络数据。

服务器业务函数：

void Service(int sock, const std::string& clientip, const uint16_t& clientport)
{
char buff[1024];
std::string who = clientip + "-" + std::to_string(clientport);

while(true)
{
ssize_t n = read(sock, buff, sizeof(buff) – 1);

if(n > 0)
{
buff[n] = '\\0';

std::cout << "Server get: "
<< buff
<< " from "
<< who
<< std::endl;

std::string respond = func_(buff);

write(sock, respond.c_str(), respond.size());
}
else if(n == 0)
{
std::cout << "Client "
<< who
<< " quit!"
<< std::endl;

close(sock);
break;
}
else
{
std::cerr << "Read Fail! "
<< strerror(errno)
<< std::endl;

close(sock);
break;
}
}
}

这里：

read();

一共有三种结果。

read > 0

表示：

真正读取到了数据

返回值：

n

就是实际读取的字节数。

因此：

buff[n] = '\\0';

把它临时当 C 风格字符串使用。

read == 0

这个特别重要。

对于 TCP Socket：

read() == 0

不是：

对方给我发送了一个长度为 0 的普通消息。

而意味着：

TCP 字节流已经到达 EOF，通常说明对端已经有序关闭了这一方向的连接。

最常见的情况就是：

客户端 close()
↓
TCP 通知服务器
↓
服务器 read()
↓
返回 0

于是服务器知道：

客户端退出了。

这也是 TCP 和 UDP 很明显的一个区别。

UDP 没有连接状态，所以对端程序退出时：

Server

并不会自动知道。

TCP 因为维护连接，所以：

read() == 0

可以让我们感知对端关闭。

read < 0

表示：

读取发生错误

此时：

errno

保存错误原因。

1.8 Client：socket 和 connect

TCP Client 初始化仍然先：

sock_ = socket(AF_INET, SOCK_STREAM, 0);

客户端一般不需要像 Server 一样手动：

bind();

因为：

客户端只要主动发起连接，操作系统就可以自动给它选择合适的本地 IP 和临时端口。

所以客户端真正新增的核心接口是：

connect();

函数原型：

int connect(int sockfd, const struct sockaddr* addr, socklen_t addrlen);

使用：

struct sockaddr_in server;
memset(&server, 0, sizeof(server));

server.sin_family = AF_INET;
inet_aton(server_ip_.c_str(), &server.sin_addr);
server.sin_port = htons(server_port_);

connect(sock_, (const struct sockaddr*)&server, sizeof(server));

这里：

bind()

传的是：

自己的地址

而：

connect()

传的是：

对方 Server 的地址

connect 成功到底意味着什么？

这个地方必须说清楚。

调用：

connect();

以后，真正负责建立 TCP 连接的不是我们的 C++ 代码。

而是：

客户端和服务器两边操作系统中的 TCP 协议栈。

connect() 会触发 TCP 建连流程。

当三次握手成功以后：

connect();

就可以成功返回。

这里并不要求服务器应用层已经执行：

accept();

因为：

三次握手
↓
TCP 内核协议栈完成

accept()
↓
应用层从内核已经完成的连接中取走一个

这是两层完全不同的事情。

所以以后一定要记住：

connect 对应“主动发起连接”，accept 对应“拿走已经建立好的连接结果”，accept 并不负责完成三次握手。

连接失败重试

服务器可能暂时没有启动。

如果：

connect();

失败就立即退出，体验并不好。

所以可以简单尝试数次：

int cnt = 5;

while(cnt)
{
int ret = connect(sock_, (const struct sockaddr*)&server, sizeof(server));

if(ret == 0)
break;

std::cerr << "网络异常，正在进行重连… 剩余连接次数: "
<< —cnt
<< std::endl;

sleep(1);
}

if(cnt == 0)
{
std::cerr << "连接失败! "
<< strerror(errno)
<< std::endl;

close(sock_);
exit(CONNECT_ERR);
}

1.9 Client 获取服务

连接成功后：

Client

同样使用：

write();
read();

通信。

void GetService()
{
char buff[1024];
std::string who = server_ip_ + "-" + std::to_string(server_port_);

while(true)
{
std::string msg;

std::cout << "Please Enter >> ";
std::getline(std::cin, msg);

write(sock_, msg.c_str(), msg.size());

ssize_t n = read(sock_, buff, sizeof(buff) – 1);

if(n > 0)
{
buff[n] = '\\0';

std::cout << "Client get: "
<< buff
<< " from "
<< who
<< std::endl;
}
else if(n == 0)
{
std::cout << "Server "
<< who
<< " quit!"
<< std::endl;

close(sock_);
break;
}
else
{
std::cerr << "Read Fail! "
<< strerror(errno)
<< std::endl;

close(sock_);
break;
}
}
}

于是一个完整请求：

Client getline()
↓
write()
↓
Server read()
↓
业务处理
↓
Server write()
↓
Client read()

基本 TCP Echo 就打通了。

1.10 基础版完整代码

server.hpp

#pragma once

#include <iostream>
#include <string>
#include <functional>
#include <cstring>
#include <cerrno>
#include <cstdlib>
#include <unistd.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>

namespace fxs_server
{
const uint16_t default_port = 8877;
const std::string default_ip = "0.0.0.0";
const int backlog = 5;

enum
{
USAGE_ERR = 1,
SOCKET_ERR,
BIND_ERR,
LISTEN_ERR
};

using func_t = std::function<std::string(std::string)>;

class TcpServer
{
public:
TcpServer(const func_t& func, const uint16_t port = default_port, const std::string ip = default_ip)
: ip_(ip), port_(port), func_(func)
{}

~TcpServer()
{}

void InitServer()
{
// 1.创建监听套接字
listen_sock_ = socket(AF_INET, SOCK_STREAM, 0);

if(listen_sock_ == –1)
{
std::cerr << "Create Socket Fail! "
<< strerror(errno)
<< std::endl;

exit(SOCKET_ERR);
}

std::cout << "Create Socket Success! "
<< listen_sock_
<< std::endl;

// 2.绑定IP和Port
struct sockaddr_in local;
memset(&local, 0, sizeof(local));

local.sin_family = AF_INET;
local.sin_addr.s_addr = inet_addr(ip_.c_str());
local.sin_port = htons(port_);

if(bind(listen_sock_, (const struct sockaddr*)&local, sizeof(local)) < 0)
{
std::cerr << "Bind IP&&Port Fail: "
<< strerror(errno)
<< std::endl;

exit(BIND_ERR);
}

// 3.进入监听状态
if(listen(listen_sock_, backlog) < 0)
{
std::cerr << "Listen Fail: "
<< strerror(errno)
<< std::endl;

exit(LISTEN_ERR);
}
}

void StartServer()
{
while(true)
{
// 1.获取新连接
struct sockaddr_in client;
socklen_t len = sizeof(client);

int service_sock = accept(listen_sock_, (struct sockaddr*)&client, &len);

if(service_sock < 0)
{
std::cerr << "Accept Fail: "
<< strerror(errno)
<< std::endl;
continue;
}

// 2.获得客户端信息
uint16_t clientport = ntohs(client.sin_port);

char clientip[32];
inet_ntop(AF_INET, &client.sin_addr, clientip, sizeof(clientip));

std::cout << "Server accept "
<< clientip
<< ":"
<< clientport
<< ", service sock: "
<< service_sock
<< std::endl;

// 3.提供服务
Service(service_sock, clientip, clientport);
}
}

void Service(int sock, const std::string& clientip, const uint16_t& clientport)
{
char buff[1024];
std::string who = clientip + "-" + std::to_string(clientport);

while(true)
{
ssize_t n = read(sock, buff, sizeof(buff) – 1);

if(n > 0)
{
buff[n] = '\\0';

std::cout << "Server get: "
<< buff
<< " from "
<< who
<< std::endl;

std::string respond = func_(buff);

write(sock, respond.c_str(), respond.size());
}
else if(n == 0)
{
std::cout << "Client "
<< who
<< " quit!"
<< std::endl;

close(sock);
break;
}
else
{
std::cerr << "Read Fail! "
<< strerror(errno)
<< std::endl;

close(sock);
break;
}
}
}

private:
int listen_sock_;
std::string ip_;
uint16_t port_;
func_t func_;
};
}

server.cc

#include <iostream>
#include <memory>
#include "server.hpp"

using namespace std;
using namespace fxs_server;

std::string Echo(const std::string& request)
{
return request;
}

void Usage(const char* program)
{
cout << "Usage:" << endl;
cout << "\\t" << program << " ServerPort" << endl;
}

int main(int argc, char* argv[])
{
if(argc != 2)
{
Usage(argv[0]);
return USAGE_ERR;
}

uint16_t port = stoi(argv[1]);

unique_ptr<TcpServer> server(new TcpServer(Echo, port));

server->InitServer();
server->StartServer();

return 0;
}

client.hpp

#pragma once

#include <iostream>
#include <string>
#include <cstring>
#include <cerrno>
#include <cstdlib>
#include <unistd.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>

namespace fxs_client
{
enum
{
USAGE_ERR = 1,
SOCKET_ERR,
CONNECT_ERR
};

class TcpClient
{
public:
TcpClient(const std::string& ip, const uint16_t port)
: server_ip_(ip), server_port_(port)
{}

~TcpClient()
{}

void InitClient()
{
sock_ = socket(AF_INET, SOCK_STREAM, 0);

if(sock_ == –1)
{
std::cerr << "Create Socket Fail! "
<< strerror(errno)
<< std::endl;

exit(SOCKET_ERR);
}

std::cout << "Create Socket Success! "
<< sock_
<< std::endl;
}

void StartClient()
{
struct sockaddr_in server;
memset(&server, 0, sizeof(server));

server.sin_family = AF_INET;
inet_aton(server_ip_.c_str(), &server.sin_addr);
server.sin_port = htons(server_port_);

int cnt = 5;

while(cnt)
{
int ret = connect(sock_, (const struct sockaddr*)&server, sizeof(server));

if(ret == 0)
break;

std::cerr << "网络异常，正在进行重连… 剩余连接次数: "
<< —cnt
<< std::endl;

sleep(1);
}

if(cnt == 0)
{
std::cerr << "连接失败! "
<< strerror(errno)
<< std::endl;

close(sock_);
exit(CONNECT_ERR);
}

std::cout << "连接成功!" << std::endl;

GetService();
}

void GetService()
{
char buff[1024];
std::string who = server_ip_ + "-" + std::to_string(server_port_);

while(true)
{
std::string msg;

std::cout << "Please Enter >> ";
std::getline(std::cin, msg);

write(sock_, msg.c_str(), msg.size());

ssize_t n = read(sock_, buff, sizeof(buff) – 1);

if(n > 0)
{
buff[n] = '\\0';

std::cout << "Client get: "
<< buff
<< " from "
<< who
<< std::endl;
}
else if(n == 0)
{
std::cout << "Server "
<< who
<< " quit!"
<< std::endl;

close(sock_);
break;
}
else
{
std::cerr << "Read Fail! "
<< strerror(errno)
<< std::endl;

close(sock_);
break;
}
}
}

private:
int sock_;
std::string server_ip_;
uint16_t server_port_;
};
}

client.cc

#include <iostream>
#include <memory>
#include "client.hpp"

using namespace std;
using namespace fxs_client;

void Usage(const char* program)
{
cout << "Usage:" << endl;
cout << "\\t" << program << " ServerIP ServerPort" << endl;
}

int main(int argc, char* argv[])
{
if(argc != 3)
{
Usage(argv[0]);
return USAGE_ERR;
}

std::string ip = argv[1];
uint16_t port = stoi(argv[2]);

unique_ptr<TcpClient> client(new TcpClient(ip, port));

client->InitClient();
client->StartClient();

return 0;
}

1.9 运行测试

编译：

make

启动服务器：

./server 8877

启动客户端：

./client 127.0.0.1 8877

客户端输入：

hello tcp

应该得到：

Client
↓
hello tcp
↓
Server
↓
hello tcp
↓
Client

还可以查看当前 TCP 监听端口：

ss -ntlp | grep 8877

netsat -nltp | grep 8877

2. 单进程服务器为什么不能同时服务多个客户端？

现在打开两个 Client：

Client A
Client B

A 先连接。

Server 执行：

accept();

获得：

A 的 service_sock

然后马上进入：

Service();

而：

Service();

内部是：

while(true)

不断为 A 服务。

因此服务器此时的执行流：

accept A
↓
Service A
↓
Service A
↓
Service A
↓
……

根本回不到下一次：

accept();

所以 B 无法被应用层处理。

为什么 B 的 connect 却可能显示成功？

这个地方特别容易产生误解。

B：

connect();

成功，只说明：

客户端和服务器内核中的 TCP 三次握手成功了。

并不代表：

服务器应用代码已经调用 accept() 并开始处理 B。

因此完全可能出现：

Client B
↓
connect 成功

Server 应用
↓
还在 Service(A)

B 对应的连接已经在：

内核

中建立完成，并等待：

accept();

取走。

所以：

connect 成功
≠
服务器业务代码已经开始服务你

这正好也说明：

连接建立属于内核 TCP 协议栈，而 accept 属于应用层获取连接。

问题真正出在：

接受新连接

和：

为旧连接提供服务

使用了同一个执行流。

所以接下来我们必须：

让“获取连接”和“处理连接”同时进行。

第一种方法：

多进程

3. 多进程版 TCP Server

核心思路：

父进程
↓
只负责 accept

子进程
↓
负责 Service

每建立一个新连接：

accept()
↓
fork()
↓
┌───────────────┐
│ │
父进程 子进程
│ │
继续 accept Service

3.1 为什么子进程能够直接使用 service_sock？

因为：

fork();

以后，子进程会继承父进程的：

文件描述符表。

所以父进程刚刚：

accept();

得到的：

service_sock

子进程同样能够访问。

假设：

父进程
fd 3 -> listen socket
fd 4 -> client socket

fork() 后：

父进程 子进程

fd 3 -> listen fd 3 -> listen
fd 4 -> client fd 4 -> client

但两边职责不同。

子进程：

只负责当前 Client

所以不需要：

listen_sock

应该：

close(listen_sock_);

父进程：

只负责继续 accept

所以不需要这个客户端的：

service_sock

应该：

close(service_sock);

注意：

这里父进程 close(service_sock) 并不会立即断开子进程和客户端的连接。

因为 fork() 后父子进程的文件描述符虽然是两份表项，但都引用内核中的同一个打开文件对象。

只有相关引用全部关闭，底层连接才真正失去引用。

3.2 最直接的多进程版本

void StartServer()
{
while(true)
{
struct sockaddr_in client;
socklen_t len = sizeof(client);

int service_sock = accept(listen_sock_, (struct sockaddr*)&client, &len);

if(service_sock < 0)
continue;

uint16_t clientport = ntohs(client.sin_port);

char clientip[32];
inet_ntop(AF_INET, &client.sin_addr, clientip, sizeof(clientip));

pid_t id = fork();

if(id < 0)
{
close(service_sock);
continue;
}
else if(id == 0)
{
close(listen_sock_);

Service(service_sock, clientip, clientport);

exit(0);
}

close(service_sock);
}
}

这样：

Client A -> Child A
Client B -> Child B
Client C -> Child C

而父进程始终：

accept
accept
accept

服务器就具备并发能力了。

3.3 子进程退出后谁来回收？

这里又出现进程编程中的老问题：

僵尸进程。

子进程完成：

Service();

以后退出。

父进程如果完全不：

wait();
waitpid();

就可能产生僵尸进程。

那是不是直接：

waitpid(id, nullptr, 0);

就行？

也不行。

因为这是：

阻塞等待

父进程就会变成：

fork Client A
↓
wait Client A
↓
Client A 不退出
↓
父进程无法 accept B

又失去了并发意义。

所以我们需要：

既回收子进程，又不能阻塞父进程继续 accept。

3.4 四种处理思路

方法一：waitpid + WNOHANG

waitpid(id, nullptr, WNOHANG);

这样：

waitpid

本身不会阻塞。

但问题是：

如果调用 waitpid() 时子进程还没退出，它会直接返回。

以后子进程真正退出时，如果父进程正阻塞在：

accept();

就可能暂时没人回收。

所以只是简单在：

fork();

后调用一次：

waitpid(..., WNOHANG);

并不能彻底解决问题。

方法二：捕获 SIGCHLD

子进程退出时，父进程会收到：

SIGCHLD

所以可以注册信号处理函数：

static void Handler(int signo)
{
while(waitpid(–1, nullptr, WNOHANG) > 0)
{}
}

然后：

signal(SIGCHLD, Handler);

这样：

父进程
↓
正常 accept

子进程退出
↓
SIGCHLD
↓
Handler
↓
waitpid

就能做到异步回收。

注意：

信号处理函数中应该尽量只调用异步信号安全的接口，不要随便塞大量 cout、内存分配等复杂逻辑。

方法三：显式忽略 SIGCHLD

Linux 下还有一种非常简单的方法：

signal(SIGCHLD, SIG_IGN);

显式将：

SIGCHLD

设置为忽略以后，Linux 可以自动处理这些退出子进程，不让它们长期处于僵尸状态。

对于我们现在这个简单 TCP Server：

signal(SIGCHLD, SIG_IGN);

是最省事的写法。

因此最终：

void StartServer()
{
signal(SIGCHLD, SIG_IGN);

while(true)
{
struct sockaddr_in client;
socklen_t len = sizeof(client);

int service_sock = accept(listen_sock_, (struct sockaddr*)&client, &len);

if(service_sock < 0)
continue;

uint16_t clientport = ntohs(client.sin_port);

char clientip[32];
inet_ntop(AF_INET, &client.sin_addr, clientip, sizeof(clientip));

pid_t id = fork();

if(id < 0)
{
close(service_sock);
continue;
}
else if(id == 0)
{
close(listen_sock_);

Service(service_sock, clientip, clientport);

exit(0);
}

close(service_sock);
}
}

方法四：双重 fork

还有一种经典技巧：

父进程
↓
fork 子进程
↓
子进程再 fork 孙子进程
↓
子进程立即退出
↓
父进程只等待这个很快退出的子进程
↓
孙子进程真正 Service

原来的父子关系被切断以后，孙子进程会由系统中的父进程接管机制负责。

这种方式能够绕开父进程长期等待业务进程的问题。

不过：

对我们当前 TCP Server 来说，没有必要专门绕这么一大圈。

知道这个方法即可。

后面讲守护进程时，还会再次看到：

fork
+
setsid
+
再次 fork

这种思路。

3.5 多进程版 TCP Server 完整代码

前面我们已经把多进程服务器中几个关键问题解决了：

父进程
↓
accept()
↓
获取新连接
↓
fork()

父进程
↓
关闭自己的 service_sock
↓
继续 accept()

子进程
↓
关闭 listen_sock
↓
Service()
↓
专门服务当前客户端

同时，子进程退出以后还需要解决：

僵尸进程

的问题。

在当前这个简单 TCP Server 中，我们采用：

signal(SIGCHLD, SIG_IGN);

让 Linux 自动处理退出的子进程。

因此，现在可以把前面的内容真正组合起来，得到一份完整的多进程 TCP Server。

当前代码结构：

TcpServer
├── server.hpp
├── server.cc
├── client.hpp
├── client.cc
└── Makefile

客户端本身不需要发生变化，仍然使用前面已经写好的 TCP Client。

这一阶段真正修改的是 Server。

server.hpp

#pragma once

#include <iostream>
#include <string>
#include <functional>
#include <cstring>
#include <cerrno>
#include <cstdlib>
#include <signal.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>

namespace fxs_server
{
const uint16_t default_port = 8877;
const std::string default_ip = "0.0.0.0";
const int backlog = 5;

enum
{
USAGE_ERR = 1,
SOCKET_ERR,
BIND_ERR,
LISTEN_ERR
};

using func_t = std::function<std::string(std::string)>;

class TcpServer
{
public:
TcpServer(const func_t& func, uint16_t port = default_port, const std::string& ip = default_ip)
: ip_(ip), port_(port), func_(func)
{}

~TcpServer()
{}

void InitServer()
{
// 1.创建监听套接字
listen_sock_ = socket(AF_INET, SOCK_STREAM, 0);

if(listen_sock_ < 0)
{
std::cerr << "Create Socket Fail: "
<< strerror(errno) << std::endl;
exit(SOCKET_ERR);
}

std::cout << "Create Socket Success: "
<< listen_sock_ << std::endl;

// 2.绑定IP和Port
struct sockaddr_in local;
memset(&local, 0, sizeof(local));

local.sin_family = AF_INET;
local.sin_port = htons(port_);
local.sin_addr.s_addr = inet_addr(ip_.c_str());

if(bind(listen_sock_, (const struct sockaddr*)&local, sizeof(local)) < 0)
{
std::cerr << "Bind IP&&Port Fail: "
<< strerror(errno) << std::endl;
exit(BIND_ERR);
}

std::cout << "Bind IP&&Port Success" << std::endl;

// 3.进入监听状态
if(listen(listen_sock_, backlog) < 0)
{
std::cerr << "Listen Fail: "
<< strerror(errno) << std::endl;
exit(LISTEN_ERR);
}

std::cout << "Listen Success" << std::endl;
}

void Service(int sock, const std::string& clientip, uint16_t clientport)
{
char buff[1024];
std::string who = clientip + ":" + std::to_string(clientport);

while(true)
{
ssize_t n = read(sock, buff, sizeof(buff) – 1);

if(n > 0)
{
buff[n] = '\\0';

std::cout << "Server get: "
<< buff
<< " from "
<< who
<< std::endl;

std::string respond = func_(buff);

write(sock, respond.c_str(), respond.size());
}
else if(n == 0)
{
std::cout << "Client "
<< who
<< " quit!"
<< std::endl;

close(sock);
break;
}
else
{
std::cerr << "Read Fail: "
<< strerror(errno)
<< std::endl;

close(sock);
break;
}
}
}

void StartServer()
{
// 子进程退出后由系统自动处理，避免产生僵尸进程
signal(SIGCHLD, SIG_IGN);

while(true)
{
// 1.父进程获取新连接
struct sockaddr_in client;
socklen_t len = sizeof(client);

int service_sock = accept(listen_sock_, (struct sockaddr*)&client, &len);

if(service_sock < 0)
{
std::cerr << "Accept Fail: "
<< strerror(errno)
<< std::endl;
continue;
}

// 2.获取客户端信息
char clientip[32];
inet_ntop(AF_INET, &client.sin_addr, clientip, sizeof(clientip));

uint16_t clientport = ntohs(client.sin_port);

std::cout << "Get a new connection -> "
<< "[" << clientip << ":" << clientport << "] "
<< "service_sock: " << service_sock
<< std::endl;

// 3.创建子进程
pid_t id = fork();

if(id < 0)
{
std::cerr << "Fork Fail: "
<< strerror(errno)
<< std::endl;

close(service_sock);
continue;
}
else if(id == 0)
{
// 子进程只负责当前客户端，不需要监听套接字
close(listen_sock_);

Service(service_sock, clientip, clientport);

exit(0);
}

// 父进程只负责继续获取新连接
close(service_sock);
}
}

private:
int listen_sock_;
std::string ip_;
uint16_t port_;
func_t func_;
};
}

这里最核心的其实就是：

pid_t id = fork();

之后父子进程职责彻底分开：

父进程
↓
close(service_sock)
↓
回到 accept()
↓
继续等待其他客户端

子进程
↓
close(listen_sock_)
↓
Service(service_sock)
↓
一直服务当前客户端

所以此时：

Client A
↓
Child Process A

Client B
↓
Child Process B

Client C
↓
Child Process C

父进程则始终专门负责：

accept();

这就解决了前面单进程版本：

Service(A)
↓
长期不返回
↓
无法继续 accept(B)

的问题。

server.cc

业务层仍然使用最简单的 Echo：

#include <iostream>
#include <memory>
#include "server.hpp"

using namespace std;
using namespace fxs_server;

std::string Echo(const std::string& request)
{
return request;
}

void Usage(const char* program)
{
cout << "Usage:" << endl;
cout << "\\t" << program << " ServerPort" << endl;
}

int main(int argc, char* argv[])
{
if(argc != 2)
{
Usage(argv[0]);
return USAGE_ERR;
}

uint16_t port = stoi(argv[1]);

unique_ptr<TcpServer> server(new TcpServer(Echo, port));

server->InitServer();
server->StartServer();

return 0;
}

可以看到，业务层完全没有因为服务器从：

单进程

改成：

多进程

而发生变化。

变化的只是：

服务器如何调度不同客户端。

运行多进程 Server

重新编译：

make clean
make

启动服务器：

./server 8877

然后分别打开多个终端。

Client A：

./client 127.0.0.1 8877

Client B：

./client 127.0.0.1 8877

Client C：

./client 127.0.0.1 8877

此时三个客户端都可以持续和服务器通信。

例如：

Client A
Please Enter >> hello A

Client B
Please Enter >> hello B

Client C
Please Enter >> hello C

三个客户端都能分别收到自己的 Echo。

这说明：

父进程
↓
仍然可以不断 accept()

多个子进程
↓
同时为不同客户端提供 Service()

查看 Server 的多个进程

此时再打开一个终端：

ps -axj | head -1 && ps -axj | grep 8877 | grep -v grep

观察父子关系

应该可以看到：

一个 Server 父进程
+
多个 Server 子进程

再观察一次文件描述符关系

这个版本还有一个特别值得复习的地方。

假设父进程：

listen_sock = 3
service_sock = 4

执行：

fork();

之后：

父进程 子进程

fd 3 fd 3
↓ ↓
listen Socket listen Socket

fd 4 fd 4
↓ ↓
Client Socket Client Socket

因为子进程继承了父进程的文件描述符表。

但是双方职责不同。

所以子进程：

close(listen_sock_);

只保留：

service_sock

用于服务客户端。

父进程：

close(service_sock);

只保留：

listen_sock

继续接受新连接。

最终：

Parent
│
└── listen_sock
│
├── accept A -> fork -> Child A -> service_sock A
│
├── accept B -> fork -> Child B -> service_sock B
│
└── accept C -> fork -> Child C -> service_sock C

这就是当前多进程 TCP Server 的完整结构。

3.6 多进程版本还有什么问题？

现在服务器确实已经能够同时服务多个客户端了。

但是它采用的是：

来一个客户端
↓
fork 一个子进程

再来一个客户端
↓
再 fork 一个子进程

进程的创建和维护成本并不低。

因为不同进程拥有各自独立的：

虚拟地址空间
页表
PCB
资源管理结构
……

而我们的需求其实只是：

让不同客户端由不同执行流同时处理。

未必真的需要：

多个独立进程

因此接下来可以考虑一种更轻量的执行流：

线程。

于是服务器继续演进：

单进程串行
↓
多进程并发
↓
多线程并发

4. 多线程版 TCP Server

多进程版本能够工作。

但每来一个客户端就：

fork();

仍然比较重。

不同进程：

拥有独立虚拟地址空间

而线程：

属于同一个进程
共享地址空间
共享文件描述符表
共享大量进程资源

所以对于这种：

一个连接
↓
交给一个独立执行流

的问题，也可以使用线程。

整体结构：

主线程
↓
accept
↓
创建新线程
↓
继续 accept

工作线程
↓
Service

4.1 ThreadData：怎么把参数传给线程？

pthread_create() 的线程入口形式：

void* Routine(void*);

而：

Service();

需要：

service_sock
client IP
client Port
TcpServer 对象本身

一个：

void*

显然直接装不下这么多独立参数。

所以把它们封装成一个对象：

class ThreadData
{
public:
ThreadData(int sock, const std::string& ip, uint16_t port, TcpServer* current)
: sock_(sock), clientip_(ip), clientport_(port), current_(current)
{}

public:
int sock_;
std::string clientip_;
uint16_t clientport_;
TcpServer* current_;
};

然后：

ThreadData* td = new ThreadData(service_sock, clientip, clientport, this);

把：

td

作为：

void*

传给线程。

4.2 为什么线程函数是 static？

普通成员函数实际上隐藏了：

this

参数。

而：

pthread_create();

要求的是普通函数形式：

void* (*)(void*);

所以类内线程入口一般写成：

static

static void* Routine(void* args)
{
pthread_detach(pthread_self());

ThreadData* td = static_cast<ThreadData*>(args);

td->current_->Service(td->sock_, td->clientip_, td->clientport_);

delete td;

return nullptr;
}

其中：

pthread_detach(pthread_self());

表示：

当前线程退出以后，系统自动回收线程资源，不要求主线程再 pthread_join()。

4.3 多线程 StartServer

void StartServer()
{
while(true)
{
struct sockaddr_in client;
socklen_t len = sizeof(client);

int service_sock = accept(listen_sock_, (struct sockaddr*)&client, &len);

if(service_sock < 0)
continue;

uint16_t clientport = ntohs(client.sin_port);

char clientip[32];
inet_ntop(AF_INET, &client.sin_addr, clientip, sizeof(clientip));

ThreadData* td = new ThreadData(service_sock, clientip, clientport, this);

pthread_t tid;
pthread_create(&tid, nullptr, Routine, td);
}
}

与多进程版本有一个非常重要的区别。

多进程：

父子进程有各自的文件描述符表

所以父进程可以：

close(service_sock);

只关闭自己那一份引用。

但是线程之间：

共享同一个进程的文件描述符表

所以主线程不能像多进程一样随便再：

close(service_sock);

否则工作线程使用的也是这个文件描述符。

4.4 多线程版 TCP Server 完整代码

到这里，多线程版本需要的内容已经全部具备了：

主线程
↓
accept()
↓
获得 service_sock
↓
构造 ThreadData
↓
pthread_create()
↓
继续 accept()

工作线程
↓
Routine()
↓
pthread_detach()
↓
Service()
↓
专门服务当前客户端

和多进程版本相比，它们解决的问题完全相同：

不能让一个客户端的 Service() 占住服务器唯一的执行流。

但是实现方式发生了变化：

多进程
↓
一个客户端对应一个子进程

多线程
↓
一个客户端对应一个工作线程

现在把前面的内容组合起来，得到完整的多线程 TCP Server。

当前项目结构仍然是：

TcpServer
├── server.hpp
├── server.cc
├── client.hpp
├── client.cc
└── Makefile

客户端代码不需要修改，继续使用前面已经完成的 TCP Client。

这一阶段真正修改的仍然只有 Server。

server.hpp

#pragma once

#include <iostream>
#include <string>
#include <functional>
#include <cstring>
#include <cerrno>
#include <cstdlib>
#include <pthread.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>

namespace fxs_server
{
const uint16_t default_port = 8877;
const std::string default_ip = "0.0.0.0";
const int backlog = 5;

enum
{
USAGE_ERR = 1,
SOCKET_ERR,
BIND_ERR,
LISTEN_ERR
};

class TcpServer;

// 将线程执行Service所需要的数据统一封装起来
class ThreadData
{
public:
ThreadData(int sock, const std::string& ip, uint16_t port, TcpServer* current)
: sock_(sock), clientip_(ip), clientport_(port), current_(current)
{}

public:
int sock_;
std::string clientip_;
uint16_t clientport_;
TcpServer* current_;
};

using func_t = std::function<std::string(std::string)>;

class TcpServer
{
public:
TcpServer(const func_t& func, uint16_t port = default_port, const std::string& ip = default_ip)
: ip_(ip), port_(port), func_(func)
{}

~TcpServer()
{}

void InitServer()
{
// 1.创建监听套接字
listen_sock_ = socket(AF_INET, SOCK_STREAM, 0);

if(listen_sock_ < 0)
{
std::cerr << "Create Socket Fail: "
<< strerror(errno) << std::endl;
exit(SOCKET_ERR);
}

std::cout << "Create Socket Success: "
<< listen_sock_ << std::endl;

// 2.绑定IP和Port
struct sockaddr_in local;
memset(&local, 0, sizeof(local));

local.sin_family = AF_INET;
local.sin_port = htons(port_);
local.sin_addr.s_addr = inet_addr(ip_.c_str());

if(bind(listen_sock_, (const struct sockaddr*)&local, sizeof(local)) < 0)
{
std::cerr << "Bind IP&&Port Fail: "
<< strerror(errno) << std::endl;
exit(BIND_ERR);
}

std::cout << "Bind IP&&Port Success" << std::endl;

// 3.进入监听状态
if(listen(listen_sock_, backlog) < 0)
{
std::cerr << "Listen Fail: "
<< strerror(errno) << std::endl;
exit(LISTEN_ERR);
}

std::cout << "Listen Success" << std::endl;
}

void StartServer()
{
while(true)
{
// 1.主线程获取新连接
struct sockaddr_in client;
socklen_t len = sizeof(client);

int service_sock = accept(listen_sock_, (struct sockaddr*)&client, &len);

if(service_sock < 0)
{
std::cerr << "Accept Fail: "
<< strerror(errno) << std::endl;
continue;
}

// 2.获取客户端IP和Port
char clientip[32];
inet_ntop(AF_INET, &client.sin_addr, clientip, sizeof(clientip));

uint16_t clientport = ntohs(client.sin_port);

std::cout << "Get a new connection -> "
<< "[" << clientip << ":" << clientport << "] "
<< "service_sock: " << service_sock
<< std::endl;

// 3.构造线程需要的参数
ThreadData* td = new ThreadData(service_sock, clientip, clientport, this);

// 4.创建线程，让工作线程负责当前客户端
pthread_t tid;
pthread_create(&tid, nullptr, Routine, td);

// 主线程不进入Service，直接继续下一轮accept
}
}

// 工作线程入口函数
static void* Routine(void* args)
{
// 分离当前线程，退出后自动回收资源
pthread_detach(pthread_self());

ThreadData* td = static_cast<ThreadData*>(args);

// 调用当前TcpServer对象的Service
td->current_->Service(td->sock_, td->clientip_, td->clientport_);

delete td;

return nullptr;
}

// 通信服务 + 业务处理
void Service(int sock, const std::string& clientip, uint16_t clientport)
{
char buff[1024];
std::string who = clientip + ":" + std::to_string(clientport);

while(true)
{
ssize_t n = read(sock, buff, sizeof(buff) – 1);

if(n > 0)
{
buff[n] = '\\0';

std::cout << "Server get: "
<< buff
<< " from "
<< who
<< std::endl;

std::string respond = func_(buff);

write(sock, respond.c_str(), respond.size());
}
else if(n == 0)
{
std::cout << "Client "
<< who
<< " quit!"
<< std::endl;

close(sock);
break;
}
else
{
std::cerr << "Read Fail: "
<< strerror(errno)
<< std::endl;

close(sock);
break;
}
}
}

private:
int listen_sock_;
std::string ip_;
uint16_t port_;
func_t func_;
};
}

这一版最核心的地方就是：

ThreadData* td = new ThreadData(service_sock, clientip, clientport, this);

pthread_t tid;
pthread_create(&tid, nullptr, Routine, td);

主线程得到连接以后，并不会自己执行：

Service();

而是创建一个线程：

service_sock
client IP
client Port
this

全部交给工作线程。

工作线程再执行：

td->current_->Service(td->sock_, td->clientip_, td->clientport_);

所以：

Main Thread
│
├── accept Client A
│ ↓
│ Thread A -> Service(A)
│
├── accept Client B
│ ↓
│ Thread B -> Service(B)
│
└── accept Client C
↓
Thread C -> Service(C)

主线程始终能够继续：

accept();

server.cc

业务层继续使用 Echo，不需要任何修改：

#include <iostream>
#include <memory>
#include "server.hpp"

using namespace std;
using namespace fxs_server;

std::string Echo(const std::string& request)
{
return request;
}

void Usage(const char* program)
{
cout << "Usage:" << endl;
cout << "\\t" << program << " ServerPort" << endl;
}

int main(int argc, char* argv[])
{
if(argc != 2)
{
Usage(argv[0]);
return USAGE_ERR;
}

uint16_t port = stoi(argv[1]);

unique_ptr<TcpServer> server(new TcpServer(Echo, port));

server->InitServer();
server->StartServer();

return 0;
}

这里再次能够看到我们前面进行业务解耦的好处。

无论服务器使用：

单进程
多进程
多线程

业务函数始终还是：

std::string Echo(const std::string& request)
{
return request;
}

变化的是：

服务器如何安排执行流去处理不同客户端。

业务本身没有发生变化。

Makefile

由于现在使用：

pthread_create();
pthread_detach();

需要链接 pthread：

.PHONY:all
all:server client

server:server.cc
g++ -o $@ $^ -std=c++11 -lpthread

client:client.cc
g++ -o $@ $^ -std=c++11

.PHONY:clean
clean:
rm -rf server client

客户端仍然使用前面已经完成的：

client.hpp
client.cc

不需要修改。

4.5 运行多线程 TCP Server

重新编译：

make clean
make

启动 Server：

./server 8877

然后分别启动多个客户端。

Client A：

./client 127.0.0.1 8877

Client B：

./client 127.0.0.1 8877

Client C：

./client 127.0.0.1 8877

三个 Client 都可以持续发送消息：

Client A
Please Enter >> hello A

Client B
Please Enter >> hello B

Client C
Please Enter >> hello C

并分别收到服务器 Echo。

此时服务器内部结构已经变成：

Server Process
│
├── Main Thread
│ ↓
│ accept()
│
├── Thread A
│ ↓
│ Service(Client A)
│
├── Thread B
│ ↓
│ Service(Client B)
│
└── Thread C
↓
Service(Client C)

这和多进程版本：

Parent
├── Child A
├── Child B
└── Child C

解决的是同一个问题。

只不过现在所有执行流都属于：

同一个 Server 进程。

使用 ps 查看线程

可以打开另一个终端：

ps -aL | grep server

其中：

-L

表示：

显示进程中的线程。

假设目前有三个客户端，那么可以看到同一个：

PID

下面存在多个不同的：

LWP

大致可以理解为：

PID
↓
属于哪个进程

LWP
↓
具体哪个线程

所以你会看到：

一个 Server PID
+
多个不同的线程 ID

这与前面的多进程版本形成非常明显的对比。

多进程：

ps -ef
↓
多个不同 PID

多线程：

ps -aL
↓
同一个 PID
+
多个线程

4.6 多进程和多线程到这里到底差在哪？

现在正好可以把两个版本放在一起。

多进程：

Server Parent
│
├── Child A -> Client A
├── Child B -> Client B
└── Child C -> Client C

多线程：

Server Process
│
├── Main Thread
├── Thread A -> Client A
├── Thread B -> Client B
└── Thread C -> Client C

它们的共同点：

主执行流
↓
只负责accept

其他执行流
↓
负责Service

不同点：

多进程
↓
不同进程拥有独立虚拟地址空间
↓
资源隔离更强
↓
创建和切换相对更重

多线程
↓
多个线程属于同一个进程
↓
共享地址空间、文件描述符等资源
↓
创建和切换相对更轻

但是多线程版本现在还有一个问题。

当前是：

Client 来了
↓
pthread_create()

又来一个 Client
↓
再 pthread_create()

再来一个
↓
继续 pthread_create()

也就是说：

线程仍然是在客户端到来以后临时创建的。

如果客户端特别多：

连接越多
↓
创建线程越多
↓
线程数量缺乏统一控制

那么有没有办法：

程序启动的时候就提前创建好固定数量的线程，客户端来了以后只需要把任务交给它们？

这就是我们前面已经学习过的：

线程池。

于是服务器继续演进：

单进程
↓
多进程
↓
多线程
↓
线程池

5. 线程池版 TCP Server

现在多线程服务器还有一个问题。

当前逻辑：

客户端来了
↓
accept
↓
pthread_create
↓
创建线程

也就是说：

客户端已经来了，我们才临时创建线程。

如果连接数量不断增加：

来一个
↓
建一个线程

再来一个
↓
再建一个线程

……

线程数量就可能不断增长。

这并不是一个很好的资源管理方式。

于是我们自然想到前面已经学习过的：

线程池。

5.1 为什么使用线程池？

线程池：

程序启动
↓
提前创建固定数量线程
↓
所有线程等待任务

客户端来了以后：

accept()
↓
构造 Task
↓
Push 进入任务队列
↓
唤醒工作线程
↓
处理 Task

所以：

连接到来

和：

线程创建

不再绑定在一起。

整个结构：

accept
↓
Task Queue
↙ ↓ ↘
thread1 thread2 thread3

这本质上就是以前的：

生产者消费者模型。

主线程：

生产任务

工作线程：

消费任务

5.2 Task 到底是什么？

在当前 TCP Server 中，一份任务至少需要：

service_sock
client IP
client Port
业务处理函数

所以：

class Task
{
public:
Task(int sockfd, const std::string& clientip, uint16_t clientport, const func_t& func)
: sockfd_(sockfd), clientip_(clientip), clientport_(clientport), func_(func)
{}

void Run()
{
char buff[1024];

ssize_t n = read(sockfd_, buff, sizeof(buff) – 1);

if(n > 0)
{
buff[n] = '\\0';

std::string respond = func_(buff);

write(sockfd_, respond.c_str(), respond.size());
}

close(sockfd_);
}

private:
int sockfd_;
std::string clientip_;
uint16_t clientport_;
func_t func_;
};

然后：

Task task(service_sock, clientip, clientport, func_);
ThreadPool<Task>::GetInstance()->Push(task);

5.3 为什么这里去掉 Service 的 while(true)？

这个地方必须讲明白。

如果线程池只有：

5 个线程

而一个任务内部：

while(true)
{
read();
write();
}

表示：

一个 Worker 一旦服务一个长连接，就一直被这个连接占住。

那么：

5 个线程

最多就只能同时被：

5 个长期连接

占用。

所以这个教学版本采用：

一个连接
↓
读取一次
↓
处理一次
↓
返回一次
↓
关闭连接

的短服务模型。

因此如果 Client 仍然想在同一个 TCP 连接中发送第二条消息：

第一次
↓
Server 处理
↓
Server close

第二次
↓
原连接已经关闭

当然会失败。

这不是：

线程池导致 TCP 不能连续通信。

更不是：

TCP 一个连接只能发送一次数据。

而只是：

我们这里的 Task::Run() 主动设计成了一次请求处理完就关闭连接。

如果业务需要：

TCP 长连接

完全可以重新设计任务模型。

这里只是利用它理解：

线程池
+
TCP Server

是如何结合的。

5.4 主线程加入线程池

核心代码：

void StartServer()
{
ThreadPool<Task>::GetInstance()->Start();

while(true)
{
struct sockaddr_in client;
socklen_t len = sizeof(client);

int service_sock = accept(listen_sock_, (struct sockaddr*)&client, &len);

if(service_sock < 0)
continue;

uint16_t clientport = ntohs(client.sin_port);

char clientip[32];
inet_ntop(AF_INET, &client.sin_addr, clientip, sizeof(clientip));

Task task(service_sock, clientip, clientport, func_);

ThreadPool<Task>::GetInstance()->Push(task);
}
}

于是：

主线程
↓
只负责 accept + Push

线程池
↓
负责真正执行 Task::Run()

5.5 线程池版 TCP Server 完整代码

到这里，线程池版本需要的几个模块已经全部具备：

主线程
↓
accept()
↓
获得一个新的 service_sock
↓
构造 Task
↓
Push 到任务队列
↓
继续 accept()

线程池中的工作线程
↓
提前创建
↓
等待任务
↓
从任务队列取出 Task
↓
Task::Run()
↓
处理客户端请求

和前面的多线程版本相比，最大的变化就是：

多线程版

Client 到来
↓
pthread_create()
↓
临时创建工作线程

变成：

线程池版

Server 启动
↓
提前创建固定数量工作线程
↓
Client 到来
↓
只需要构造 Task
↓
放入任务队列

这一阶段项目结构变成：

TcpServer
├── ThreadPool.hpp
├── Task.hpp
├── server.hpp
├── server.cc
├── client.hpp
├── client.cc
└── Makefile

其中：

client.hpp
client.cc

继续使用前面已经完成的 TCP Client，不需要修改。

这一阶段新增的是：

ThreadPool.hpp
Task.hpp

并重新修改：

server.hpp

ThreadPool.hpp

这里直接使用我们之前学习过的线程池。

线程池本质上仍然是一个：

生产者消费者模型。

主线程
↓
生产 Task
↓
任务队列

工作线程
↓
消费 Task
↓
执行 Run()

代码如下：

#pragma once

#include <iostream>
#include <vector>
#include <string>
#include <queue>
#include <pthread.h>

struct ThreadInfo
{
pthread_t tid;
std::string name;
};

static const int default_num = 5;

template<class T>
class ThreadPool
{
private:
void Lock()
{
pthread_mutex_lock(&mutex_);
}

void Unlock()
{
pthread_mutex_unlock(&mutex_);
}

void Wakeup()
{
pthread_cond_signal(&cond_);
}

void ThreadSleep()
{
pthread_cond_wait(&cond_, &mutex_);
}

bool IsQueueEmpty()
{
return tasks_.empty();
}

public:
static void* HandlerTask(void* args)
{
ThreadPool<T>* tp = static_cast<ThreadPool<T>*>(args);

while(true)
{
tp->Lock();

while(tp->IsQueueEmpty())
tp->ThreadSleep();

T task = tp->Pop();

tp->Unlock();

task.Run();
}

return nullptr;
}

void Start()
{
int num = threads_.size();

for(int i = 0; i < num; i++)
{
threads_[i].name = "thread-" + std::to_string(i + 1);
pthread_create(&threads_[i].tid, nullptr, HandlerTask, this);
}
}

T Pop()
{
T task = tasks_.front();
tasks_.pop();

return task;
}

void Push(const T& task)
{
Lock();

tasks_.push(task);

Wakeup();

Unlock();
}

static ThreadPool<T>* GetInstance()
{
if(tp_ == nullptr)
{
pthread_mutex_lock(&lock_);

if(tp_ == nullptr)
tp_ = new ThreadPool<T>();

pthread_mutex_unlock(&lock_);
}

return tp_;
}

private:
ThreadPool(int num = default_num)
: threads_(num)
{
pthread_mutex_init(&mutex_, nullptr);
pthread_cond_init(&cond_, nullptr);
}

~ThreadPool()
{
pthread_mutex_destroy(&mutex_);
pthread_cond_destroy(&cond_);
}

ThreadPool(const ThreadPool<T>&) = delete;
ThreadPool<T>& operator=(const ThreadPool<T>&) = delete;

private:
std::vector<ThreadInfo> threads_;
std::queue<T> tasks_;

pthread_mutex_t mutex_;
pthread_cond_t cond_;

static ThreadPool<T>* tp_;
static pthread_mutex_t lock_;
};

template<class T>
ThreadPool<T>* ThreadPool<T>::tp_ = nullptr;

template<class T>
pthread_mutex_t ThreadPool<T>::lock_ = PTHREAD_MUTEX_INITIALIZER;

线程池启动以后：

ThreadPool<Task>::GetInstance()->Start();

就会提前创建：

5 个 Worker

这些线程开始时没有任务，于是：

Lock
↓
发现任务队列为空
↓
pthread_cond_wait
↓
线程休眠

主线程以后：

Push(task);

就会：

任务入队
↓
pthread_cond_signal
↓
唤醒一个 Worker
↓
取出 Task
↓
Task::Run()

Task.hpp

那么什么是：

Task

呢？

前面的多线程版本把一个连接直接交给：

Service()

处理。

现在使用线程池以后，不能再：

accept
↓
创建线程
↓
Service

而是需要先把：

这次需要做的事情

封装成一个对象。

一份 TCP 服务任务至少需要知道：

service_sock
客户端 IP
客户端 Port
业务处理函数

因此：

Task task(service_sock, clientip, clientport, func_);

本质上就是：

把一次网络服务所需要的所有信息打包。

Task.hpp：

#pragma once

#include <iostream>
#include <string>
#include <functional>
#include <cstring>
#include <cerrno>
#include <unistd.h>

namespace fxs_server
{
using func_t = std::function<std::string(std::string)>;

class Task
{
public:
Task(int sockfd, const std::string& clientip, uint16_t clientport, const func_t& func)
: sockfd_(sockfd), clientip_(clientip), clientport_(clientport), func_(func)
{}

~Task()
{}

void Run()
{
char buff[1024];
std::string who = clientip_ + ":" + std::to_string(clientport_);

ssize_t n = read(sockfd_, buff, sizeof(buff) – 1);

if(n > 0)
{
buff[n] = '\\0';

std::cout << "Server get: "
<< buff
<< " from "
<< who
<< std::endl;

std::string respond = func_(buff);

write(sockfd_, respond.c_str(), respond.size());
}
else if(n == 0)
{
std::cout << "Client "
<< who
<< " quit!"
<< std::endl;
}
else
{
std::cerr << "Read Fail: "
<< strerror(errno)
<< std::endl;
}