JAVA并发中CPU亲和性差导致线程抖动的系统级优化指南
各位朋友，大家好！今天我们来探讨一个经常被忽视但又至关重要的并发性能优化话题：JAVA并发中CPU亲和性差导致线程抖动的系统级优化。在多核处理器时代，充分利用CPU资源是提升并发程序性能的关键。然而，如果线程在不同的CPU核心之间频繁迁移，就会产生严重的性能问题，即线程抖动。
什么是线程抖动？
线程抖动（Thread Thrashing）指的是线程在不同的CPU核心之间频繁切换运行的状态。每次线程切换都需要刷新CPU缓存，导致之前缓存的数据失效，从而需要重新从内存中加载数据。这个过程会消耗大量的CPU时间，降低程序的整体性能，同时也增加了延迟。
想象一下，一个厨师（线程）需要不断地在不同的厨房（CPU核心）之间切换，每次切换都需要重新熟悉厨房的布局，找到需要的食材和工具。这种频繁的切换肯定会降低他的工作效率。
CPU亲和性：将线程绑定到特定的CPU核心
CPU亲和性（CPU Affinity）是指将一个线程或进程绑定到特定的CPU核心上运行。通过设置CPU亲和性，可以避免线程在不同的CPU核心之间迁移，从而提高CPU缓存的命中率，减少线程抖动，提升程序的性能。
为什么要关注CPU亲和性？
- 缓存命中率提升： 当线程固定在某个CPU核心上运行时，它可以有效地利用该核心的缓存，减少从主内存中读取数据的次数，从而降低延迟。
- 减少上下文切换开销： 线程迁移会导致上下文切换开销增加，包括保存和恢复线程的状态、刷新TLB（Translation Lookaside Buffer）等。CPU亲和性可以减少上下文切换，降低开销。
- NUMA架构优化： 在NUMA（Non-Uniform Memory Access）架构下，每个CPU核心都有自己的本地内存。将线程绑定到与其本地内存相关的CPU核心上，可以减少跨NUMA节点的内存访问，从而提高性能。
如何检测线程抖动？
在优化之前，我们需要先检测是否存在线程抖动的问题。以下是一些常用的检测方法：
-
系统监控工具： 使用
vmstat、top、htop等系统监控工具，观察CPU的使用率、上下文切换次数（cs）和进程迁移次数（mig）。如果上下文切换次数和进程迁移次数很高，而CPU利用率不高，则可能存在线程抖动。 -
性能分析工具： 使用
perf、oprofile等性能分析工具，可以更详细地分析CPU的使用情况，包括每个线程在不同CPU核心上的运行时间、缓存命中率等。 -
JVM监控工具： 使用
JConsole、VisualVM等JVM监控工具，可以监控JVM的线程活动，包括线程的状态、CPU使用率等。 -
自定义监控： 在代码中添加自定义监控，记录线程在不同CPU核心上的运行时间。可以使用
ThreadMXBean获取线程的CPU时间，并使用Thread.currentThread().getId()获取线程ID。
使用vmstat检测线程抖动的例子：
在Linux终端中运行vmstat 1，每隔1秒输出一次系统状态信息。关注以下几列：
- cs： 每秒上下文切换次数。
- in： 每秒中断次数。
- us： 用户态CPU时间百分比。
- sy： 系统态CPU时间百分比。
- id： 空闲CPU时间百分比。
- wa： 等待I/O的CPU时间百分比。
如果cs和in的值很高，而us和sy的值不高，id的值较高，则可能存在线程抖动。
如何设置CPU亲和性？
设置CPU亲和性的方法取决于操作系统和编程语言。以下是一些常用的方法：
-
Linux：
-
taskset命令： 可以使用taskset命令在启动进程时设置CPU亲和性。例如，将进程绑定到CPU核心0和1：taskset -c 0,1 ./my_java_program -
pthread_setaffinity_np函数： 在C/C++程序中，可以使用pthread_setaffinity_np函数设置线程的CPU亲和性。 -
Java： 理论上Java本身不直接提供设置CPU亲和性的API。但是，可以通过JNI调用底层的
pthread_setaffinity_np函数来实现。或者，利用一些开源库，例如Java Native Access (JNA)来调用系统API。
-
-
Windows：
-
SetProcessAffinityMask和SetThreadAffinityMask函数： 可以使用这两个函数设置进程或线程的CPU亲和性。 -
任务管理器： 在任务管理器中，可以右键单击进程，选择“设置优先级”，然后选择“设置关联性”，选择要绑定的CPU核心。
-
Java中使用JNA设置CPU亲和性的例子：
import com.sun.jna.Library;
import com.sun.jna.Native;
import com.sun.jna.Pointer;
import com.sun.jna.Structure;
import java.util.Arrays;
import java.util.List;
public class CpuAffinity {
public interface Libc extends Library {
Libc INSTANCE = (Libc) Native.load("c", Libc.class);
int sched_setaffinity(int pid, int cpusetsize, Pointer cpuset);
int sched_getaffinity(int pid, int cpusetsize, Pointer cpuset);
}
public static class CpuSet extends Structure {
public long[] cpu_mask = new long[16]; // Assuming max 1024 CPUs
@Override
protected List<String> getFieldOrder() {
return Arrays.asList("cpu_mask");
}
public void zero() {
Arrays.fill(cpu_mask, 0);
}
public void set(int cpu) {
int idx = cpu / 64;
int bit = cpu % 64;
cpu_mask[idx] |= (1L << bit);
}
public boolean isSet(int cpu) {
int idx = cpu / 64;
int bit = cpu % 64;
return (cpu_mask[idx] & (1L << bit)) != 0;
}
@Override
public String toString() {
StringBuilder sb = new StringBuilder();
for (int i = 0; i < cpu_mask.length; i++) {
sb.append(String.format("0x%016X ", cpu_mask[i]));
}
return sb.toString();
}
}
public static boolean setAffinity(int pid, int cpu) {
CpuSet cpuset = new CpuSet();
cpuset.zero();
cpuset.set(cpu);
int result = Libc.INSTANCE.sched_setaffinity(pid, Native.getNativeSize(CpuSet.class), cpuset.getPointer());
return result == 0;
}
public static CpuSet getAffinity(int pid) {
CpuSet cpuset = new CpuSet();
int result = Libc.INSTANCE.sched_getaffinity(pid, Native.getNativeSize(CpuSet.class), cpuset.getPointer());
if (result == 0) {
cpuset.read();
return cpuset;
} else {
return null;
}
}
public static void main(String[] args) {
int pid = 0; // Get current process ID
String osName = System.getProperty("os.name").toLowerCase();
if (!osName.contains("win")) {
pid = getProcessId(); // Use reflection to get PID on non-Windows
} else {
System.out.println("CPU affinity setting via JNA is not recommended on Windows. Use Windows API instead.");
return;
}
int cpu = 0; // Bind to CPU core 0
if (setAffinity(pid, cpu)) {
System.out.println("Successfully set affinity to CPU core " + cpu + " for PID " + pid);
} else {
System.err.println("Failed to set affinity to CPU core " + cpu + " for PID " + pid);
}
CpuSet currentAffinity = getAffinity(pid);
if (currentAffinity != null) {
System.out.println("Current CPU affinity: " + currentAffinity);
} else {
System.err.println("Failed to get current CPU affinity for PID " + pid);
}
}
// Get PID using reflection (for non-Windows)
private static int getProcessId() {
try {
java.lang.management.RuntimeMXBean runtime = java.lang.management.ManagementFactory.getRuntimeMXBean();
java.lang.reflect.Field jvm = runtime.getClass().getDeclaredField("jvm");
jvm.setAccessible(true);
sun.management.VMManagement mgmt = (sun.management.VMManagement) jvm.get(runtime);
java.lang.reflect.Method pid_method = mgmt.getClass().getDeclaredMethod("getProcessId");
pid_method.setAccessible(true);
return (Integer) pid_method.invoke(mgmt);
} catch (Exception e) {
return -1;
}
}
}注意事项：
- 需要添加JNA的依赖。
- 这段代码在Windows下不推荐使用，因为Windows有自己的API来处理CPU亲和性。在Windows下使用JNA可能导致不可预测的结果。
- 在非Windows平台上，它使用反射来获取进程ID。
这段代码首先定义了一个Libc接口，它继承自JNA的Library接口，并声明了sched_setaffinity和sched_getaffinity两个函数，这两个函数是Linux系统调用，用于设置和获取线程的CPU亲和性。
然后，定义了一个CpuSet类，用于表示CPU集合。cpu_mask是一个long数组，用于存储CPU掩码。每个long值代表64个CPU核心，如果某个CPU核心被设置，则对应的bit位为1。
setAffinity函数用于设置进程的CPU亲和性。它首先创建一个CpuSet对象，然后将指定的CPU核心设置为1，最后调用sched_setaffinity函数设置进程的CPU亲和性。
getAffinity函数用于获取进程的CPU亲和性。它首先创建一个CpuSet对象，然后调用sched_getaffinity函数获取进程的CPU亲和性，最后返回CpuSet对象。
在main函数中，首先获取当前进程的ID，然后将进程绑定到CPU核心0。最后，获取进程的CPU亲和性并打印出来。
-
Cgroup (Control Groups)：
Cgroups是Linux内核提供的一种机制，用于限制、控制和隔离进程组（process groups）的资源使用（如CPU、内存、磁盘I/O等）。 虽然cgroups主要用于资源限制，但它也可以用于设置CPU亲和性。-
创建Cgroup:
sudo mkdir /sys/fs/cgroup/cpuset/mygroup -
配置允许使用的CPU核心:
sudo sh -c "echo 0-3 > /sys/fs/cgroup/cpuset/mygroup/cpuset.cpus" sudo sh -c "echo 0 > /sys/fs/cgroup/cpuset/mygroup/cpuset.mems"上面的命令将cgroup
mygroup限制为只能使用CPU核心0到3，并且使用内存节点0。 -
将进程添加到Cgroup:
找到进程的PID，然后将其写入cgroup的tasks文件：echo <PID> | sudo tee /sys/fs/cgroup/cpuset/mygroup/tasks -
运行程序:
你可以使用cgexec命令来在cgroup中运行程序：cgexec -g cpuset:mygroup java YourJavaProgram
-
-
使用
CompletableFuture和自定义线程池Java的
CompletableFuture提供了一种异步编程的方式，结合自定义线程池，可以更灵活地控制线程的执行环境。 我们可以创建具有特定CPU亲和性的线程池，并将CompletableFuture的任务提交到这些线程池中。import java.util.concurrent.*; import java.util.concurrent.atomic.AtomicInteger; public class AffinityThreadPool { public static ExecutorService createAffinityThreadPool(int threads, final int[] cpus) { return Executors.newFixedThreadPool(threads, new ThreadFactory() { private final AtomicInteger threadNumber = new AtomicInteger(1); @Override public Thread newThread(Runnable r) { Thread thread = new Thread(r, "AffinityThread-" + threadNumber.getAndIncrement()); if (!Thread.currentThread().isDaemon()) thread.setDaemon(false); // 设置为非守护线程 else thread.setDaemon(true); int cpu = cpus[(threadNumber.get() - 2) % cpus.length]; // 轮询CPU setAffinity(getPid(), cpu); //设置线程亲和性 return thread; } }); } // ... (getProcessId() and setAffinity() methods from the previous example) public static void main(String[] args) throws Exception { int[] cpus = {0, 1, 2, 3}; // 指定允许使用的CPU核心 ExecutorService executor = createAffinityThreadPool(4, cpus); for (int i = 0; i < 10; i++) { final int taskNumber = i; CompletableFuture.supplyAsync(() -> { System.out.println("Task " + taskNumber + " running on thread: " + Thread.currentThread().getName()); try { Thread.sleep(100); // 模拟耗时操作 } catch (InterruptedException e) { Thread.currentThread().interrupt(); } return "Result from task " + taskNumber; }, executor).thenAccept(result -> System.out.println(result + " processed by thread: " + Thread.currentThread().getName())); } executor.shutdown(); executor.awaitTermination(1, TimeUnit.MINUTES); } }此代码创建了一个自定义线程池，其中每个线程都绑定到指定的CPU核心。
CompletableFuture用于异步执行任务，并将任务提交到该线程池。
优化策略
-
识别关键线程： 确定程序中对性能影响最大的线程，例如处理高并发请求的线程或执行计算密集型任务的线程。
-
绑定关键线程到特定的CPU核心： 将这些关键线程绑定到特定的CPU核心上，可以提高缓存命中率，减少线程抖动。
-
NUMA架构优化： 在NUMA架构下，尽量将线程绑定到与其本地内存相关的CPU核心上，减少跨NUMA节点的内存访问。可以使用
numactl命令或相应的API来设置NUMA亲和性。 -
避免过度绑定： 不要将所有线程都绑定到CPU核心上，特别是对于I/O密集型线程。过度绑定可能会导致CPU资源浪费，降低程序的整体性能。
-
动态调整： 根据程序的运行情况，动态调整线程的CPU亲和性。例如，可以根据CPU的负载情况，将线程迁移到空闲的CPU核心上。
-
使用线程池： 使用线程池可以有效地管理线程，避免频繁创建和销毁线程的开销。在创建线程池时，可以设置线程的CPU亲和性。
-
避免竞争： 尽量避免线程之间的竞争，例如使用锁或原子变量。竞争会导致线程阻塞，增加上下文切换的开销。
代码示例：使用ThreadLocalRandom代替Random
java.util.Random类是线程安全的，但在多线程环境下，多个线程同时访问同一个Random实例会导致竞争，降低性能。可以使用java.util.concurrent.ThreadLocalRandom类代替Random类，ThreadLocalRandom是线程本地的随机数生成器，可以避免线程之间的竞争。
import java.util.concurrent.ThreadLocalRandom;
public class RandomExample {
public static void main(String[] args) {
// 使用ThreadLocalRandom生成随机数
int randomNumber = ThreadLocalRandom.current().nextInt(100);
System.out.println("Random number: " + randomNumber);
}
}代码示例：使用Disruptor
Disruptor是一个高性能的并发框架，可以用于构建高性能的队列。Disruptor使用RingBuffer作为数据结构，避免了锁的使用，从而提高了并发性能。
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.util.DaemonThreadFactory;
public class DisruptorExample {
public static class LongEvent {
private long value;
public void set(long value) {
this.value = value;
}
public long get() {
return value;
}
}
public static class LongEventFactory implements com.lmax.disruptor.EventFactory<LongEvent> {
@Override
public LongEvent newInstance() {
return new LongEvent();
}
}
public static class LongEventHandler implements com.lmax.disruptor.EventHandler<LongEvent> {
@Override
public void onEvent(LongEvent event, long sequence, boolean endOfBatch) {
System.out.println("Event: " + event.get());
}
}
public static void main(String[] args) throws Exception {
int bufferSize = 1024;
Disruptor<LongEvent> disruptor = new Disruptor<>(
new LongEventFactory(),
bufferSize,
DaemonThreadFactory.INSTANCE);
disruptor.handleEventsWith(new LongEventHandler());
disruptor.start();
RingBuffer<LongEvent> ringBuffer = disruptor.getRingBuffer();
for (long i = 0; i < 10; i++) {
long sequence = ringBuffer.next();
try {
LongEvent event = ringBuffer.get(sequence);
event.set(i);
} finally {
ringBuffer.publish(sequence);
}
}
Thread.sleep(1000);
disruptor.shutdown();
}
}代码示例：使用ForkJoinPool
ForkJoinPool是Java 7引入的一个用于并行执行任务的线程池。它可以将一个大任务分解成多个小任务，然后并行执行这些小任务，最后将结果合并。ForkJoinPool使用工作窃取算法，可以有效地利用CPU资源。
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveTask;
public class ForkJoinExample {
public static class SumTask extends RecursiveTask<Long> {
private static final int THRESHOLD = 100;
private final long start;
private final long end;
public SumTask(long start, long end) {
this.start = start;
this.end = end;
}
@Override
protected Long compute() {
if (end - start <= THRESHOLD) {
long sum = 0;
for (long i = start; i <= end; i++) {
sum += i;
}
return sum;
} else {
long middle = (start + end) / 2;
SumTask leftTask = new SumTask(start, middle);
SumTask rightTask = new SumTask(middle + 1, end);
leftTask.fork();
rightTask.fork();
return leftTask.join() + rightTask.join();
}
}
}
public static void main(String[] args) {
ForkJoinPool pool = new ForkJoinPool();
SumTask task = new SumTask(1, 1000);
long result = pool.invoke(task);
System.out.println("Sum: " + result);
}
}一些具体的案例分析：
以下是一些实际案例，说明了如何通过设置CPU亲和性来优化JAVA并发程序的性能：
| 场景 | 问题描述 | 优化方案 | 效果 |
|---|---|---|---|
| 高并发Web服务器 | Web服务器在处理大量并发请求时，线程在不同的CPU核心之间频繁切换，导致CPU缓存失效，增加了延迟。 | 将处理请求的关键线程绑定到特定的CPU核心上。 | 降低了请求延迟，提高了吞吐量。 |
| 大数据处理 | 大数据处理程序需要处理大量的数据，线程在不同的CPU核心之间频繁切换，导致CPU缓存失效，降低了处理速度。 | 将处理数据的线程绑定到与其本地内存相关的CPU核心上，减少跨NUMA节点的内存访问。 | 提高了数据处理速度，降低了延迟。 |
| 实时音视频处理 | 实时音视频处理程序对延迟要求非常高，线程在不同的CPU核心之间频繁切换，导致延迟增加，影响用户体验。 | 将处理音视频数据的线程绑定到特定的CPU核心上，避免线程抖动。 | 降低了延迟，提高了用户体验。 |
| 游戏服务器 | 游戏服务器需要处理大量的游戏逻辑，线程在不同的CPU核心之间频繁切换，导致游戏卡顿。 | 将处理游戏逻辑的线程绑定到特定的CPU核心上，避免线程抖动。 | 减少了游戏卡顿，提高了游戏体验。 |
| 使用消息队列（如Kafka）的消费者 | Kafka消费者在处理消息时，线程可能在不同的CPU核心之间切换，导致消息处理延迟增加。 | 将Kafka消费者的线程绑定到特定的CPU核心上。 | 降低了消息处理延迟，提高了吞吐量。 |
| 执行机器学习模型的推理服务 | 机器学习模型的推理服务通常需要大量的计算资源，并且对延迟敏感。线程在不同的CPU核心之间切换会导致性能下降。 | 将执行推理任务的线程绑定到特定的CPU核心上，并确保这些核心能够充分利用CPU的特性（如AVX指令集）。 | 显著降低了推理延迟，提高了服务性能。 |
总结一些经验
通过以上讨论，我们可以得出以下结论：
- 线程抖动是影响并发程序性能的重要因素。
- CPU亲和性是一种有效的避免线程抖动的技术。
- 设置CPU亲和性需要根据具体的应用场景进行调整。
- 除了CPU亲和性，还可以使用其他技术来优化并发程序的性能，例如使用线程池、避免竞争等。
采取行动，优化你的代码
希望今天的分享能够帮助大家更好地理解JAVA并发中CPU亲和性的重要性，并能够在实际开发中运用这些技术来优化程序的性能。 记住，性能优化是一个持续的过程，需要不断地学习和实践。 谢谢大家！