快捷導(dǎo)航

基于Java在netty中實(shí)現(xiàn)線程和CPU綁定

更新時(shí)間：2022年05月20日 08:51:23 作者：程序那些事

這篇文章主要介紹了基于Java在netty中實(shí)現(xiàn)線程和CPU綁定，文章圍繞主題的相關(guān)內(nèi)容展開(kāi)詳細(xì)介紹，具有一定的參考價(jià)值，需要的小伙伴可以參考一下

簡(jiǎn)介

使用java thread affinity庫(kù)我們可以將線程綁定到特定的CPU或者CPU核上，通過(guò)減少線程在CPU之間的切換，從而提升線程執(zhí)行的效率。

雖然netty已經(jīng)夠優(yōu)秀了，但是誰(shuí)不想更加優(yōu)秀一點(diǎn)呢？于是一個(gè)想法產(chǎn)生了，那就是能不能把a(bǔ)ffinity庫(kù)用在netty中呢？答案是肯定的，一起來(lái)看看吧。

引入affinity

affinity是以jar包的形式提供出去的，目前最新的正式版本是3.20.0，所以我們需要這樣引入：

<!-- https://mvnrepository.com/artifact/net.openhft/affinity -->
<dependency>
    <groupId>net.openhft</groupId>
    <artifactId>affinity</artifactId>
    <version>3.20.0</version>
</dependency>

引入affinity之后，會(huì)在項(xiàng)目的依賴(lài)庫(kù)中添加一個(gè)affinity的lib包，這樣我們就可以在netty中愉快的使用affinity了。

AffinityThreadFactory

有了affinity，怎么把a(bǔ)ffinity引入到netty中呢?

我們知道affinity是用來(lái)控制線程的，也就是說(shuō)affinity是跟線程有關(guān)的。而netty中跟線程有關(guān)的就是EventLoopGroup,先看一下netty中EventLoopGroup的基本用法，這里以NioEventLoopGroup為例,NioEventLoopGroup有很多構(gòu)造函數(shù)的參數(shù)，其中一種是傳入一個(gè)ThreadFactory：

    public NioEventLoopGroup(ThreadFactory threadFactory) {
        this(0, threadFactory, SelectorProvider.provider());
    }

這個(gè)構(gòu)造函數(shù)表示NioEventLoopGroup中使用的線程都是由threadFactory創(chuàng)建而來(lái)的。這樣以來(lái)我們就找到了netty和affinity的對(duì)應(yīng)關(guān)系。只需要構(gòu)造affinity的ThreadFactory即可。

剛好affinity中有一個(gè)AffinityThreadFactory類(lèi)，專(zhuān)門(mén)用來(lái)創(chuàng)建affinity對(duì)應(yīng)的線程。

接下來(lái)我們來(lái)詳細(xì)了解一下AffinityThreadFactory。

AffinityThreadFactory可以根據(jù)提供的不同AffinityStrategy來(lái)創(chuàng)建對(duì)應(yīng)的線程。AffinityStrategy表示的是線程之間的關(guān)系。

在affinity中，有5種線程關(guān)系，分別是：

SAME_CORE - 線程會(huì)運(yùn)行在同一個(gè)CPU core中。
SAME_SOCKET - 線程會(huì)運(yùn)行在同一個(gè)CPU socket中，但是不在同一個(gè)core上。
DIFFERENT_SOCKET - 線程會(huì)運(yùn)行在不同的socket中。
DIFFERENT_CORE - 線程會(huì)運(yùn)行在不同的core上。
ANY - 只要是可用的CPU資源都可以。

這些關(guān)系是通過(guò)AffinityStrategy中的matches方法來(lái)實(shí)現(xiàn)的：

boolean matches(int cpuId, int cpuId2);

matches傳入兩個(gè)參數(shù)，分別是傳入的兩個(gè)cpuId。

我們以SAME_CORE為例來(lái)看看這個(gè)mathes方法到底是怎么工作的：

    SAME_CORE {
        @Override
        public boolean matches(int cpuId, int cpuId2) {
            CpuLayout cpuLayout = AffinityLock.cpuLayout();
            return cpuLayout.socketId(cpuId) == cpuLayout.socketId(cpuId2) &&
                    cpuLayout.coreId(cpuId) == cpuLayout.coreId(cpuId2);
        }
    }

可以看到它的邏輯是先獲取當(dāng)前CPU的layout，CpuLayout中包含了cpu個(gè)數(shù)，sockets個(gè)數(shù)，每個(gè)sockets的cpu核數(shù)等基本信息。

并且提供了三個(gè)方法根據(jù)給定的cpuId返回對(duì)應(yīng)的socket、core和thread信息：

    int socketId(int cpuId);
    int coreId(int cpuId);
    int threadId(int cpuId);

matches方法就是根據(jù)傳入的cpuId找到對(duì)應(yīng)的socket,core信息進(jìn)行比較，從而生成了5中不同的策略。

先看一下AffinityThreadFactory的構(gòu)造函數(shù)：

    public AffinityThreadFactory(String name, boolean daemon, @NotNull AffinityStrategy... strategies) {
        this.name = name;
        this.daemon = daemon;
        this.strategies = strategies.length == 0 ? new AffinityStrategy[]{AffinityStrategies.ANY} : strategies;
    }

可以傳入thread的name前綴，和是否是守護(hù)線程，最后如果strategies不傳的話(huà)，默認(rèn)使用的是AffinityStrategies.ANY策略，也就是說(shuō)為線程分配任何可以綁定的CPU。

接下來(lái)看下這個(gè)ThreadFactory是怎么創(chuàng)建新線程的：

public synchronized Thread newThread(@NotNull final Runnable r) {
        String name2 = id <= 1 ? name : (name + '-' + id);
        id++;
        Thread t = new Thread(new Runnable() {
            @Override
            public void run() {
                try (AffinityLock ignored = acquireLockBasedOnLast()) {
                    r.run();
                }
            }
        }, name2);
        t.setDaemon(daemon);
        return t;
    }
    private synchronized AffinityLock acquireLockBasedOnLast() {
        AffinityLock al = lastAffinityLock == null ? AffinityLock.acquireLock() : lastAffinityLock.acquireLock(strategies);
        if (al.cpuId() >= 0)
            lastAffinityLock = al;
        return al;
    }

從上面的代碼可以看出，創(chuàng)建的新線程會(huì)以傳入的name為前綴，后面添加1，2，3，4這種后綴。并且根據(jù)傳入的是否是守護(hù)線程的標(biāo)記，將調(diào)用對(duì)應(yīng)線程的setDaemon方法。

重點(diǎn)是Thread內(nèi)部運(yùn)行的Runnable內(nèi)容，在run方法內(nèi)部，首先調(diào)用acquireLockBasedOnLast方法獲取lock，在獲得lock的前提下運(yùn)行對(duì)應(yīng)的線程方法，這樣就會(huì)將當(dāng)前運(yùn)行的Thread和CPU進(jìn)行綁定。

從acquireLockBasedOnLast方法中，我們可以看出AffinityLock實(shí)際上是一個(gè)鏈?zhǔn)浇Y(jié)構(gòu)，每次請(qǐng)求的時(shí)候都調(diào)用的是lastAffinityLock的acquireLock方法，如果獲取到lock，則將lastAffinityLock進(jìn)行替換，用來(lái)進(jìn)行下一個(gè)lock的獲取。

有了AffinityThreadFactory，我們只需要在netty的使用中傳入AffinityThreadFactory即可。

在netty中使用AffinityThreadFactory

上面講到了要在netty中使用affinity，可以將AffinityThreadFactory傳入EventLoopGroup中。對(duì)于netty server來(lái)說(shuō)可以有兩個(gè)EventLoopGroup，分別是acceptorGroup和workerGroup，在下面的例子中我們將AffinityThreadFactory傳入workerGroup，這樣后續(xù)work中分配的線程都會(huì)遵循AffinityThreadFactory中配置的AffinityStrategies策略，來(lái)獲得對(duì)應(yīng)的CPU：

//建立兩個(gè)EventloopGroup用來(lái)處理連接和消息
        EventLoopGroup acceptorGroup = new NioEventLoopGroup(acceptorThreads);
        //創(chuàng)建AffinityThreadFactory
        ThreadFactory threadFactory = new AffinityThreadFactory("affinityWorker", AffinityStrategies.DIFFERENT_CORE,AffinityStrategies.DIFFERENT_SOCKET,AffinityStrategies.ANY);
        //將AffinityThreadFactory加入workerGroup
        EventLoopGroup workerGroup = new NioEventLoopGroup(workerThreads,threadFactory);
        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(acceptorGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        public void initChannel(SocketChannel ch) throws Exception {
                            ch.pipeline().addLast(new AffinityServerHandler());
                        }
                    })
                    .option(ChannelOption.SO_BACKLOG, 128)
                    .childOption(ChannelOption.SO_KEEPALIVE, true);

            // 綁定端口并開(kāi)始接收連接
            ChannelFuture f = b.bind(port).sync();

            // 等待server socket關(guān)閉
            f.channel().closeFuture().sync();
        } finally {
            //關(guān)閉group
            workerGroup.shutdownGracefully();
            acceptorGroup.shutdownGracefully();
        }

為了獲取更好的性能，Affinity還可以對(duì)CPU進(jìn)行隔離，被隔離的CPU只允許執(zhí)行本應(yīng)用的線程，從而獲得更好的性能。

要使用這個(gè)特性需要用到linux的isolcpus。這個(gè)功能主要是將一個(gè)或多個(gè)CPU獨(dú)立出來(lái)，用來(lái)執(zhí)行特定的Affinity任務(wù)。

isolcpus命令后面可以接CPU的ID,或者可以修改/boot/grub/grub.conf文件，添加要隔離的CPU信息如下：