遊戲伺服器中的Netty應用以及原始碼剖析

2022-08-26 18:03:32

一、Reactor模式和Netty執行緒模型

最近因為工作需要，學習了一段時間Netty的原始碼，並做了一個簡單的分享，研究還不是特別深入，繼續努力。因為分享也不涉及公司業務，所以這裡也把這次對原始碼的研究成果分享出來以下都是在遊戲伺服器開發中針對Netty使用需要了解知識點以及相關優化

這次分享主要設計以下內容

Netty執行緒模型
Netty對TCP相關引數的設定和具體含義
Netty對Epoll的封裝
Netty的優雅關閉

使用者端連線數的限制

記憶體資源
CPU資源

埠號資源

cat /proc/sys/net/ipv4/ip_local_port_range

檔案描述符資源

系統級：當前系統可開啟的最大數量，通過 cat /proc/sys/fs/file-max 檢視
使用者級：指定使用者可開啟的最大數量，通過 cat /etc/security/limits.conf 檢視
程序級：單個程序可開啟的最大數量，通過 cat /proc/sys/fs/nr_open 檢視
執行緒資源 BIO/NIO

1. BIO模型

所有操作都是同步阻塞（accept，read）
使用者端連線數與伺服器執行緒數比例是1:1

2. NIO模型

非阻塞IO
通過selector實現可以一個執行緒管理多個連線
通過selector的事件註冊（OP_READ/OP_WRITE/OP_CONNECT/OP_ACCEPT），處理自己感興趣的事件

使用者端連線數與伺服器執行緒數比例是n:1

3. Reacor模型

①. 單Reacor單執行緒模型

所有IO在同一個NIO執行緒完成（處理連線，分派請求，編碼，解碼，邏輯運算，傳送）

優點：

編碼簡單
不存在共用資源競爭
並行安全

缺點：

單執行緒處理大量鏈路時，效能無法支撐，不能合理利用多核處理
執行緒過載後，處理速度變慢，會導致訊息積壓
一旦執行緒掛掉，整個通訊層不可用 redis使用的就是reactor單程序模型，redis由於都是記憶體級操作，所以使用此模式沒什麼問題

reactor單執行緒模型圖：

netty reactor單執行緒模型圖：

Netty對應實現方式

// Netty對應實現方式：建立io執行緒組是，boss和worker，使用同一個執行緒組，並且執行緒數為1
EventLoopGroup ioGroup = new NioEventLoopGroup(1);
b.group(ioGroup, ioGroup)
        .channel(NioServerSocketChannel.class)
        .childHandler(initializer);
ChannelFuture f = b.bind(portNumner);
cf = f.sync();
f.get();

②. 單Reactor多執行緒模型

根據單執行緒模型，io處理中最耗時的編碼，解碼，邏輯運算等cpu消耗較多的部分，可提取出來使用多執行緒實現，並充分利用多核cpu的優勢

優點：

多執行緒處理邏輯運算，提高多核CPU利用率

缺點：

對於單Reactor來說，大量連結的IO事件處理依然是效能瓶頸

reactor多執行緒模型圖：

netty reactor多執行緒模型圖：

Netty對應實現方式

// Netty對應實現方式：建立io執行緒組是，boss和worker，使用同一個執行緒組，並且執行緒數為1，把邏輯運算部分投遞到使用者自定義執行緒處理
EventLoopGroup ioGroup = new NioEventLoopGroup(1);
b.group(ioGroup, ioGroup)
        .channel(NioServerSocketChannel.class)
        .childHandler(initializer);
ChannelFuture f = b.bind(portNumner);
cf = f.sync();
f.get();

③. 主從Reactor多執行緒模型

根據多執行緒模型，可把它的效能瓶頸做進一步優化，即把reactor由單個改為reactor執行緒池，把原來的reactor分為mainReactor和subReactor

優點：

解決單Reactor的效能瓶頸問題（Netty/Nginx採用這種設計）

reactor主從多執行緒模型圖：

netty reactor主從多執行緒模型圖：

Netty對應實現方式

// Netty對應實現方式：建立io執行緒組boss和worker，boss執行緒數為1，work執行緒數為cpu*2(一般IO密集可設定為2倍cpu核數)
EventLoopGroup bossGroup = new NioEventLoopGroup(1);
EventLoopGroup workerGroup = new NioEventLoopGroup();
b.group(bossGroup, workerGroup)
        .channel(NioServerSocketChannel.class)
        .childHandler(initializer);
ChannelFuture f = b.bind(portNumner);
cf = f.sync();
f.get();

④. 部分原始碼分析

建立group範例

// 1.構造引數不傳或傳0，預設取系統引數設定，沒有引數設定，取CPU核數*2
super(nThreads == 0 ? DEFAULT_EVENT_LOOP_THREADS : nThreads, executor, args);
private static final int DEFAULT_EVENT_LOOP_THREADS;
static {
    DEFAULT_EVENT_LOOP_THREADS = Math.max(1, SystemPropertyUtil.getInt(
            "io.netty.eventLoopThreads", NettyRuntime.availableProcessors() * 2));
}
// 2.不同版本的JDK會有不同版本的SelectorProvider實現，Windows下的是WindowsSelectorProvider
public NioEventLoopGroup(int nThreads, Executor executor) {
    //預設selector,最終實現類似：https://github.com/frohoff/jdk8u-jdk/blob/master/src/macosx/classes/sun/nio/ch/DefaultSelectorProvider.java
    //basic flow: 1 java.nio.channels.spi.SelectorProvider 2 META-INF/services 3 default
    this(nThreads, executor, SelectorProvider.provider());
}
// 3.建立nThread個EventExecutor，並封裝到選擇器chooser，chooser會根據執行緒數分別有兩種實現（GenericEventExecutorChooser和PowerOfTwoEventExecutorChooser，演演算法不同，但實現邏輯一樣，就是均勻的分配執行緒處理）
EventExecutorChooserFactory.EventExecutorChooser chooser;
children = new EventExecutor[nThreads];
for (int i = 0; i < nThreads; i ++) {
    // ...
    children[i] = newChild(executor, args);
    // ...
}
chooser = chooserFactory.newChooser(children);

設定group

// 兩種方式設定group
// parent和child使用同一個group，呼叫仍然是分別設定parent和child
@Override
public ServerBootstrap group(EventLoopGroup group) {
    return group(group, group);
}
ServerBootstrap.group(EventLoopGroup parentGroup, EventLoopGroup childGroup){
    // 具體程式碼略，可直接參考原始碼
    // 裡面實現內容是把parentGroup繫結到this.group，把childGroup繫結到this.childGroup
}

Netty啟動

// 呼叫順序
ServerBootstrap:bind() -> doBind() -> initAndRegister()
private ChannelFuture doBind(final SocketAddress localAddress) {
    final ChannelFuture regFuture = initAndRegister();
    // ...
    doBind0(regFuture, channel, localAddress, promise);
    // ...
}
final ChannelFuture initAndRegister() {
    // 建立ServerSocketChannel
    Channel channel = channelFactory.newChannel();
    // ...
    // 開始register
    ChannelFuture regFuture = config().group().register(channel);
    // register呼叫順序
    // next().register(channel) -> (EventLoop) super.next() -> chooser.next()
    // ...
}

由以上原始碼可得知，bind只在起服呼叫一次，因此bossGroup僅呼叫一次regist，也就是僅呼叫一次next，因此只有一根執行緒是有用的，其餘執行緒都是廢棄的，所以bossGroup執行緒數設定為1即可

// 啟動BossGroup執行緒並繫結本地SocketAddress
private static void doBind0(
        final ChannelFuture regFuture, final Channel channel,
        final SocketAddress localAddress, final ChannelPromise promise) {
    channel.eventLoop().execute(new Runnable() {
        @Override
        public void run() {
            if (regFuture.isSuccess()) {
                channel.bind(localAddress, promise).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
            } else {
                promise.setFailure(regFuture.cause());
            }
        }
    });
}

使用者端連線

// 訊息事件讀取
NioEventLoop.run() -> processSelectedKeys() -> ... -> ServerBootstrapAcceptor.channelRead
// ServerBootstrapAcceptor.channelRead處理使用者端連線事件
// 最後一行的childGroup.register的邏輯和上面的程式碼呼叫處一樣
public void channelRead(ChannelHandlerContext ctx, Object msg) {
    child.pipeline().addLast(childHandler);
    setChannelOptions(child, childOptions, logger);
    setAttributes(child, childAttrs);
    childGroup.register(child)
}

二、select/poll和epoll

1.概念

select(時間複雜度O(n))：用一個fd陣列儲存所有的socket，然後通過死迴圈遍歷呼叫作業系統的select方法找到就緒的fd

while(1) {
  nready = select(list);
  // 使用者層依然要遍歷，只不過少了很多無效的系統呼叫
  for(fd <-- fdlist) {
    if(fd != -1) {
      // 唯讀已就緒的檔案描述符
      read(fd, buf);
      // 總共只有 nready 個已就緒描述符，不用過多遍歷
      if(--nready == 0) break;
    }
  }
}

poll(時間複雜度O(n))：同select，不過把fd陣列換成了fd連結串列，去掉了fd最大連線數（1024個）的數量限制

epoll(時間複雜度O(1))：解決了select/poll的幾個缺陷

呼叫需傳入整個fd陣列或fd連結串列，需要拷貝資料到核心
核心層需要遍歷檢查檔案描述符的就緒狀態
核心僅返回可讀檔案描述符個數，使用者仍需自己遍歷所有fd

epoll是作業系統基於事件關聯fd，做了以下優化：

核心中儲存一份檔案描述符集合，無需使用者每次都重新傳入，只需告訴核心修改的部分即可。（epoll_ctl）
核心不再通過輪詢的方式找到就緒的檔案描述符，而是通過非同步 IO 事件喚醒。（epoll_wait）
核心僅會將有 IO 事件的檔案描述符返回給使用者，使用者也無需遍歷整個檔案描述符集合。

epoll僅在Linux系統上支援

2.jdk提供selector

// DefaultSelectorProvider.create方法在不同版本的jdk下有不同實現，建立不同Selector
// Windows版本的jdk，其實現中呼叫的是native的poll方法
public static SelectorProvider create() {
    return new WindowsSelectorProvider();
}
// Linux版本的jdk
public static SelectorProvider create() {
    String str = (String)AccessController.doPrivileged(new GetPropertyAction("os.name"));
    if (str.equals("SunOS")) {
        return createProvider("sun.nio.ch.DevPollSelectorProvider");
    }
    if (str.equals("Linux")) {
        return createProvider("sun.nio.ch.EPollSelectorProvider");
    }
    return new PollSelectorProvider();
}

3.Netty提供的Epoll封裝

netty依然基於epoll做了一層封裝，主要做了以下事情：

（1）java的nio預設使用水平觸發，Netty的Epoll預設使用邊緣觸發，且可設定

邊緣觸發：當狀態變化時才會發生io事件。
水平觸發：只要滿足條件，就觸發一個事件（只要有資料沒有被獲取，核心就不斷通知你）

（2）Netty的Epoll提供更多的nio的可配引數。

（3）呼叫c程式碼，更少gc，更少synchronized 具體可以參考原始碼NioEventLoop.run和EpollEventLoop.run進行對比

4.Netty相關類圖

執行緒組類圖

channel類圖

5.設定Netty為EpollEventLoop

// 建立指定的EventLoopGroup
bossGroup = new EpollEventLoopGroup(1, new DefaultThreadFactory("BOSS_LOOP"));
workerGroup = new EpollEventLoopGroup(32, new DefaultThreadFactory("IO_LOOP"));
b.group(bossGroup, workerGroup)
        // 指定channel的class
        .channel(EpollServerSocketChannel.class)
        .childHandler(initializer);
// 其中channel(clz)方法是通過class來new一個反射ServerSocketChannel建立工廠類
public B channel(Class<? extends C> channelClass) {
    if (channelClass == null) {
        throw new NullPointerException("channelClass");
    }
    return channelFactory(new ReflectiveChannelFactory<C>(channelClass));
}
final ChannelFuture initAndRegister() {
    // ...
    Channel channel = channelFactory.newChannel();
    // ...
}

三、Netty相關引數

1.SO_KEEPALIVE

childOption(ChannelOption.SO_KEEPALIVE, true)

TCP鏈路探活

2.SO_REUSEADDR

option(ChannelOption.SO_REUSEADDR, true)

重用處於TIME_WAIT但是未完全關閉的socket地址，讓埠釋放後可立即被重用。預設關閉，需要手動開啟

3.TCP_NODELAY

childOption(ChannelOption.TCP_NODELAY, true)

IP報文格式

TCP報文格式

開啟則禁用TCP Negal演演算法，優點低延時，缺點在大量小封包的情況下，網路利用率低

關閉則開啟TCP Negal演演算法，優點提高網路利用率（資料快取到一定量才傳送），缺點延時高

Negal演演算法

如果包長度達到MSS（maximum segment size最大分段長度），則允許傳送；
如果該包含有FIN，則允許傳送；
設定了TCP_NODELAY選項，則允許傳送；
未設定TCP_CORK選項（是否阻塞不完整報文）時，若所有發出去的小封包（包長度小於MSS）均被確認，則允許傳送；
上述條件都未滿足，但發生了超時（一般為200ms），則立即傳送。

MSS計算規則 MSS的值是在TCP三次握手建立連線的過程中，經通訊雙方協商確定的 802.3標準裡，規定了一個以太幀的資料部分(Payload)的最大長度是1500個位元組(MTU)

MSS = MTU - IP首部 - TCP首部
乙太網環境下：
MTU = 1500位元組
IP首部 = 32*5/4 = 160bit = 20位元組
TCP首部 = 32*5/4 = 160bit = 20位元組
最終得出MSS = 1460位元組

結論：因為遊戲伺服器的實時性要求，在網路頻寬足夠的情況下，建議開啟TCP_NODELAY，關閉Negal演演算法，頻寬可以浪費，響應必須及時

注意：需要使用者端伺服器均關閉Negal演演算法，否則仍然會有延遲傳送，影響傳輸速度

4.SO_BACKLOG

option(ChannelOption.SO_BACKLOG, 100)

作業系統核心中維護的兩個佇列

syns queue：儲存syn到達，但沒完成三次握手的半連線

cat /proc/sys/net/ipv4/tcp_max_syn_backlog

accpet queue：儲存完成三次握手，核心等待accept呼叫的連線

cat /proc/sys/net/core/somaxconn

netty對於backlog的預設值設定在NetUtil類253行

SOMAXCONN = AccessController.doPrivileged(new PrivilegedAction<Integer>() {
    @Override
    public Integer run() {
        // 1.設定預設值
        int somaxconn = PlatformDependent.isWindows() ? 200 : 128;
        File file = new File("/proc/sys/net/core/somaxconn");
        if (file.exists()) {
            // 2.檔案存在，讀取作業系統設定
            in = new BufferedReader(new FileReader(file));
            somaxconn = Integer.parseInt(in.readLine());
        } else {
            // 3.檔案不存在，從各個引數中讀取
            if (SystemPropertyUtil.getBoolean("io.netty.net.somaxconn.trySysctl", false)) {
                tmp = sysctlGetInt("kern.ipc.somaxconn");
                if (tmp == null) {
                    tmp = sysctlGetInt("kern.ipc.soacceptqueue");
                    if (tmp != null) {
                        somaxconn = tmp;
                    }
                } else {
                    somaxconn = tmp;
                }
            }
        }
    }
}

結論：

Linux下/proc/sys/net/core/somaxconn一定存在，所以backlog一定取得它的值，我參考prod機器的引數設定的65535，也就是不設定backlog的情況下，伺服器執行快取65535個全連線

5.ALLOCATOR和RCVBUF_ALLOCATOR

預設分配ByteBuffAllocator賦值如下： ByteBufUtil.java

static {
    //以io.netty.allocator.type為準，沒有的話，安卓平臺用非池化實現，其他用池化實現
    String allocType = SystemPropertyUtil.get(
            "io.netty.allocator.type", PlatformDependent.isAndroid() ? "unpooled" : "pooled");
    allocType = allocType.toLowerCase(Locale.US).trim();
    ByteBufAllocator alloc;
    if ("unpooled".equals(allocType)) {
        alloc = UnpooledByteBufAllocator.DEFAULT;
        logger.debug("-Dio.netty.allocator.type: {}", allocType);
    } else if ("pooled".equals(allocType)) {
        alloc = PooledByteBufAllocator.DEFAULT;
        logger.debug("-Dio.netty.allocator.type: {}", allocType);
    } else {
        //io.netty.allocator.type設定的不是"unpooled"或者"pooled"，就用池化實現。
        alloc = PooledByteBufAllocator.DEFAULT;
        logger.debug("-Dio.netty.allocator.type: pooled (unknown: {})", allocType);
    }
    DEFAULT_ALLOCATOR = alloc;
}

RCVBUF_ALLOCATOR預設AdaptiveRecvByteBufAllocator

public class DefaultChannelConfig implements ChannelConfig {
    // ...
    public DefaultChannelConfig(Channel channel) {
        this(channel, new AdaptiveRecvByteBufAllocator());
    }
    // ...
}

四、Netty關閉

/**
 * Shortcut method for {@link #shutdownGracefully(long, long, TimeUnit)} with sensible default values.
 *
 * @return the {@link #terminationFuture()}
 */
Future<?> shutdownGracefully();
/**
 * Signals this executor that the caller wants the executor to be shut down.  Once this method is called,
 * {@link #isShuttingDown()} starts to return {@code true}, and the executor prepares to shut itself down.
 * Unlike {@link #shutdown()}, graceful shutdown ensures that no tasks are submitted for <i>'the quiet period'</i>
 * (usually a couple seconds) before it shuts itself down.  If a task is submitted during the quiet period,
 * it is guaranteed to be accepted and the quiet period will start over.
 *
 * @param quietPeriod the quiet period as described in the documentation
                     靜默期：在此期間，仍然可以提交任務
 * @param timeout     the maximum amount of time to wait until the executor is {@linkplain #shutdown()}
 *                    regardless if a task was submitted during the quiet period
                     超時時間：等待所有任務執行完的最大時間
 * @param unit        the unit of {@code quietPeriod} and {@code timeout}
 *
 * @return the {@link #terminationFuture()}
 */
Future<?> shutdownGracefully(long quietPeriod, long timeout, TimeUnit unit);
// 抽象類中的實現
static final long DEFAULT_SHUTDOWN_QUIET_PERIOD = 2;
static final long DEFAULT_SHUTDOWN_TIMEOUT = 15;
@Override
public Future<?> shutdownGracefully() {
    return shutdownGracefully(DEFAULT_SHUTDOWN_QUIET_PERIOD, DEFAULT_SHUTDOWN_TIMEOUT, TimeUnit.SECONDS);
}

把NIO執行緒的狀態位設定成ST_SHUTTING_DOWN狀態，不再處理新的訊息（不允許再對外傳送訊息）；
退出前的預處理操作：把傳送佇列中尚未傳送或者正在傳送的訊息傳送完、把已經到期或者在退出超時之前到期的定時任務執行完成、把使用者註冊到NIO執行緒的退出Hook任務執行完成；
資源的釋放操作：所有Channel的釋放、多路複用器的去註冊和關閉、所有佇列和定時任務的清空取消，最後是NIO執行緒的退出。

以上就是遊戲伺服器中的Netty應用以及原始碼剖析的詳細內容，更多關於Netty遊戲伺服器的資料請關注it145.com其它相關文章！