Java高级用法之绑定CPU的线程Thread Affinity简介

什么是Thread Affinity？

Thread Affinity（线程亲和性）是指将一个线程绑定到一个指定的 CPU 上面，使得线程只在这个特定的 CPU 上运行。在高性能计算和计算机游戏等领域，Thread Affinity 被广泛使用，以提高应用的执行效率。

Thread Affinity 的作用

Thread Affinity 的主要作用是优化 CPU cache、提高 CPU 性能和减少缓存污染。在 Java 应用程序中，CPU cache 缓存的存储器数据通常是以 CPU cache 行（line）为单位读入和写出的。当一个线程在多个 CPU 上来回切换时，它的 CPU cache 行可能会被其他线程修改或销毁，这会导致缓存污染和应用程序性能下降。

如何在Java中使用Thread Affinity

在Java中，JNA是一个使用Thread Affinity的非常优秀的库。JNA（Java Native Access）是一个可以在Java语言中调用C/C++等本地语言的库，使得Java应用程序可以直接使用本地代码中的函数和方法。

Thread Affinity的JNA页面可以在https://github.com/peter-lawrey/Java-Thread-Affinity找到JNA库，并且简单易用。下面将提供两个示例，用于说明如何在Java中使用Thread Affinity。

示例1：将线程绑定到指定的CPU上

import com.sun.jna.platform.win32.Kernel32;
import com.sun.jna.platform.win32.WinNT;
import java.util.HashSet;
import java.util.Set;

public class ThreadAffinityExample{

    public static void main(String[] args) {
        Set<Integer> cpuSet = new HashSet<>();
        cpuSet.add(0);  // 将线程绑定到第一个CPU上
        Kernel32 kernel32 = Kernel32.INSTANCE;
        WinNT.HANDLE handle = kernel32.GetCurrentThread();
        int processAffinityMask = kernel32.GetProcessAffinityMask(handle);
        int systemAffinityMask = kernel32.GetSystemAffinityMask(handle);
        int threadAffinityMask = processAffinityMask & systemAffinityMask;
        if(cpuSet.stream().anyMatch(i -> ((threadAffinityMask >> i) & 1) == 1)) {
            kernel32.SetThreadAffinityMask(handle, 1L << 0);
        } else {
            System.err.println("CPU #" + 0 + " is not supported.");
        }
    }
}

在这个示例中，我们使用了 JNA 库，并通过 Kernel32 类的实例化方法 INSTANCE 获取 JNA 平台上模拟的 Windows 操作系统的内核对象。我们使用了 GetProcessAffinityMask() 和 GetSystemAffinityMask() 方法来获取当前线程的 进程和系统仿真的CPU掩码。然后，我们将 Thread Affinity 设置为我们要绑定线程的 CPU 上。

示例2：为CPU核心分配线程

import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

public class ThreadAffinityExample2 {

    private static final int NR_CPU = Runtime.getRuntime().availableProcessors();

    private static final AtomicBoolean[] bound = new AtomicBoolean[NR_CPU];

    static {
        for (int i = 0; i < bound.length; i++) {
            bound[i] = new AtomicBoolean(false);
        }
    }

    public static void main(String[] args) throws InterruptedException {
        for (int i = 0; i < args.length; i++) {
            Runnable r = () -> {
                int id;
                while ((id = getNextUnbound()) < bound.length) {
                    if (bound[id].compareAndSet(false, true)) {
                        System.out.println("Thread " + Thread.currentThread().getId() + " bound to " + id);
                        cpuBind(id);
                        break;
                    }
                }
                try {
                    TimeUnit.SECONDS.sleep(3);
                } catch (InterruptedException ie) {
                    Thread.currentThread().interrupt();
                } finally {
                    if (id < bound.length) {
                        System.out.println(
                                "Thread " + Thread.currentThread().getId() + " unbound from " + id);
                        bound[id].set(false);
                    } else {
                        System.out.println("Thread " + Thread.currentThread().getId() + " can't find a CPU to bind.");
                    }
                }
            };
            new Thread(r).start();
        }
    }

    private static void cpuBind(int id) {
        String[] cmdarray = {"/bin/bash", "-c", "echo " + id + " > /proc/self/task/" + Thread.currentThread().getId() + "/cpuset"};
        try {
            Runtime.getRuntime().exec(cmdarray).waitFor();
        } catch (Exception e) {
            System.err.println("Thread " + Thread.currentThread().getId() + " failed to bind to CPU " + id);
        }
    }

    private static int getNextUnbound() {
        for (int i = 0; i < bound.length; i++) {
            if (!bound[i].get()) {
                return i;
            }
        }
        return bound.length;
    }

}

在这个示例中，我们定义了一个长度为NR_CPU的boolean类型数组，用于跟踪CPU核心是否已经与线程绑定。我们使用了getNextUnbound()方法来获取未绑定到CPU核心上的线程，并使用cpuBind()方法将线程绑定到发现的第一个未绑定的CPU核心上。最后，我们用了TimeUnit.SECONDS.sleep(3)来模拟线程在指定的CPU核心上执行命令的行为，并调用bound[id].set(false)来解除与当前线程绑定的CPU核心。

请注意，在示例中，为了绑定线程到指定的CPU核心上，我们使用了“cpuset”接口（/proc/self/task/<PID>/cpuset）。cpuset是Linux内核的一部分，它允许我们将一个或多个进程绑定到一个或多个CPU核心。但是，cpuset只能在Linux系统上使用，并且需要超级用户权限。