目录
前言
UDP服务器的完善
线程的封装
结构定义
接口实现
环形队列
结构定义
接口实现
加锁
信号量的申请与释放
入队与出队
整体组装
初始化与析构
信息接收线程
消息发送线程
TCP套接字
创建套接字
listen
accept
收发操作
客户端的编写
进一步完善
多进程
多线程
总结
前言
上篇文章中我们介绍了套接字编程，简单构建了一个 UDP 服务器，今天在此基础上添加并行的模块，之后再进行 TCP 套接字的介绍，并同样完成一个服务端的构建。
UDP服务器的完善
🎃之前的版本中，我们将数据的读取和发送同样写在一个串行的逻辑之中，因此若此时我们未向命令行输入数据则无法收到服务器发送的信息。
🎃因此，我们可以使用线程分别进行消息的读取和发送，同时我们想要将这个程序构建成一个小型的聊天室，因此服务器接收到一个数据时要向已连接上的所有用户进行广播。所以我们还需要使用一个哈希表将用户的相关数据管理起来。
🎃同时，我们还可以维护一个环形队列，收线程就作为生产者而发线程为消费者，二者分别进行队列的插入与读取，当一方不满足条件时阻塞即可。
线程的封装
结构定义
🎃对于一个线程，我们需要封装他的 tid 、回调函数、当前状态，还可以给线程取个名字。
🎃在初始化列表完成成员的初始化，而构造函数中根据传入线程的序列号生成一个线程名。
class Thread
{
public:
typedef enum
{
NEW = 0,
RUNNING,
EXIT
} ThreadStatus;
using func_t = std::function<void()>;
Thread(int num, func_t func) : _tid(0), _status(NEW), _func(func)
{
char name[128];
snprintf(name, sizeof(name), "thread-%d", num);
_name = name;
}
~Thread()
{
}
private:
pthread_t _tid;
std::string _name;
func_t _func;
ThreadStatus _status;
};
接口实现
🎃之后我们只需要简单封装对应的接口即可，首先便是成员的获取。而获取 tid 时需要判断一下当前线程是否在运行，否则直接返回 0。
int status() { return _status; }
std::string name() { return _name; }
pthread_t threadid()
{
if (_status == RUNNING)
return _tid;
else
return 0;
}🎃接下来就是线程运行的老三样了，即运行、等待、回调。
🎃很明显，在构造函数时我们并未创建线程，因为线程创建的这个过程在运行的函数中进行实现。
void run()
{
int n = pthread_create(&_tid, nullptr, RunHelper, this);
if (n != 0)
exit(1);
_status = RUNNING;
}🎃接下来实现线程运行的逻辑，首先将传进来的参数转换回 Thread* 再回调函数。
🎃而我们还进行了运算符重载，可以直接回调，但在调用前应先判断对应的函数是否为空。
static void* RunHelper(void* args) // 由于本函数是静态成员函数，线程调用时还需要外部传this指针进来
{
Thread* ts = (Thread*)args; // 强转恢复this指针
(*ts)();
return nullptr;
}
void operator()()
{
if (_func != nullptr)
_func();
}🎃最后就是线程的运行和等待了，run 函数中我们直接创建一个线程，入口函数就为上面写的 RunHelper ，之后将线程的状态设置为运行状态。join 就是对接口进行简单的封装，若等待出错就进行报错。
void run()
{
int n = pthread_create(&_tid, nullptr, RunHelper, this);
if (n != 0)
exit(1);
_status = RUNNING;
}
void join()
{
int n = pthread_join(_tid, nullptr);
if (n != 0)
{
std::cout << n << std::endl;
std::cerr << "main thread join thread " << _name << " error" << std::endl;
}
}环形队列
结构定义
🎃环形队列的底层其实就是一个数组，通过维护下标获取插入和读取的位置，同时其为一个临界资源，涉及到生产者间与消费者间的访问冲突，因此还需要两个锁，接着我们还可以使用信号量维护环形队列中对应的数据量。
🎃在构造函数中我们完成各个成员的初始化，对于数据和空间的信号量，显然一开始环形队列为空，因此数据量为 0，而空间的量为环形队列的大小。同时，在析构时就需要完成信号量和锁的销毁。
const int N = 50;
template <class T>
class RingQueue
{
public:
RingQueue(int num = N) : _ring(num), _cap(num)
{
sem_init(&_data_sem, 0, 0);
sem_init(&_space_sem, 0, num);
pthread_mutex_init(&_c_mutex, nullptr);
pthread_mutex_init(&_p_mutex, nullptr);
_c_step = _p_step = 0;
}
~RingQueue()
{
sem_destroy(&_data_sem);
sem_destroy(&_space_sem);
pthread_mutex_destroy(&_c_mutex);
pthread_mutex_destroy(&_p_mutex);
}
private:
std::vector<T> _ring;
int _cap;
sem_t _data_sem; // 数据的信号量
sem_t _space_sem; // 空间的信号量
int _c_step; // 消费位置
int _p_step; // 生产位置
pthread_mutex_t _c_mutex; // 消费者间的锁
pthread_mutex_t _p_mutex; // 生产者间的锁
};接口实现
加锁
🎃就是简单对对应接口的封装，两个锁都可以直接调用进行加锁的操作。
void Lock(pthread_mutex_t &m) // 加锁
{
pthread_mutex_lock(&m);
}
void UnLock(pthread_mutex_t &m) // 解锁
{
pthread_mutex_unlock(&m);
}信号量的申请与释放
🎃这里同样是对系统接口的再次封装，经由这些步骤便可使得我们的代码更加规范，能够用统一的视角来看待变量。
void P(sem_t& s) // 申请信号量
{
sem_wait(&s);
}
void V(sem_t& s) // 释放信号量
{
sem_post(&s);
}入队与出队
🎃当我们要插入数据时，先申请空间信号量，若无空余便阻塞，之后直接加锁接下来我们便能够进行数据的插入，插入位置迭代后需要 % 上环形队列的大小才能进行环形的读取与插入。数据插入完毕后便可以解锁，最后释放一个数据的信号量，表示我们成功插入一个数据。
void push(const T in)
{
P(_space_sem); // 申请空间信号量
Lock(_p_mutex); // 生产者加锁
_ring[_p_step++] = in; // 数据写入
_p_step %= _cap;
UnLock(_p_mutex);
V(_data_sem); // 释放数据的信号量
}🎃而读取的操作类似，只不过申请的是数据的信号量，释放的是空间的信号量，同时这里使用的是输出型参数，读取的数据填充到指针里即可。
void pop(T* out)
{
P(_data_sem); // 查看数据数量
Lock(_c_mutex); // 消费者加锁
*out = _ring[_c_step++]; // 读取数据
_c_step %= _cap;
UnLock(_c_mutex);
V(_space_sem); // 释放空间的信号量
}整体组装
🎃增加了那么多新的组件，接下来我们进行服务器功能的完善。
🎃首先便是需要增加成员，因为要维护在线用户因此需要一个哈希表进行管理，同时，我们这个的表也是被多线程访问的，因此还需要一个锁进行保护。接着还需要使用两个线程分别进行数据的读取和发送，以及一个环形队列存储要发送的信息。
private:
int _sock;
uint16_t _port;
pthread_mutex_t lock;
std::unordered_map<std::string, struct sockaddr_in> OnlineUser;
RingQueue<std::string> rq;
Thread* c;
Thread* p;初始化与析构
🎃因为新增了许多成员，我们需要在构造函数中进行部分成员的初始化。这里线程的入口函数分别为接下来要实现的收发操作，同时因为二者为成员函数，因此我们需要手动为其绑定 this 指针作为第一个参数。
UdpServer(int port = defaultport)
: _port(port)
{
pthread_mutex_init(&lock, nullptr);
p = new Thread(1, std::bind(&UdpServer::Recv, this));
c = new Thread(2, std::bind(&UdpServer::Broadcast, this));
}🎃在网络前期准备后，便可以直接进行线程的运行。
if (bind(_sock, (sockaddr*)&local, sizeof(local)))
{
std::cout << "bind socket error: " << strerror(errno) << std::endl;
exit(BIND_ERR);
}
std::cout << "bind socket success" << std::endl;
c->run();
p->run();🎃而析构时就进行线程的等待，待到线程终止就删除 new 出来的对象。
~UdpServer()
{
pthread_mutex_destroy(&lock);
c->join();
p->join();
delete c;
delete p;
}信息接收线程
🎃前面的操作大部分的都讲过了，接收到数据后提取发送方的相关信息，将其加入到表中，而要发送回用户的字符串则放进环形队列中。
void Recv()
{
char buffer[1024];
while (true)
{
// 接收信息
struct sockaddr_in peer;
socklen_t len = sizeof(peer);
int n = recvfrom(_sock, buffer, sizeof(buffer) - 1, 0, (sockaddr*)&peer, &len);
if (n > 0)
buffer[n] = '\0';
else
continue;
// 提取客户端信息
std::string clientip = inet_ntoa(peer.sin_addr);
uint16_t port = ntohs(peer.sin_port);
std::cout << "get message from " << clientip << "-" << port << ": " << buffer << std::endl;
std::string name = clientip;
name += "-";
name += std::to_string(port);
addUser(name, peer);
std::string message = name + "echo# " + buffer;
rq.push(message);
}
}🎃因为用户的注册表是一个临界资源，因此在访问前一定要先加锁，这里使用的是封装的一个类，类似于智能指针的作用，直接理解为加锁函数结束时解锁即可，若此时注册表中未有对应名称的用户，就将用户的 ip 信息插入进去。
void addUser(const std::string& name, const struct sockaddr_in& peer)
{
LockGuard lg(&lock);
if (!OnlineUser.count(name))
OnlineUser[name] = peer;
}消息发送线程
🎃而这个广播线程就时时刻刻从环形队列中拿取数据，通过遍历注册表将消息发给所有用户。
🎃这里有个小细节，我们使用一个 vector 先将用户的 ip 信息记录下来，临界区域结束后再进行发送，因为若是直接在锁内进行数据的发送就会占用过久临界资源，进而影响程序的运行效率。
void Broadcast()
{
while (true)
{
std::string sendstring;
rq.pop(&sendstring);
std::vector<sockaddr_in> v;
{
LockGuard lockguard(&lock);
for (auto& usr : OnlineUser)
{
v.push_back(usr.second);
}
}
for (auto& usr : v)
{
sendto(_sock, sendstring.c_str(), sendstring.size(), 0, (sockaddr*)&usr, sizeof(usr));
}
}
}🎃于客户端而言，将接收操作交由一个线程处理，便能够做到即便未发数据，也能够同步收到服务器发送的消息。
TCP套接字
🎃接下来我们就进行 TCP 套接字编程的讲解，之前就讲过 TCP 是有连接、可靠的传输方式，自然也代表着其通信过程相较于 UDP 更为复杂。
创建套接字
🎃UDP 在创建套接字时做的工作，TCP 也都要做，即创建套接字 fd 、填充 sockaddr_in、bind，但需要注意的是创建 socket 时使用的是 SOCK_STREAM 选项。
socket(AF_INET, SOCK_STREAM, 0);listen
🎃接下来 TCP 服务器还需要调用 listen 才能完成前置准备工作。
🎃第一个参数即为前面创建的 socket 文件描述符，第二个参数表示为挂起连接队列的最大长度，定义一个不大不小的值即可，这里就设置成了 32。
static const int backlog = 32;
void initserver()
{
_listensock = socket(AF_INET, SOCK_STREAM, 0);
if (_listensock < 0)
{
std::cerr << "create socket error" << std::endl;
exit(SOCKET_ERR);
}
struct sockaddr_in local;
local.sin_family = AF_INET;
local.sin_addr.s_addr = INADDR_ANY;
local.sin_port = htons(_port);
if (bind(_listensock, (sockaddr *)&local, sizeof(local)) != 0)
{
std::cerr << "bind socket error" << std::endl;
exit(BIND_ERR);
}
if (listen(_listensock, backlog) != 0)
{
std::cerr << "listen socket error" << std::endl;
exit(LISTEN_ERR);
}
}accept
🎃若说 UDP 的传输方式像发快递，那么 TCP 的传输方式就像餐馆的运行方式。
🎃一个餐馆的运行的方式可以分成两部分，分别是餐馆内部和餐馆外部，外部有人负责揽客，而内部则有服务员负责处理用户就餐的请求。
🎃同样，前面我们创建的套接字负责的就是揽客工作，当有外部有请求尝试连接便会通过 accept 函数返回。
🎃而我们之后用于通信的 fd 其实是 accept 返回的，接下来就可以凭借这个 fd 进行消息的收发了。
void start()
{
while (true)
{
struct sockaddr_in client;
socklen_t len = sizeof(client);
int sock = accept(_listensock, (sockaddr *)&client, &len);
if (sock < 0)
{
std::cerr << "accept socket error" << std::endl;
continue;
}
std::string clientip = inet_ntoa(client.sin_addr);
uint16_t port = ntohs(client.sin_port);
std::string name = clientip;
name += "-";
name += std::to_string(port);
std::cout << "获取新连接成功 " << sock << " from " << _listensock << "," << name << std::endl;
service(sock, clientip, port);
}
}收发操作
🎃因为 TCP 通信面向字节流，而流式服务都可以直接使用 read 进行读取，这里我们便直接使用 read 即可。
🎃同时，read 的返回值有三种不同的情况，大于 0 时表示为读取数据的字节数，等于 0 表示断开连接，小于 0 就表示出错。
🎃读取的时候记得给 /0 留一个位置，接下来根据业务进行处理后便可以发回给用户了，同样我们可以直接使用 write 进行数据发送。
void service(int sock, const std::string &ip, const uint16_t &port)
{
char buffer[1024];
std::string name = ip + '-' + std::to_string(port);
while (true)
{
ssize_t s = read(sock, buffer, sizeof(buffer) - 1);
if (s > 0)
{
buffer[s] = '\0';
std::string res = _func(buffer); //调用回调函数进行业务处理
std::cout << name << "# " << res << std::endl;
write(sock, buffer, sizeof(buffer) - 1);
}
else if (s == 0)
{
close(sock);
std::cout << ip << " quit" << std::endl;
break;
}
else
{
close(sock);
std::cout << "recv error" << std::endl;
break;
}
}
}🎃经过这几个步骤，TCP 服务器的通信框架便搭建起来了，接下来就进行客户端的编写。
客户端的编写
🎃首先我们从命令行参数中获取服务器的 ip 与端口号，接着创建套接字并填充 sockaddr_in 结构，接下来便可以与服务器进行连接了。
int main(int args, char *argv[])
{
if (args != 3)
{
usage(argv[0]);
exit(USAGE_ERR);
}
std::string server_ip = argv[1];
uint16_t server_port = atoi(argv[2]);
int sock = socket(AF_INET, SOCK_STREAM, 0);
if (sock < 0)
{
cerr << "create socket error" << endl;
exit(SOCKET_ERR);
}
struct sockaddr_in server;
server.sin_family = AF_INET;
server.sin_addr.s_addr = inet_addr(server_ip.c_str());
server.sin_port = htons(server_port);
}🎃接下来就需要使用 connect 函数进行与服务器的连接了。
🎃参数于我们而言相当熟悉了，我们还可以写一个简单的重连逻辑，若连接不上就直接使进程退出了。
int cnt = 1;
while (connect(sock, (sockaddr *)&server, sizeof(server)) != 0)
{
cout << "正在重连(" << cnt++ << ")" << endl;
sleep(1);
if (cnt > 5)
break;
}
if (cnt > 5)
{
cerr << "连接失败" << endl;
exit(CONNECT_ERR);
}🎃建立连接后便可以进行数据的收发了，还是与服务端相同的操作，而当 read 返回值为 0 时就代表断开连接。
char buffer[1024];
while (true)
{
std::string line;
cout << "please Enter# ";
getline(cin, line);
write(sock, line.c_str(), line.size());
ssize_t s = read(sock, buffer, sizeof(buffer) - 1);
if (s > 0)
{
buffer[s] = '\0';
cout << "server echo# " << buffer << endl;
}
else if (s == 0)
{
cerr << "服务器崩溃" << endl;
break;
}
else
{
cerr << "read error:" << strerror(errno) << endl;
break;
}
}进一步完善
🎃经过上面的构建，虽然我们能够构建出通信的环境，但我们会发现只有第一个用户能够连接上服务器。
🎃这是因为我们在 accept 后直接就串行地为用户提供服务，又因为服务逻辑是一个死循环，因此就无法再次接收新用户连接了。
🎃所以接下来我们就使用多线程或多进程的方法使其能够支持多用户的连接。
多进程
🎃需要注意的一点是，无论是线程还是进程，若是再进行等待，同样会使主进程/线程阻塞，因此需要忽略掉子进程/线程返回的相关信息。
🎃对于多进程而言可以直接使用 signal 进行忽略。
signal(SIGCHLD, SIG_IGN);🎃但这里我使用了另一种方法， 在子进程创建出来后，再创建一个子进程并使其父进程退出，使其成为一个孤儿进程。这样就将其与主进程分离了。
🎃同时，在创建进程后我们需要将不需要的文件描述符关闭。
// 多进程
int id = fork();
if (id < 0)
{
close(sock);
continue;
}
else if (id == 0) // 子进程
{
close(_listensock);
if (fork() > 0)
exit(0);
service(sock, clientip, port);
exit(0);
}多线程
🎃为了方便使用，我们维护了一个结构用于存储线程需要使用到的信息，之后创建线程将这个结构传进去就行。
class ThreadData
{
public:
ThreadData(int fd, const std::string &ip, uint16_t port, TcpServer *ts)
: socket(fd), clientip(ip), clientport(port), current(ts)
{
}
public:
int socket;
std::string clientip;
uint16_t clientport;
TcpServer *current;
};🎃先构建出对应结构，接下来便可以进行线程的创建。
// 多进程
pthread_t tid;
ThreadData *td = new ThreadData(sock, clientip, clientport, this);
pthread_create(&tid, nullptr, threadRoutine, td);🎃线程进入入口函数的第一件事便是将自身独立出来，将参数转换出来后便可以通过 this 指针调用服务。
static void *threadRoutine(void *args)
{
pthread_detach(pthread_self());
ThreadData *td = static_cast<ThreadData *>(args);
td->current->service(td->socket, td->clientip, td->clientport);
delete td;
return nullptr;
}🎃由此我们便完成了 TCP 服务端的简单构建。
总结
🎃对于 UDP 服务器的构建，需要以下三步前置工作：
- 创建套接字
- 构建 sockaddr_in 结构
- bind
🎃而 TCP 服务器的构建则需要再此基础上再添加两步：
- listen
- accept
🎃同时，其有连接的性质，注定了服务端需要使用并行的方式进行用户连接与数据收发。
🎃于客户端而言，TCP 连接时只需要比 UDP 多出一步与服务器 connect 。
