如何在 C/C++ 中进行 TCP 连接池

Question

我正在使用 C++ 设计一个分布式服务器/客户端系统，其中许多客户端通过 TCP 向许多服务器发送请求，服务器抛出一个线程来处理请求并发送回它的响应。在我的用例中，只有有限数量的客户端会访问服务器，我需要非常高的性能。客户端和服务器发送的数据都很小，但非常频繁。因此，创建连接并在使用后将其拆除是昂贵的。所以我想使用连接缓存来解决这个问题：一旦连接创建，它就会被存储在一个缓存中以备将来使用。（假设客户端的数量不会超过缓存的大小）。

我的问题是：

1. 我看到有人说连接池是一种客户端技术。如果这个连接池只用在客户端，那么它第一次连接到服务器，然后发送数据。这个建立连接的动作会触发服务器端的 accept() 函数，该函数返回一个用于从客户端接收的套接字。因此，当客户端想要使用现有连接（在缓存中）时，它不会建立新连接，而只是发送数据。问题是，如果没有建立连接，谁会在服务器端触发 accept() 并抛出一个线程？
2. 如果连接池也需要在服务器端实现，我怎么知道请求来自哪里？因为我只能从 accept() 获取客户端地址，但同时 accept() 已经为该请求创建了一个新套接字，所以没有必要使用缓存连接。

任何答案和建议将不胜感激。或者任何人都可以给我一个连接池或连接缓存的例子？

Answer 1

我看到有人说连接池是一种客户端技术。 ...如果没有建立连接，谁会在服务器端触发accept()并抛出一个线程？

首先，连接池不仅仅是一种客户端技术；这是一种连接模式技术。它适用于两种类型的对等点（“服务器”和“客户端”）。

其次，不需要调用accept 来启动线程。程序可以以任何他们喜欢的原因启动线程......他们可以启动线程只是为了启动更多线程，在线程创建的大规模并行循环中。 （编辑：我们称之为“叉子炸弹”）

最后，一个高效的线程池实现不会为每个客户端启动一个线程。每个线程通常占用 512KB-4MB（计算堆栈空间和其他上下文信息），因此如果您有 10000 个客户端，每个客户端都占用这么多内存，那就浪费了很多内存。

我想这样做，但不知道如何在多线程情况下这样做。

你不应该在这里使用多线程......至少，直到你有一个使用单线程的解决方案，并且你认为它不够快。目前您没有该信息；你只是猜测，猜测并不能保证优化。

世纪之交有解决the C10K problem的FTP服务器；他们能够在任何给定时间处理 10000 个客户端，浏览、下载或空闲，就像用户在 FTP 服务器上所做的那样。 他们不是通过使用线程来解决这个问题，而是通过使用非阻塞和/或异步套接字和/或调用。

澄清一下，那些网络服务器在一个线程上处理了数千个连接！一种典型的方法是使用select，但我并不是特别喜欢这种方法，因为它需要一系列相当丑陋的循环。我更喜欢在 Windows 上使用 ioctlsocket，在其他 POSIX 操作系统上使用 fcntl 来将文件描述符设置为非阻塞模式，例如：

#ifdef WIN32
ioctlsocket(fd, FIONBIO, (u_long[]){1});
#else
fcntl(fd, F_SETFL, fcntl(fd, F_GETFL, 0) | O_NONBLOCK);
#endif

此时recv和read在fd上操作时不会阻塞；如果没有可用的数据，它们会立即返回一个错误值，而不是等待数据到达。这意味着您可以在多个套接字上循环。

如果服务器端也需要实现连接池，我怎么知道请求来自哪里？

将客户端 fd 与其 struct sockaddr_storage 以及您需要存储的有关客户端的任何其他状态信息一起存储在 struct 中，您可以根据自己的感觉声明。如果这最终是 4KB（这是一个相当大的struct，通常与他们需要的一样大），那么其中的 10000 个将只占用大约 40000KB（~40MB）。即使是今天的手机也应该没有问题。考虑根据您的需要完成以下代码：

struct client {
    struct sockaddr_storage addr;
    socklen_t addr_len;
    int fd;
    /* Other stateful information */
};

#define BUFFER_SIZE 4096
#define CLIENT_COUNT 10000

int main(void) {
    int server;
    struct client client[CLIENT_COUNT] = { 0 };
    size_t client_count = 0;
    /* XXX: Perform usual bind/listen */
    #ifdef WIN32
    ioctlsocket(server, FIONBIO, (u_long[]){1});
    #else
    fcntl(server, F_SETFL, fcntl(server, F_GETFL, 0) | O_NONBLOCK);
    #endif

    for (;;) {
        /* Accept connection if possible */
        if (client_count < sizeof client / sizeof *client) {
            struct sockaddr_storage addr = { 0 };
            socklen_t addr_len = sizeof addr;
            int fd = accept(server, &addr, &addr_len);
            if (fd != -1) {
#               ifdef WIN32
                ioctlsocket(fd, FIONBIO, (u_long[]){1});
#               else
                fcntl(fd, F_SETFL, fcntl(fd, F_GETFL, 0) | O_NONBLOCK);
#               endif
                client[client_count++] = (struct client) { .addr = addr
                                                         , .addr_len = addr_len
                                                         , .fd = fd };
            }
        }
        /* Loop through clients */
        char buffer[BUFFER_SIZE];
        for (size_t index = 0; index < client_count; index++) {
            ssize_t bytes_recvd = recv(client[index].fd, buffer, sizeof buffer, 0);
#           ifdef WIN32
            int closed = bytes_recvd == 0
                      || (bytes_recvd < 0 && WSAGetLastError() == WSAEWOULDBLOCK);
#           else
            int closed = bytes_recvd == 0
                      || (bytes_recvd < 0 && errno == EAGAIN) || errno == EWOULDBLOCK;
#           endif
            if (closed) {
                close(client[index].fd);
                client_count--;
                memmove(client + index, client + index + 1, (client_count - index) * sizeof client);
                continue;
            }
            /* XXX: Process buffer[0..bytes_recvd-1] */
        }

        sleep(0); /* This is necessary to pass control back to the kernel,
                   * so it can queue more data for us to process
                   */
    }
}

假设您想在客户端连接连接，代码看起来非常相似，但显然不需要accept 相关的代码。假设您有一个 clients 数组，您想要 connect，您可以使用非阻塞连接调用来一次执行所有连接，如下所示：

size_t index = 0, in_progress = 0;
for (;;) {
    if (client[index].fd == 0) {
        client[index].fd = socket(/* TODO */);
#       ifdef WIN32
        ioctlsocket(client[index].fd, FIONBIO, (u_long[]){1});
#       else
        fcntl(client[index].fd, F_SETFL, fcntl(client[index].fd, F_GETFL, 0) | O_NONBLOCK);
#       endif
    }
#   ifdef WIN32
    in_progress += connect(client[index].fd, (struct sockaddr *) &client[index].addr, client[index].addr_len) < 0
                && (WSAGetLastError() == WSAEALREADY
                ||  WSAGetLastError() == WSAEWOULDBLOCK
                ||  WSAGetLastError() == WSAEINVAL);
#   else
    in_progress += connect(client[index].fd, (struct sockaddr *) &client[index].addr, client[index].addr_len) < 0
                && (errno == EALREADY
                ||  errno == EINPROGRESS);
#   endif
    if (++index < sizeof client / sizeof *client) {
        continue;
    }
    index = 0;
    if (in_progress == 0) {
        break;
    }
    in_progress = 0;
}

至于优化，考虑到这应该能够处理 10000 个客户端，可能需要一些小的调整，所以您不需要多个线程。

尽管如此，通过将来自 mutex 集合的项目与 clients 相关联，并在 非阻塞套接字操作之前使用 非阻塞 pthread_mutex_trylock，上述循环可以适应于在多个线程中同时运行，同时处理同一组套接字。这为所有符合 POSIX 的平台提供了一个工作模型，无论是 Windows、BSD 还是 Linux，但它并不是一个完美的最佳模型。为了达到最优，我们必须进入 异步 世界，这个世界因系统而异：

• Windows 使用WSA* functions with call-backs。
• BSD 和 Linux 使用 the somewhat similar kqueue and epoll, respectively。

编码前面提到的“非阻塞套接字操作”抽象可能是值得的，因为这两种异步机制在接口方面存在很大差异。像其他一切一样，不幸的是，我们必须编写抽象，以便我们的 Windows 相关代码在 POSIX 兼容系统上仍然清晰易读。作为奖励，这将允许我们将服务器处理（即 accept 和随后的任何内容）与客户端处理（即 connect 和随后的任何内容）混合在一起，所以我们的 server 循环可以成为客户端循环（反之亦然）。