ZLToolKit使用多线程技术，每一个线程中都创建了自己私有的epoll实例(非linux系统使用select，不做分析)，并以ET模式监听同一个server fd的读(UDP还有写)事件，这种方式我们知道是有惊群效应的，所以需要给每一个fd事件加上EPOLLEXCLUSIVE标志位(内核4.5+支持)来避免惊群。后续客户端的fd将均匀的分配到这多个线程中处理。

每一个线程都能接受新的客户端连接，并且管理一部分已有的客户端连接。

1.4、UDP服务器如何实现监听socket与会话socket的分离

对于TCP服务器，由于TCP是有连接的，server fd使用listen系统调用，只用来处理client连接过程，然后使用accept为客户端连接分配一个新的client fd，后续针对该客户端的会话数据读写就和server fd没关系了。说白了，就是TCP四元组在这个过程中就唯一确定并缓存下来了，后续该客户端发送的数据，通过查询四元组，就可以唯一匹配到对应的client fd上。

可以看到，TCP服务器的监听socket与会话socket天然就是分开的，每一个客户端都有独立的client fd。

但是对于UDP服务器，由于它是无连接的，不用经过三次握手的过程就可以直接发送数据，一个UDP socket没有服务器客户端之分，对于服务器来说，一个socket就可以接受所有的客户端数据，这必然导致数据的处理变得非常复杂，比如，如何区分不同的客户端，如何高效且安全的在多线程下处理客户端数据等。

我们看下ZLToolKit是如何解决的。在TCP服务器中，有一个只处理客户端连接的socket(listen fd)，然后每一个客户端都有自己的client socket。接受新连接和客户端的数据交互是分开的。

而UDP中，这两者是混杂在一起的。所以，我们可以模拟TCP，当server fd对应的socket上收到某个客户端发来的数据后，不通过server socket去响应该客户端，而是新建立一个socket，该socket与server socket绑定相同的IP和端口(通过SO_REUSEADDR实现)，并且，为了确定唯一的四元组，在该socket上调用connect API与客户端进行了关联。这样，后续客户端发送的数据将被转发到这个新的socket上来处理。

当然，使用这种方式实现的UDP服务器还有很多弊端，比如客户端程序必须绑定固定的IP和端口，不然每次客户端都随机使用一个端口来发送数据，会导致服务器的四元组失效。更详细的介绍可以参见以下文章：

告知你不为人知的 UDP：连接性和负载均衡 - 知乎

二、代码实现

2.1、新socket的创建过程

?createSession函数中，以下三行代码就是创建新socket的过程，新socket与server socket绑定相同的IP和端口，然后再connect到peer客户端，确定四元组。

socket->bindUdpSock(_socket->get_local_port(), _socket->get_local_ip());
socket->bindPeerAddr((struct sockaddr *)addr_str.data(), addr_str.size());
//在connect peer后再取消绑定关系, 避免在 server 的 socket 或其他cloned server中收到后续数据包.
SockUtil::dissolveUdpSock(_socket->rawFD());

2.2、处理server fd多次收到同一client连接发送的数据

一般仅在一个新的客户端连接第一次发送数据时，data才会在server fd上被接收到，其它时间都会发送到已创建的新socket上。但这不是绝对的，所以服务器还需要处理多次收到同一client发送数据的情况。

void UdpServer::onRead(const Buffer::Ptr &buf, sockaddr *addr, int addr_len) {
    const auto id = makeSockId(addr, addr_len);
    onRead_l(true, id, buf, addr, addr_len);
}

void UdpServer::onRead_l(bool is_server_fd, const UdpServer::PeerIdType &id, const Buffer::Ptr &buf, sockaddr *addr, int addr_len) {
    // udp server fd收到数据时触发此函数；大部分情况下数据应该在peer fd触发，此函数应该不是热点函数
    bool is_new = false;
    if (auto session = getOrCreateSession(id, buf, addr, addr_len, is_new)) {
        if (session->getPoller()->isCurrentThread()) {
            //当前线程收到数据，直接处理数据
            session->onRecv(buf);
        } else {
            //数据漂移到其他线程，需要先切换线程
            WarnL << "udp packet incoming from other thread";
            std::weak_ptr<Session> weak_session = session;
            //由于socket读buffer是该线程上所有socket共享复用的，所以不能跨线程使用，必须先拷贝一下
            auto cacheable_buf = std::make_shared<BufferString>(buf->toString());
            session->async([weak_session, cacheable_buf]() {
                if (auto strong_session = weak_session.lock()) {
                    strong_session->onRecv(cacheable_buf);
                }
            });
        }

#if !defined(NDEBUG)
        if (!is_new) {
            TraceL << "udp packet incoming from " << (is_server_fd ? "server fd" : "other peer fd");
        }
#endif
    }
}

server fd接收到客户端数据时，获取已存在的会话或者创建一个新的会话(创建新的sokcet的过程)，该会话将处理后续所有这个客户端的数据，server fd将不会再收到该客户端的数据。

这里有个注意点，在创建新的会话时，可能客户端已经发送了多条数据了，这多次数据都发送到了server fd上，server fd在处理第一条数据时，虽然已经创建新的会话处理后续客户端发送的数据，但是已经发出来的数据还是会在server fd上处理，所以，此处才需要查找该客户端对应的已存在会话，让其来处理这部分已经发送出来的数据。

为什么要判断是不是当前线程呢？从上边的描述可以看到，server fd可能收到多次客户端发来的数据，在第一条数据时，我们已经创建了一个新的会话socket，这个会话socket可能会被放在其它poller线程去监听，所以，server fd上后续的数据，就需要转移到这个poller线程中，确保同一个client发送过来的数据始终在一个线程中处理。但实际上，ZLToolkit的实现中，将新的会话socket还是放在当前线程中处理了，并没有放在其它线程。以下代码可以看到，新socket使用的还是server socket的poller线程。

auto socket = createSocket(_poller, buf, addr, addr_len);

而在TCP服务器的实现中，是选择了负载最轻的线程来管理新的会话socket，所以存在socket被其它线程管理的情况，参见以下代码：

Socket::Ptr TcpServer::onBeforeAcceptConnection(const EventPoller::Ptr &poller) {
    assert(_poller->isCurrentThread()); //在_poller线程则允许继续执行，执行accept操作的就是当前_poller线程
    //此处改成自定义获取poller对象，防止负载不均衡
    return createSocket(EventPollerPool::Instance().getPoller(false));
}

2.3、新socket分配到其它线程可能存在的问题

createSession函数中，以下代码本人觉得可能存在问题。

   if (socket->getPoller()->isCurrentThread()) {
        //该socket分配在本线程，直接创建session对象，并处理数据
        return session_creator();
    }

    //该socket分配在其他线程，需要先拷贝buffer，然后在其所在线程创建session对象并处理数据
    auto cacheable_buf = std::make_shared<BufferString>(buf->toString());
    socket->getPoller()->async([session_creator, cacheable_buf]() {
        //在该socket所在线程创建session对象
        auto session = session_creator();
        if (session) {
            //该数据不能丢弃，给session对象消费
            session->onRecv(cacheable_buf);
        }
    });
    return s_null_session;

如果新的会话socket分配在其它线程，将使用异步任务执行，假设会话在还没有创建成功前，当前server线程再次收到该客户端数据，在_session_map中没有查询到对应的会话，就会再次触发创建新会话的操作。

目前的实现新会话socket还是在当前线程，所以不存在上述问题。