首页 文章 精选 留言 我的

精选列表

搜索[智能解析],共10000篇文章
优秀的个人博客,低调大师

Java 虚拟线程探究与性能解析

虚拟线程是由Java运行时而不是操作系统实现的Java线程,和传统线程(或称之为平台线程)之间的主要区别在于,我们可以很容易地在同一个Java进程中运行大量活动的虚拟线程,甚至数百万个。大量的虚拟线程赋予了它们强大的功能:通过允许服务器并发处理更多的请求,它们可以更有效地运行以thread-per-request(每个请求一个线程)的方式编写的服务器应用程序,从而实现更高的吞吐量和更少的硬件浪费。 一直听闻Java虚拟线程的“威名”很久了,于是最近做个人项目的时候便尝试使用JDK21进行开发,研究一下所谓的虚拟线程的原理与实现。技术水平有限,欢迎一起交流探讨~ 背景 引入虚拟线程是为了减少编写、维护和观察高吞吐量并发应用程序的工作量。 对于应用提供的接口,其响应时间一定,那么此时其吞吐量与应用程序能够同时处理的请求数量(即并发数量)成正比。假设一个接口的响应耗时为50ms,而应用程序可以同时并发处理10个请求,那么每秒就有200(1s/50ms*10)个请求的吞吐量。此时如果应用程序可以将并发处理请求的能力提升到100,那么每秒则能达到2000的吞吐量。显然提高并发处理的线程数可以显著提高应用的吞吐量,然而Java中平台线程是昂贵的资源,默认每个平台线程消耗1MB栈内存,即 JVM 中运行的平台线程数量有上限。此外操作系统对于支持的最大线程也是有限制的,并不能无限制的增加内核线程的数量。下图为系统支持的最大线程数: 在大多数JVM的实现中,Java线程是和操作系统线程是一对一映射的(如下图),如果我们使用thread-per-request的形式(常见的如Tomcat、Jetty都是这样的模型),即为每个请求创建一个线程进行处理,那么很快便会到达操作系统线程数上限。 如果请求是IO密集型,那么大多线程都是处于阻塞等待IO返回的情况,会出现线程资源已经耗尽,而CPU利用率却很低。因此,若一个平台线程专用于用户请求,对高并发用户的应用程序,就非常容易出现线程池打满,后续请求进入阻塞的情况。 一些希望充分利用硬件的开发人员放弃了thread-per-request的形式,转而采用响应式编程。即请求处理代码不是从头到尾都在一个线程上进行,而是在等待 I/O 操作完成时将其线程返回到池中,以便线程可以为其他请求提供服务。这种细粒度的线程共享(在这种共享中,代码仅在线程执行计算时保留在线程上,而不是在等待 I/O 时保留线程)允许大量并发操作,而不会长时间占用线程。 然而这种方式虽然消除了操作系统线程稀缺性对吞吐量的限制,但它显著提高程序的理解成本和调试成本。它采用一组单独的 I/O 方法,这些方法不等待 I/O 操作完成,而是稍后向回调发出完成信号。开发人员必须将其请求处理逻辑分解为小阶段,然后将它们组合到一个顺序管道中。在响应式编程中,请求的每个阶段都可能在不同的线程上执行,并且每个线程都以交错方式运行属于不同请求的阶段,这种方式非常的复杂,创建响应式通道、调试以及理解它们的执行流程都非常的困难,更别说遇到异常时的排查。 虚拟线程的引入解决了上述的问题。Java 运行时以一种切断 Java 线程与操作系统线程的一对一对应关系的方式来实现 Java 线程,即虚拟线程。正如操作系统通过将大型虚拟地址空间映射到有限数量的物理 RAM 来给人一种内存充足的错觉一样,Java 运行时可以通过将大量虚拟线程映射到少量的操作系统线程来给人一种线程充足的错觉。 平台线程(java.lang.Thread) 是以传统方式实现的实例,作为操作系统线程的薄包装器,与系统线程一一映射,而虚拟线程不绑定到特定操作系统线程的实例。“thread-per-request”样式的应用程序代码可以在请求的整个持续时间内在虚拟线程中运行,但虚拟线程仅在 CPU 上执行计算时使用操作系统线程。虚拟线程具备与异步样式相同的可伸缩性,只是它的实现是透明的,不需要我们额外的理解和开发成本。当在虚拟线程中运行的代码进行阻塞的 I/O 操作时,运行时将自动挂起虚拟线程,直到以后可以恢复为止。对于 Java 开发人员来说,虚拟线程只是创建成本低廉且几乎无限丰富的线程。硬件利用率接近最佳状态,允许高并发性,从而实现高吞吐量,同时应用程序与 Java 平台及其工具的多线程设计保持和谐。 实现 ▐定义 与平台线程一样,虚拟线程也 是 java.lang.Thread 的 一个实例,但是虚拟线程不绑定到特定的操作系统线程。虚拟线程仍在操作系统线程上运行代码,区别是当虚拟线程上运行的代码调用阻塞 I/O 操作时,Java 运行时会将其挂起直到它可以恢复,与挂起的虚拟线程关联的操作系统线程此时可以自由地对其他虚拟线程执行操作。 虚拟线程的实现方式与虚拟内存类似。为了模拟大量内存,操作系统将一个大的虚拟地址空间映射到有限的 RAM。同样,为了模拟大量线程,Java 运行时将大量的虚拟线程映射到少量的操作系统线程。 与平台线程不同,虚拟线程通常具有较浅的调用堆栈,通常只执行一次 HTTP 客户端调用或一次 JDBC 查询。尽管虚拟线程支持线程局部变量(ThreadLocal)和可继承的线程局部变量(InheritedThreadLocal),我们应该仔细考虑使用它们,因为单个 JVM 可能支持数百万个虚拟线程。 虚拟线程适用于运行那些大部分时间是被阻塞着的任务,即 IO 密集型操作,而不适用于长时间运行的 CPU 密集型操作。因为虚拟线程不是更快的线程,它们运行代码的速度并不比平台线程快。它们的存在是为了提供可伸缩性(更高的吞吐量),而不是速度(更低的延迟)。 ▐原理 final class VirtualThread extends BaseVirtualThread { private static final ForkJoinPool DEFAULT_SCHEDULER = createDefaultScheduler(); private final Executor scheduler; private final Continuation cont; private final Runnable runContinuation; private volatile Thread carrierThread; VirtualThread(Executor scheduler, String name, int characteristics, Runnable task) { super(name, characteristics, /*bound*/ false); Objects.requireNonNull(task); // choose scheduler if not specified if (scheduler == null) { Thread parent = Thread.currentThread(); if (parent instanceof VirtualThread vparent) { scheduler = vparent.scheduler; } else { scheduler = DEFAULT_SCHEDULER; } } this.scheduler = scheduler; this.cont = new VThreadContinuation(this, task); this.runContinuation = this::runContinuation; } private static ForkJoinPool createDefaultScheduler() { ForkJoinWorkerThreadFactory factory = pool -> { PrivilegedAction<ForkJoinWorkerThread> pa = () -> new CarrierThread(pool); return AccessController.doPrivileged(pa); }; PrivilegedAction<ForkJoinPool> pa = () -> { int parallelism, maxPoolSize, minRunnable; String parallelismValue = System.getProperty("jdk.virtualThreadScheduler.parallelism"); String maxPoolSizeValue = System.getProperty("jdk.virtualThreadScheduler.maxPoolSize"); String minRunnableValue = System.getProperty("jdk.virtualThreadScheduler.minRunnable"); if (parallelismValue != null) { parallelism = Integer.parseInt(parallelismValue); } else { parallelism = Runtime.getRuntime().availableProcessors(); } if (maxPoolSizeValue != null) { maxPoolSize = Integer.parseInt(maxPoolSizeValue); parallelism = Integer.min(parallelism, maxPoolSize); } else { maxPoolSize = Integer.max(parallelism, 256); } if (minRunnableValue != null) { minRunnable = Integer.parseInt(minRunnableValue); } else { minRunnable = Integer.max(parallelism / 2, 1); } Thread.UncaughtExceptionHandler handler = (t, e) -> { }; boolean asyncMode = true; // FIFO return new ForkJoinPool(parallelism, factory, handler, asyncMode, 0, maxPoolSize, minRunnable, pool -> true, 30, SECONDS); }; return AccessController.doPrivileged(pa); } private void runContinuation() { // the carrier must be a platform thread if (Thread.currentThread().isVirtual()) { throw new WrongThreadException(); } // set state to RUNNING int initialState = state(); if (initialState == STARTED || initialState == UNPARKED || initialState == YIELDED) { // newly started or continue after parking/blocking/Thread.yield if (!compareAndSetState(initialState, RUNNING)) { return; } // consume parking permit when continuing after parking if (initialState == UNPARKED) { setParkPermit(false); } } else { // not runnable return; } mount(); try { cont.run(); } finally { unmount(); if (cont.isDone()) { afterDone(); } else { afterYield(); } } }} 虚拟线程有几个核心的对象: Continuation:译为“续延”,是用户真实任务的包装器,虚拟线程会把任务包装到一个Continuation实例中,当任务需要阻塞挂起的时候,会调用Continuation的yield操作进行阻塞 Scheduler:译为“调度器”,会把任务提交到一个平台线程池中执行,虚拟线程中维护了一个默认的调度器DEFAULT_SCHEDULER,这是一个 ForkJoinPool 实例,最大线程数默认是系统核心线程数,最大为 256,可以通过 jdk.virtualThreadScheduler.maxPoolSize 进行设置。 carrier:载体线程(Thread对象),指的是负责执行虚拟线程中任务的平台线程。 runContinuation:一个Runnable对象,用于在任务运行或继续之前,虚拟线程将装载到当前线程上。当任务完成或完成时,将其卸载。 具体虚拟线程的工作流程后续可能会再深入源码进行分析。 使用 ▐使用Thread类创建虚拟线程 // name(String prefix, Integer start) p0:前缀 p1:计数器初始值Thread.Builder.OfVirtual virtualThreadBuilder = Thread.ofVirtual().name("worker-", 0);Thread worker0 = virtualThreadBuilder.start(this::doSomethings);worker0.join();System.out.print("finish worker-0 running");Thread worker1 = virtualThreadBuilder.start(this::doSomethings);worker1.join();System.out.print("finish worker-1 running"); 调用 Thread.ofVirtual() 方法会创建一个用于创建虚拟线程的 Thread.Builder 实例。 ▐使用Executors创建虚拟线程 try (ExecutorService executorService = Executors.newVirtualThreadPerTaskExecutor()) { Future<?> submit = executorService.submit(this::doSomethings); submit.get(); System.out.print("finish running");} 虚拟线程既便宜又丰富,因此永远不应该被池化,应该为每个应用程序任务创建一个新的虚拟线程。使用 newVirtualThreadPerTaskExecutor 创建的是一个没有线程数量限制的线程池(并不是一个典型的线程池,并不是为了复用线程而存在),其会为每个提交的任务创建一个新的虚拟线程进行处理。 ▐使用虚拟线程实现服务端 public class Server { public static void main(String[] args) { Set<String> platformSet = new HashSet<>(); new Thread(() -> { try { Thread.sleep(10000); System.out.println(platformSet.size()); } catch (InterruptedException e) { throw new RuntimeException(e); } }).start(); try (ServerSocket serverSocket = new ServerSocket(9999)) { Thread.Builder.OfVirtual clientThreadBuilder = Thread.ofVirtual().name("client", 1); while (true) { Socket clientSocket = serverSocket.accept(); clientThreadBuilder.start(() -> { String platformName = Thread.currentThread().toString().split("@")[1]; platformSet.add(platformName); try ( BufferedReader in = new BufferedReader(new InputStreamReader(clientSocket.getInputStream())); PrintWriter out = new PrintWriter(clientSocket.getOutputStream(), true); ) { String inputLine; while ((inputLine = in.readLine()) != null) { System.out.println(inputLine + "(from:" + Thread.currentThread() + ")"); out.println(inputLine); } } catch (IOException e) { System.err.println(e.getMessage()); } }); } } catch (IOException e) { System.err.println("Exception caught when trying to listen on port 999"); System.err.printf(e.getMessage()); } }} 监听客户端连接,每次有客户端连接则创建一个虚拟线程进行处理,并在虚拟线程运行时将其平台线程的名称加入到Set中,另外有一个线程睡眠10秒后打印出Set的大小,则可以看出这些虚拟线程实际上用了多少个平台线程。 public class Client { public static void main(String[] args) throws InterruptedException { Thread.Builder.OfVirtual builder = Thread.ofVirtual().name("client", 1); for (int i = 0; i < 100000; i++) { builder.start(() -> { try ( Socket serverSocket = new Socket("localhost", 9999); BufferedReader in = new BufferedReader(new InputStreamReader(serverSocket.getInputStream())); PrintWriter out = new PrintWriter(serverSocket.getOutputStream(), true); ) { out.println("hello"); String inputLine; while ((inputLine = in.readLine()) != null) { System.out.println(inputLine); } } catch (UnknownHostException e) { System.err.println("Don't know about localhost"); } catch (IOException e) { System.err.println("Couldn't get I/O for the connection to localhost"); } }); } Thread.sleep(1000000000); }} 创建10w个客户端连接服务端并发送消息,主线程长时间睡眠避免程序直接结束。 服务端最终使用了19个平台线程(与CPU核心线程数有关)用于处理10w个客户端连接。 ▐虚拟线程调度和固定 当平台线程运行时,由操作系统进行调度,而当虚拟线程运行时,由 Java 运行时进行调度。当 Java 运行时调度一个虚拟线程时,会将这个虚拟线程挂载在一个平台线程上,之后同样由操作系统内核进行调度。这里被挂载的平台线程被称为 carrier(搬运工)。当虚拟线程被阻塞时,会从 carrier 上取消挂载,此时 carrier 是空闲的,Java 运行时可以调度其他虚拟线程挂载在其上。这个过程对于内核线程是无感知的,可以避免使用平台线程遇到阻塞时出现内核态与用户态切换带来的开销,并且充分利用 CPU 计算性能,提高应用程序的吞吐量。 当虚拟线程被固定(pinned)在 carrier 上时,即使遇到阻塞也不会取消挂载。在以下场景虚拟线程会被固定: 1. 虚拟线程执行的方法或块被 synchronized 关键字标识时 2. 虚拟线程运行外部函数时 固定不会使应用程序出错,但可能会阻碍其可伸缩性。可以尝试通过使用java.util.concurrent.locks.ReentrantLock.synchronized 来修改频繁运行的块和方法以及保护可能长时间的I/O操作,以避免频繁和长时间的固定。 注意事项 由于虚拟线程是Java.lang.Thread的实现,并且遵循自Java SE 1.0以来指定Java.lang.Thread的相同规则,因此开发人员不需要学习使用它们的新概念。然而,由于无法生成非常多的平台线程(多年来Java中唯一可用的线程实现),因此产生了旨在应对其高成本的实践。当应用于虚拟线程时,这些做法会适得其反,必须摒弃。 ▐写简单的同步代码,使用阻塞的API 虚拟线程可以显著提高以thread-per-request的方式编写的服务器的吞吐量(而不是延迟)。在这种风格中,服务器在整个持续时间内专用一个线程来处理每个传入请求。 阻塞平台线程的代价很高,因为它占用了系统线程(相对稀缺的资源),而并没有做多少有意义的工作,因而在过去我们可能会使用异步非阻塞的方式来实现一些功能,然而虚拟线程可以有很多,所以阻塞它们的成本很低,因此我们应该以直接的同步风格编写代码,并使用阻塞I/O api。 例如,下面以非阻塞、异步风格编写的代码不会从虚拟线程中获得太多好处。 CompletableFuture.supplyAsync(info::getUrl, pool) .thenCompose(url -> getBodyAsync(url, HttpResponse.BodyHandlers.ofString())) .thenApply(info::findImage) .thenCompose(url -> getBodyAsync(url, HttpResponse.BodyHandlers.ofByteArray())) .thenApply(info::setImageData) .thenAccept(this::process) .exceptionally(t -> { t.printStackTrace(); return null; }); 下面以同步风格编写并使用简单阻塞IO的代码将受益匪浅: try { String page = getBody(info.getUrl(), HttpResponse.BodyHandlers.ofString()); String imageUrl = info.findImage(page); byte[] data = getBody(imageUrl, HttpResponse.BodyHandlers.ofByteArray()); info.setImageData(data); process(info);} catch (Exception ex) { t.printStackTrace();} 这样的代码也更容易在调试器中进行调试,在分析器中进行概要分析,或者使用线程转储进行观察。以这种风格编写的堆栈越多,虚拟线程的性能和可观察性就越好。用其他风格编写的程序或框架,如果没有为每个任务指定一个线程,就不应该期望从虚拟线程中获得显著的好处。避免将同步、阻塞代码与异步框架混在一起。 ▐不要共用虚拟线程 虚拟线程虽然具有与平台线程相同的行为,但它们不应该表示相同的程序概念。平台线程是稀缺的,因此是一种宝贵的资源。需要管理宝贵的资源,管理平台线程的最常用方法是使用线程池。接下来需要回答的问题是,池中应该有多少线程? 但是虚拟线程非常多,因此每个线程不应该代表一些共享的、池化的资源,而应该代表一个任务,线程从托管资源转变为应用程序域对象。我们应该有多少个虚拟线程的问题变得很明显,就像我们应该使用多少个字符串在内存中存储一组用户名的问题一样:虚拟线程的数量总是等于应用程序中并发任务的数量。 为了将每个应用程序任务表示为一个线程,不要像下面的例子那样使用共享线程池执行器: Future<ResultA> f1 = sharedThreadPoolExecutor.submit(task1);Future<ResultB> f2 = sharedThreadPoolExecutor.submit(task2);// ... use futures 应该采用以下的方式: try (var executor = Executors.newVirtualThreadPerTaskExecutor()) { Future<ResultA> f1 = executor.submit(task1); Future<ResultB> f2 = executor.submit(task2); // ... use futures} 代码仍然使用ExecutorService,但是 Executors.newVirtualThreadPerTaskExecutor()返回的实例并不会复用虚拟线程。相反,它为每个提交的任务创建一个新的虚拟线程。 void handle(Request request, Response response) { var url1 = ... var url2 = ... try (var executor = Executors.newVirtualThreadPerTaskExecutor()) { var future1 = executor.submit(() -> fetchURL(url1)); var future2 = executor.submit(() -> fetchURL(url2)); response.send(future1.get() + future2.get()); } catch (ExecutionException | InterruptedException e) { response.fail(e); }} String fetchURL(URL url) throws IOException { try (var in = url.openStream()) { return new String(in.readAllBytes(), StandardCharsets.UTF_8); }} 此外,ExecutorService本身是轻量级的,我们可以创建一个新的,就像处理任何简单的对象一样。并不用将这个对象保存起来每次使用相同的这个实例,而是在需要的时候创建一个就行了。 您应该创建一个新的虚拟线程,如上所示,即使是小型的、短暂的并发任务也是如此。根据经验,如果应用程序从来没有10,000个或更多的虚拟线程,那么它不太可能从虚拟线程中获益。要么它的负载太轻,不需要更好的吞吐量,要么没有向虚拟线程表示足够多的任务。 ▐使用信号量限制并发 有时需要限制某个操作的并发性。例如,某些外部服务可能无法处理十个以上的并发请求。使用平台线程时可以用线程池的大小来限制并发的数量,而在使用虚拟线程时,如果希望限制访问某些服务的并发性,则应该使用专门为此目的设计的Semaphore类。下面的例子演示了这个类: Semaphore sem = new Semaphore(10);// ......Executors.newVirtualThreadPerTaskExecutor().submit(() -> { try { // 执行任务前信号量-1,表示多了一个并发线程在执行了,剩下的可同时执行数量减少 // 如果信号量(许可证)为0,则阻塞直到有其他线程执行完成释放许可证 sem.acquire(); doSomething(); } catch (InterruptedException e) { throw new RuntimeException(e); } finally { // 执行完任务后信号量+1 sem.release(); }}); ▐不要在线程局部变量中缓存昂贵的可重用对象 虚拟线程支持线程局部变量,就像平台线程一样。通常,线程局部变量用于将一些特定于上下文的信息与当前运行的代码相关联,例如当前事务和用户ID。对于虚拟线程,使用线程局部变量是完全合理的。而线程局部变量的另一个作用是缓存可重用对象,这些对象缓存在线程局部变量中,供不同时间运行在线程上的多个任务重用,目的是减少实例化的次数和内存中的实例数量。 这与虚拟线程的设计完全不符,只有当多个任务共享并重用线程(因此是缓存在线程本地的昂贵对象)时,这种缓存才有用,就像平台线程被池化时一样。在线程池中运行时,可能会调用许多任务,但由于线程池只包含几个线程,因此对象只会被实例化几次(每个线程池一次),然后缓存并重用。但是,虚拟线程永远不会被池化,也不会被不相关的任务重用。因为每个任务都有自己的虚拟线程,所以来自不同任务的每次调用都会触发这个缓存变量的实例化。此外,由于可能有大量的虚拟线程并发地运行,昂贵的对象可能会消耗相当多的内存。这些结果与线程局部缓存想要实现的目标完全相反。 ▐避免长时间和频繁的固定 当前虚拟线程实现的一个限制是,在synchronized块或方法内部执行阻塞操作会导致JDK的虚拟线程调度器阻塞宝贵的操作系统线程,而在块或方法外部执行阻塞操作则不会,我们称这种情况为pinning。如果阻塞操作既长又频繁,那么固定可能会对服务器的吞吐量产生不利影响。保护短期操作,比如内存操作,或者不经常使用synchronized块或方法的操作。 如果固定存在时间较长且频繁的地方,那么在这些特定的地方用ReentrantLock代替synchronized(不需要在synchronized保护时间较短或不频繁的操作的地方替换synchronized)。下面是一个长时间且频繁使用同步块的例子。 synchronized(lockObj) { frequentIO();} 采用以下的实现替换: lock.lock();try { frequentIO();} finally { lock.unlock();} 性能测试 JDK:OpenJDK21.0.4 物理机:Win11 & i5-14600KF(14核20线程) ▐平台线程与虚拟线程简单对比示例 public class PerformanceTest { private static final int REQUEST_NUM = 10000; public static void main(String[] args) { long vir = 0, p1 = 0, p2 = 0, p3 = 0, p4 = 0; for (int i = 0; i < 3; i++) { vir += testVirtualThread(); p1 += testPlatformThread(200); p2 += testPlatformThread(500); p3 += testPlatformThread(800); p4 += testPlatformThread(1000); System.out.println("--------------"); } System.out.println("虚拟线程平均耗时:" + vir / 3 + "ms"); System.out.println("平台线程[200]平均耗时:" + p1 / 3 + "ms"); System.out.println("平台线程[500]平均耗时:" + p2 / 3 + "ms"); System.out.println("平台线程[800]平均耗时:" + p3 / 3 + "ms"); System.out.println("平台线程[1000]平均耗时:" + p4 / 3 + "ms"); } private static long testVirtualThread() { long startTime = System.currentTimeMillis(); ExecutorService executorService = Executors.newVirtualThreadPerTaskExecutor(); for (int i = 0; i < REQUEST_NUM; i++) { executorService.submit(PerformanceTest::handleRequest); } executorService.close(); long useTime = System.currentTimeMillis() - startTime; System.out.println("虚拟线程耗时:" + useTime + "ms"); return useTime; } private static long testPlatformThread(int poolSize) { long startTime = System.currentTimeMillis(); ExecutorService executorService = Executors.newFixedThreadPool(poolSize); for (int i = 0; i < REQUEST_NUM; i++) { executorService.submit(PerformanceTest::handleRequest); } executorService.close(); long useTime = System.currentTimeMillis() - startTime; System.out.printf("平台线程[%d]耗时:%dms\n", poolSize, useTime); return useTime; } private static void handleRequest() { try { Thread.sleep(300); } catch (InterruptedException e) { throw new RuntimeException(e); } }} 测试结果: 虚拟线程耗时:654ms平台线程[200]耗时:15551ms平台线程[500]耗时:6241ms平台线程[800]耗时:4069ms平台线程[1000]耗时:3137ms--------------虚拟线程耗时:331ms平台线程[200]耗时:15544ms平台线程[500]耗时:6227ms平台线程[800]耗时:4047ms平台线程[1000]耗时:3126ms--------------虚拟线程耗时:326ms平台线程[200]耗时:15552ms平台线程[500]耗时:6228ms平台线程[800]耗时:4054ms平台线程[1000]耗时:3151ms--------------虚拟线程平均耗时:437ms平台线程[200]平均耗时:15549ms平台线程[500]平均耗时:6232ms平台线程[800]平均耗时:4056ms平台线程[1000]平均耗时:3138ms 由于虚拟线程可以无限制的创建,而平台线程受线程池大小约束,因而1万个请求并不能同时处理,后续的请求需要等待前面的请求处理完成释放线程后才能进行,所以明显耗时远高于使用虚拟线程。 ▐简单的Web服务测试 springboot-web版本(Tomcat/10.1.19):3.2.3 / springboot-webflux版本(Netty):3.2.3 编写简单的测试程序,使用Thread.sleep模拟300ms的阻塞,使用Jmeter模拟3000个用户的并发请求。 web版本程序: @RestControllerpublic class TestController { @GetMapping("get") public String get() { try {// System.out.println(Thread.currentThread()); Thread.sleep(300); } catch (InterruptedException e) { throw new RuntimeException(e); } return "ok"; }} 通过application.yaml配置文件控制线程数量和是否启用虚拟线程: server: tomcat: threads: max: 200spring: threads: virtual: enabled: false # 是否启用虚拟线程 webflux版本程序: @Configurationpublic class TestWebClient { @Bean public RouterFunction<ServerResponse> routes() { return route( GET("/get"), request -> ok() .contentType(MediaType.APPLICATION_JSON) .body(fromPublisher(Mono.just("ok").delayElement(Duration.ofMillis(300)), String.class)) ); }} 测试结果: 使用平台线程数 吞吐量(req/s) 平均响应时间(ms) 90% 95% 99% 虚拟线程 20 5217 316 311 344 354 平台线程 200 624.5 2660 4407 4782 4801 平台线程 512 1564.1 984 1683 1693 1787 平台线程 800 2340 661 1067 1070 1075 WebFlux 5281.3 310 314 321 325 可以看到使用虚拟线程和WebFlux响应式时的吞吐量远超使用普通线程池,且虚拟线程的吞吐量并不比WebFlux差,而使用虚拟线程不需要进行复杂的响应式编程,只需要配置启用虚拟线程即可实现高吞吐量。 结语 总之,Java虚拟线程的引入是对现代并发编程模型的一次革新,它不仅简化了并发编程的复杂度,还极大地提升了应用的并发处理能力和资源利用率,为构建高性能、可扩展的服务器端应用提供了新的思路和工具。随着技术的成熟和普及,虚拟线程有望成为未来Java并发编程的标准实践之一。 参考资料 Virtual Threads: https://docs.oracle.com/en/java/javase/21/core/virtual-threads.html#GUID-DC4306FC-D6C1-4BCC-AECE-48C32C1A8DAA JEP 444: Virtual Threads : https://openjdk.org/jeps/444#Thread-local-variables Spring Webflux : https://springdoc.cn/spring-webflux/ ¤ 拓展阅读 ¤ 3DXR技术| 终端技术| 音视频技术 服务端技术|技术质量|数据算法 本文分享自微信公众号 - 大淘宝技术(AlibabaMTT)。 如有侵权,请联系 support@oschina.cn 删除。 本文参与“OSC源创计划”,欢迎正在阅读的你也加入,一起分享。

优秀的个人博客,低调大师

并发编程-CompletableFuture解析 | 京东物流技术团队

1、CompletableFuture介绍 CompletableFuture对象是JDK1.8版本新引入的类,这个类实现了两个接口,一个是Future接口,一个是CompletionStage接口。 CompletionStage接口是JDK1.8版本提供的接口,用于异步执行中的阶段处理,CompletionStage定义了一组接口用于在一个阶段执行结束之后,要么继续执行下一个阶段,要么对结果进行转换产生新的结果等,一般来说要执行下一个阶段都需要上一个阶段正常完成,这个类也提供了对异常结果的处理接口 2、CompletableFuture的API 2.1 提交任务 在CompletableFuture中提交任务有以下几种方式: public static CompletableFuture<Void> runAsync(Runnable runnable) public static CompletableFuture<Void> runAsync(Runnable runnable, Executor executor) public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier) public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, Executor executor) 这四个方法都是用来提交任务的,不同的是supplyAsync提交的任务有返回值,runAsync提交的任务没有返回值。两个接口都有一个重载的方法,第二个入参为指定的线程池,如果不指定,则默认使用ForkJoinPool.commonPool()线程池。在使用的过程中尽量根据不同的业务来指定不同的线程池,方便对不同线程池进行监控,同时避免业务共用线程池相互影响。 2.2 结果转换 2.2.1 thenApply public <U> CompletableFuture<U> thenApply(Function<? super T,? extends U> fn) public <U> CompletableFuture<U> thenApplyAsync(Function<? super T,? extends U> fn) public <U> CompletableFuture<U> thenApplyAsync(Function<? super T,? extends U> fn, Executor executor) thenApply这一组函数入参是Function,意思是将上一个CompletableFuture执行结果作为入参,再次进行转换或者计算,重新返回一个新的值。 2.2.2 handle public <U> CompletableFuture<U> handle(BiFunction<? super T, Throwable, ? extends U> fn) public <U> CompletableFuture<U> handleAsync(BiFunction<? super T, Throwable, ? extends U> fn) public <U> CompletableFuture<U> handleAsync(BiFunction<? super T, Throwable, ? extends U> fn, Executor executor) handle这一组函数入参是BiFunction,该函数式接口有两个入参一个返回值,意思是处理上一个CompletableFuture的处理结果,同时如果有异常,需要手动处理异常。 2.2.3 thenRun public CompletableFuture<Void> thenRun(Runnable action) public CompletableFuture<Void> thenRunAsync(Runnable action) public CompletableFuture<Void> thenRunAsync(Runnable action, Executor executor) thenRun这一组函数入参是Runnable函数式接口,该接口无需入参和出参,这一组函数是在上一个CompletableFuture任务执行完成后,在执行另外一个接口,不需要上一个任务的结果,也不需要返回值,只需要在上一个任务执行完成后执行即可。 2.2.4 thenAccept public CompletableFuture<Void> thenAccept(Consumer<? super T> action) public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action) public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action, Executor executor) thenAccept这一组函数的入参是Consumer,该函数式接口有一个入参,没有返回值,所以这一组接口的意思是处理上一个CompletableFuture的处理结果,但是不返回结果。 2.2.5 thenAcceptBoth public <U> CompletableFuture<Void> thenAcceptBoth(CompletionStage<? extends U> other, BiConsumer<? super T, ? super U> action) public <U> CompletableFuture<Void> thenAcceptBothAsync(CompletionStage<? extends U> other, BiConsumer<? super T, ? super U> action) public <U> CompletableFuture<Void> thenAcceptBothAsync(CompletionStage<? extends U> other, BiConsumer<? super T, ? super U> action, Executor executor) thenAcceptBoth这一组函数入参包括CompletionStage以及BiConsumer,CompletionStage是JDK1.8新增的接口,在JDK中只有一个实现类:CompletableFuture,所以第一个入参就是CompletableFuture,这一组函数是用来接受两个CompletableFuture的返回值,并将其组合到一起。BiConsumer这个函数式接口有两个入参,并且没有返回值,BiConsumer的第一个入参就是调用方CompletableFuture的执行结果,第二个入参就是thenAcceptBoth接口入参的CompletableFuture的执行结果。所以这一组函数意思是将两个CompletableFuture执行结果合并到一起。 2.2.6 thenCombine public <U,V> CompletableFuture<V> thenCombine(CompletionStage<? extends U> other, BiFunction<? super T,? super U,? extends V> fn) public <U,V> CompletableFuture<V> thenCombineAsync(CompletionStage<? extends U> other, BiFunction<? super T,? super U,? extends V> fn) public <U,V> CompletableFuture<V> thenCombineAsync(CompletionStage<? extends U> other, BiFunction<? super T,? super U,? extends V> fn, Executor executor) thenCombine这一组函数和thenAcceptBoth类似,入参都包含一个CompletionStage,也就是CompletableFuture对象,意思也是组合两个CompletableFuture的执行结果,不同的是thenCombine的第二个入参为BiFunction,该函数式接口有两个入参,同时有一个返回值。所以与thenAcceptBoth不同的是,thenCombine将两个任务结果合并后会返回一个全新的值作为出参。 2.2.7 thenCompose public <U> CompletableFuture<U> thenCompose(Function<? super T, ? extends CompletionStage<U>> fn) public <U> CompletableFuture<U> thenComposeAsync(Function<? super T, ? extends CompletionStage<U>> fn) public <U> CompletableFuture<U> thenComposeAsync(Function<? super T, ? extends CompletionStage<U>> fn, Executor executor) thenCompose这一组函数意思是将调用方的执行结果作为Function函数的入参,同时返回一个新的CompletableFuture对象。 2.3 回调方法 public CompletableFuture<T> whenComplete(BiConsumer<? super T, ? super Throwable> action) public CompletableFuture<T> whenCompleteAsync(BiConsumer<? super T, ? super Throwable> action) public CompletableFuture<T> whenCompleteAsync(BiConsumer<? super T, ? super Throwable> action, Executor executor) whenComplete方法意思是当上一个CompletableFuture对象任务执行完成后执行该方法。BiConsumer函数式接口有两个入参没有返回值,这两个入参第一个是CompletableFuture任务的执行结果,第二个是异常信息。表示处理上一个任务的结果,如果有异常,则需要手动处理异常,与handle方法的区别在于,handle方法的BiFunction是有返回值的,而BiConsumer是没有返回值的。 以上方法都有一个带有Async的方法,带有Async的方法表示是异步执行的,会将该任务放到线程池中执行,同时该方法会有一个重载的方法,最后一个参数为Executor,表示异步执行可以指定线程池执行。为了方便进行控制,最好在使用CompletableFuture时手动指定我们的线程池。 2.4 异常处理 public CompletableFuture<T> exceptionally(Function<Throwable, ? extends T> fn) exceptionally是用来处理异常的,当任务抛出异常后,可以通过exceptionally来进行处理,也可以选择使用handle来进行处理,不过两者有些不同,hand是用来处理上一个任务的结果,如果有异常情况,就处理异常。而exceptionally可以放在CompletableFuture处理的最后,作为兜底逻辑来处理未知异常。 2.5 获取结果 public static CompletableFuture<Void> allOf(CompletableFuture<?>... cfs) public static CompletableFuture<Object> anyOf(CompletableFuture<?>... cfs) allOf是需要入参中所有的CompletableFuture任务执行完成,才会进行下一步; anyOf是入参中任何一个CompletableFuture任务执行完成都可以执行下一步。 public T get() throws InterruptedException, ExecutionException public T get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException public T getNow(T valueIfAbsent) public T join() get方法一个是不带超时时间的,一个是带有超时时间的。 getNow方法则是立即返回结果,如果还没有结果,则返回默认值,也就是该方法的入参。 join方法是不带超时时间的等待任务完成。 3、CompletableFuture原理 join方法同样表示获取结果,但是join与get方法有什么区别呢。 public T join() { Object r; return reportJoin((r = result) == null ? waitingGet(false) : r); } public T get() throws InterruptedException, ExecutionException { Object r; return reportGet((r = result) == null ? waitingGet(true) : r); } public T get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException { Object r; long nanos = unit.toNanos(timeout); return reportGet((r = result) == null ? timedGet(nanos) : r); } public T getNow(T valueIfAbsent) { Object r; return ((r = result) == null) ? valueIfAbsent : reportJoin(r); } 以上是CompletableFuture类中两个方法的代码,可以看到两个方法几乎一样。区别在于reportJoin/reportGet,waitingGet方法是一致的,只不过参数不一样,我们在看下reportGet与reportJoin方法。 private static <T> T reportGet(Object r) throws InterruptedException, ExecutionException { if (r == null) // by convention below, null means interrupted throw new InterruptedException(); if (r instanceof AltResult) { Throwable x, cause; if ((x = ((AltResult)r).ex) == null) return null; if (x instanceof CancellationException) throw (CancellationException)x; if ((x instanceof CompletionException) && (cause = x.getCause()) != null) x = cause; throw new ExecutionException(x); } @SuppressWarnings("unchecked") T t = (T) r; return t; } private static <T> T reportJoin(Object r) { if (r instanceof AltResult) { Throwable x; if ((x = ((AltResult)r).ex) == null) return null; if (x instanceof CancellationException) throw (CancellationException)x; if (x instanceof CompletionException) throw (CompletionException)x; throw new CompletionException(x); } @SuppressWarnings("unchecked") T t = (T) r; return t; } 可以看到这两个方法很相近,reportGet方法判断了r对象是否为空,并抛出了中断异常,而reportJoin方法没有判断,同时reportJoin抛出的都是运行时异常,所以join方法也是无需手动捕获异常的。 我们在看下waitingGet方法 private Object waitingGet(boolean interruptible) { Signaller q = null; boolean queued = false; int spins = -1; Object r; while ((r = result) == null) { if (spins < 0) spins = SPINS; else if (spins > 0) { if (ThreadLocalRandom.nextSecondarySeed() >= 0) --spins; } else if (q == null) q = new Signaller(interruptible, 0L, 0L); else if (!queued) queued = tryPushStack(q); else if (interruptible && q.interruptControl < 0) { q.thread = null; cleanStack(); return null; } else if (q.thread != null && result == null) { try { ForkJoinPool.managedBlock(q); } catch (InterruptedException ie) { q.interruptControl = -1; } } } if (q != null) { q.thread = null; if (q.interruptControl < 0) { if (interruptible) r = null; // report interruption else Thread.currentThread().interrupt(); } } postComplete(); return r; } 该waitingGet方法是通过while的方式循环判断是否任务已经完成并产生结果,如果结果为空,则会一直在这里循环,这里需要注意的是在这里初始化了一下spins=-1,当第一次进入while循环的时候,spins是-1,这时会将spins赋值为一个常量,该常量为SPINS。 private static final int SPINS = (Runtime.getRuntime().availableProcessors() > 1 ? 1 << 8 : 0); 这里判断可用CPU数是否大于1,如果大于1,则该常量为 1<< 8,也就是256,否则该常量为0。 第二次进入while循环的时候,spins是256大于0,这里做了减一的操作,下次进入while循环,如果还没有结果,依然是大于0继续做减一的操作,此处用来做短时间的自旋等待结果,只有当spins等于0,后续会进入正常流程判断。 我们在看下timedGet方法的源码 private Object timedGet(long nanos) throws TimeoutException { if (Thread.interrupted()) return null; if (nanos <= 0L) throw new TimeoutException(); long d = System.nanoTime() + nanos; Signaller q = new Signaller(true, nanos, d == 0L ? 1L : d); // avoid 0 boolean queued = false; Object r; // We intentionally don't spin here (as waitingGet does) because // the call to nanoTime() above acts much like a spin. while ((r = result) == null) { if (!queued) queued = tryPushStack(q); else if (q.interruptControl < 0 || q.nanos <= 0L) { q.thread = null; cleanStack(); if (q.interruptControl < 0) return null; throw new TimeoutException(); } else if (q.thread != null && result == null) { try { ForkJoinPool.managedBlock(q); } catch (InterruptedException ie) { q.interruptControl = -1; } } } if (q.interruptControl < 0) r = null; q.thread = null; postComplete(); return r; } timedGet方法依然是通过while循环的方式来判断是否已经完成,timedGet方法入参为一个纳秒值,并通过该值计算出一个deadline截止时间,当while循环还未获取到任务结果且已经达到截止时间,则抛出一个TimeoutException异常。 4、CompletableFuture实现多线程任务 这里我们通过CompletableFuture来实现一个多线程处理异步任务的例子。 这里我们创建10个任务提交到我们指定的线程池中执行,并等待这10个任务全部执行完毕。 每个任务的执行流程为第一次先执行加法,第二次执行乘法,如果发生异常则返回默认值,当10个任务执行完成后依次打印每个任务的结果。 public void demo() throws InterruptedException, ExecutionException, TimeoutException { // 1、自定义线程池 ExecutorService executorService = new ThreadPoolExecutor(5, 10, 60L, TimeUnit.SECONDS, new LinkedBlockingQueue<>(100)); // 2、集合保存future对象 List<CompletableFuture<Integer>> futures = new ArrayList<>(10); for (int i = 0; i < 10; i++) { int finalI = i; CompletableFuture<Integer> future = CompletableFuture // 提交任务到指定线程池 .supplyAsync(() -> this.addValue(finalI), executorService) // 第一个任务执行结果在此处进行处理 .thenApplyAsync(k -> this.plusValue(finalI, k), executorService) // 任务执行异常时处理异常并返回默认值 .exceptionally(e -> this.defaultValue(finalI, e)); // future对象添加到集合中 futures.add(future); } // 3、等待所有任务执行完成,此处最好加超时时间 CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).get(5, TimeUnit.MINUTES); for (CompletableFuture<Integer> future : futures) { Integer num = future.get(); System.out.println("任务执行结果为:" + num); } System.out.println("任务全部执行完成!"); } private Integer addValue(Integer index) { System.out.println("第" + index + "个任务第一次执行"); if (index == 3) { int value = index / 0; } return index + 3; } private Integer plusValue(Integer index, Integer num) { System.out.println("第" + index + "个任务第二次执行,上次执行结果:" + num); return num * 10; } private Integer defaultValue(Integer index, Throwable e) { System.out.println("第" + index + "个任务执行异常!" + e.getMessage()); e.printStackTrace(); return 10; } 作者:京东物流 丁冬 来源:京东云开发者社区 自猿其说Tech

优秀的个人博客,低调大师

解析内存中的高性能图结构

在进行各种图处理、图计算、图查询的时候,内存或是硬盘中如何存储图结构是一个影响性能的关键因素。本文主要分析了几种常见的内存图结构,及其时间、空间复杂度,希望对你有所启发。 通常来说,对于图结构的几种常见的基础操作: 插入一个点 插入一个边 删除一个边 删除一个点的全部邻边 找到一个点的全部邻边 找到一个点的另一个邻点 全图扫描 获取一个点的入度或者出度 这些图相关的操作,除了要关心时间复杂度之外,需要考虑空间占用的问题。 对于大多数实时读写型的系统,增删改查的性能问题会比较重要,它们比较关注上面 1-6 的操作;对于部分密集计算的系统,对批量读取的性能会比较重视,侧重上面 5-8 的操作。 不过遗憾的是,无论是常规的图查询,还是进阶的图计算,根据 RUM 猜想[1],读快、写快、又省空间这样”既要又要也要”的好事是不存在的。 下面,我们介绍几个数据结构并给出少量的定量分析。 我们先从三个典型的方案(邻接矩阵、压缩稀疏矩阵和邻接表)说起,再介绍几种近几年的研究的变种结构 PCSR、VCSR、CSR++。 邻接矩阵 Adjacency Matrix 用矩阵来表示图结构是大学里数据结构课程中都学过的知识,也是一种非常直观的办法。 点 i 与点 j 之间有一条边,等价对应于矩阵上 $M_{ij}$ 为 1。当然,这里的点 ID 是需要连续排列无空洞。 用矩阵来表示图结构有明显的好处,可以复用大量线性代数的研究成果:比如图上模式匹配类的问题等价于矩阵的相乘。 下面是一个模式匹配问题,它的大体意思是在全图上寻找这样一种图结构: Match (src)-[friend * 2..4]->(fof) WHERE src.age > 30 RETURN fof 对于这样一个问题可以直接对应右边的矩阵操作: 比如,2 跳遍历等价于矩阵 F 自乘;2 - 4 跳遍历分别等价于 F^2、F^3、F^4;取属性等价于乘以一个过滤矩阵。 更进一步,由于矩阵操作是天生可以分块并行加速,这在性能上有极大的优势。 下面,我们对邻接矩阵操作进行一个简单的定量分析 操作 时间复杂度 备注 插入一个点 O(n^2) 对于矩阵来说,增加一个点意味着整个矩阵的维度增加,通常需要另外开辟一块空间 插入一个边 O(1) 增加一个边只是将对应的位置置 1 删除一个边 O(1) 置0 删除一个点的全部邻边 O(n) 对于某个点所有出边的删除对应某一行的置0。入边对应某一列,可以批量操作 找到一个点的全部邻边 O(n) 找到一个点的给定邻点 O(1) 全图扫描 O(n^2) 其中,n=|V|,m=|E|。 优化上,批量操作(CPU Cache/SSD block)可以线性增加性能,例如 O(n) 可以降低到 O(n/B),但不影响定量分析。 由于绝大多数图结构是极其稀疏的,因此简单用邻接矩阵来表示图结构,其内存会有夸张的浪费。更为严重的是,当有多种边类型时,每种边类型各需要一个邻接矩阵。这使得裸用矩阵在实际情况中只能处理很小数据量的场景。当然,对于现代服务器动辄几百 G 的内存,如果只有几亿点边的数据量,像是 twitter2010,这并不会是很严重的问题。但大多数情况下,条件允许的话,大家还是希望找到一些更加经济的结构。 压缩稀疏矩阵 CSR/CSC 压缩稀疏矩阵是一种非常流行和紧凑的图结构表示方式,大多数图计算系统都采用 CSR。 这里简单介绍一下 CSR 的结构: 对于点 IDx,取邻居 ID 就是 F[N[x]] 到 F[N[x+1]-1]。 例如,查找点 ID2 的邻居,对应为 F[N[2]] 到 F[N[2+1]-1],对应到上图也即 1 6 8。查找点 ID7 的邻居,对应为 F[N[7]] 到 F[N[7+1]-1],也就是 2 和 4。 另外,CSR 记录的仅是出边的信息,如果要考虑入边就使用 CSC,其原理是类似的。 操作 时间复杂度 备注 插入一个点 O(1) 在点矢量尾部增加 插入一个边 <u,v> O(m+n) 边矢量,从 u 对应的邻居开始,都向后移动一位;点矢量,从 u 对应位置开始每个值加 1 删除一个边 O(m+n) 插入边的逆过程 删除一个点 v 的全部邻边 O(m+n) 边矢量移动 deg(v) 位,点矢量 u 对应位置置 0 找到一个点的全部邻边 O(deg(v)) 找到一个点的给定邻点 O(log(deg(v))) deg(v) 内的排序查找 全图扫描 O(m+n) CSR 的空间占用是 O(|V|+|E|),在空间节省和顺序查找上是极其高效的。但在大量插入时,压缩稀疏矩阵和邻接矩阵一样,需要重新开辟空间,效率很低。所以,它适合于计算密集场景但不适合增改频繁的场景。 CSR 还有一个显著的优点是可以快速获取每个点的出入度,只要计算 N[x+1]-N[x],这在判断一些点是否为超级节点时很方便。如果不是稀疏矩阵的话,通常会用另外一个单独的结构来记录出入度。 此外,CSR 不容易实现并发修改,其每次插入都需要对两个矢量进行位移,这并不高效。 这里推荐两个相关的开源项目,可以进一步了解下 CSR 的使用: https://github.com/DrTimothyAldenDavis/GraphBLAS https://github.com/GraphBLAS/LAGraph 邻接链表 Adjacency List 和基于矩阵的方式不同,邻接链表 AL 空间上有优势,但对于边的读写上会略微慢一点(指针在内存中不能连续移动)。AL 的做法是把邻边(出边)用 list 或者有序 list 的方式串连起来。由此延伸的一个变种是邻边从 list 改为 array。 操作 时间复杂度 备注 插入一个点 O(1) 尾插入 插入一个边 <u,v> O(log(deg(u))) 有序 list 删除一个边 O(log(deg(u))) 删除一个点 v 的全部邻边 O(1) 找到一个点的全部邻边 O(deg(u)) 找到一个点的给定邻点 O(log(deg(u))) 全图扫描 O(n+m) AL 相比 CSR 通常不能直接获得点的出入度,可以通过可以单独维护一个字段实现该功能。 此外,邻接表并不需要 ID 连续排布,对于频繁增删点的场景特别友好。AL 对于并发修改的支持也更友好,天然在点级别有并行度。当然,对于超级节点,直接用 list 的方式,还是会有些性能的问题要考虑;优化上可能会进一步改造成 Blocked list 的方式,可以带来更好的数据局部性和细颗粒度。 由于 ID 不用严格连续排布,AL 的一个常见变种就是 Tree。 Tree 在这种结构中,一个点和其所有的邻边被建模成了 key-value,key 是点 ID,value 是所有邻边的编码。Key 通过 Tree 的方式组织在一起。这里的树可以是 B-Tree 等各变种 Tree。虽然本文没有讨论图属性,但 value 中也是可以存放 value。 操作 时间复杂度 备注 插入一个点 O(log(n)) 插入一个边 <u,v> O(deg(u) log(n)) 删除一个边 O(deg(u) log(n)) 删除一个点 v 的全部邻边 O(log(n)) 找到一个点的全部邻边 O(deg(u) log(n)) 找到一个点的给定邻点 O(log(n deg(u)) 全图扫描 O(n+m) 为了控制访问颗粒度,每个叶子通常会被限定为固定的大小(页)。这就是在数据库类系统里面最常见的办法 B-tree。为了增改方便,也可以把每页的 in-place update 改成 Copy-On-Write 的方式;一个典型的代表就是 LLAMA[2],但这种多版本的读取通常会需要更多的空间,并且当有大量累积修改时,需要定期的多版本合并以降低跨快照读。某种程度上,它和 LSM-Tree 的思路有些接近。 基于 CSR 的变种 PCSR 和 VCSR 由于 CSR 在空间和读性能上有很大的优势,但在插入时的耗时和空间上都很弱,因此本节几个变种的主要目的都是为了改善其弱项,大体思路都是分块和 buffer。 在 CSR 的边矢量进行增删时可以注意到,主要耗时是在对于矢量的元素位移上。因此,一个直观的思路是预留一些插入空白位,在删除时也不立刻回收这些空白。 而分块思想,是指将一些局部数据放在同一个分块内,例如 Tree 中每个 page 就是一种分块的方式。与此对应的是,buffer 空白之间的连续区域。 PCSR PCSR [3]的基本思想是:对于点矢量,其元素从一个值改为对应边矢量中对应邻边位置的 <起点,终点>。而对于边矢量,在这些分块所对应边界放置哨兵 sentinel,上图中的 S,上图的 “-“ 对应预留插入位置留空。 事实上,原文中,对于边矢量,其本质是实现为一个 B-Tree,本文先简化成一个 Array。 操作 时间复杂度 备注 插入一个点 O(1) 直接在点矢量和边矢量的尾插入 插入一个边 <u,v> O(log(deg(u)) 边矢量对应分片的二分+空位查找 删除一个边 O(log(deg(u)) 删除一个点v的全部邻边 O(1) 边矢量对应分片置空,点矢量对应 ID 位置置空 找到一个点的全部邻边 O(deg(u)) 找到一个点的给定邻点 O(log(deg(u)) 全图扫描 O(n+m) 可以看到,PCSR 的预留位置多少都是需要重平衡,不能过多也不能不足。特别是大批量增删时,对预留位置的处理会是一个较重的操作。 此外,如果把 PMA 的 B-Tree 以及需要 rebalance 的本质考虑进去,其和前述 Tree 方式的区别并不是很大。 VCSR VCSR[4]主要是对 PCSR 的一个改进,其朴素思想是 PCSR 的留空是均匀的,而大多数图结构的出入度,是存在 20-80 这样的幂率特征,而 PCSR 的一个主要痛点是频繁的 rebalance。VCSR 的做法是为每个分块预留空间正比于其分块内的点的数量,即:边矢量中,一个分块内,如果点的数量多,就多预留一些空位。在直觉上,点数量多时,其分块对应的边插入会更多一些,这样可以减少 rebalance 的频率。 此外,VCSR 还有些版本号之类的优化。 CSR++ 事实上,CSR++[5]在设计上其实更接近一种 AL/Tree 的变种,而不是 CSR。它主要有三个方面的优化: 第一,对于 Vertex Array 再分段,将一个大的 Array 拆成多段,这样可以有更细的读写颗粒度。通过 段 ID + 点 ID 来定位每个点和其邻边。 第二,Vertex Array 中每个元素,除了记录点 ID 之外,对于邻边数量很少的点,直接把邻居 ID 也对齐地塞 inline 进去。 对于大多数的点,其邻边就不需要单独的 Edge Array 来存储了。 可以看到这种方式在图比较稀疏的时候,对于 CPU Cache 扫描是很友好的。 第三,对于每个点的邻边,采用copy-on-write、标记删除等常见的优化办法,构建成类似 std::vector 结构。 小结 最后,由于在图查询、图存储和图计算不同场景下,对于图结构的读写扫描和生命周期都有些不同的要求,不同的数据结构也有不同的优劣。 当然,本文只是讨论了图结构可以放在内存中的情况。当不得不把部分数据放在硬盘上时,问题就完全不同了。当然本文也没有讨论不同 CPU 对于不同距离内存的性能差异 NUMA,或者跨进程通信带来的影响。 延伸阅读 最后,我们来了解下在图计算/图算法上的图操作。 图算法中的图操作 在图计算中,存在多种图结构算法,可能会涉及多种基础操作。 中心性 Centrality Algorithms,一种衡量一个节点在整个网络图中所在中心程度的算法,包括:度中心性、接近中心性、介数中心性等,会涉及“找到一个点的全部邻边”、“找到一个点的另一个邻点”、“全图扫描”的操作组合。 度中心性通过节点的度数,即关联的边数来刻画节点的受欢迎程度,这将会要求找到一个点的所有邻边。 接近中心性,通过计算每个节点到全图其他所有节点的路径和来刻画节点与其他所有节点的关系密切程度,这将会要求进行全图扫描,查找点和图中所有点的路径信息。 介数中心性,则用于衡量一个顶点出现在其他任意两个顶点对之间最短路径上的次数,从而来刻画节点的重要性,这将会要求进行全图扫描,找到一个点和它的邻点及路径信息 PageRank,又称网页排名、谷歌左侧排名,是一种由搜索引擎根据网页之间相互的超链接计算的技术作为网页排名的要素之一的排名方法。它以 Google 公司创办人拉里·佩奇 Larry Page 之姓来命名。Google 用它来体现网页的相关性和重要性,在搜索引擎优化操作中是经常被用来评估网页优化的成效因素之一。一般来说,PageRank 的值越高,表示其被很多其他网页链接到,说明它的重要性很高;而如果一个 PageRank 很高的网页链接到其他网页,被链接的网页的 PageRank值也会随之提高。PageRank 的计算过程如下: 假设由 4 个网页组成的集合:A、B、C、D,每个页面的初始 PageRank 值相同,为了满足概率值为 0-1 之间,假设这个值为 0.25。 如果所有页面都只链接至 A,如下图,则 A 点的值将是 B、C、D 的 PageRank 值的和。 倘若是下图这种图结构: B 链接到 C,D 链接到 B 和 C,A 点的值计算的方式如下: 这里的 2、1、3,分别是 B 点对外链接的 2 条边,C 点对外链接的 1 条边,D 点对外链接的 3 条边。 换句话说,算法将根据每个页面连出总数平分该页面的 PR 值,并将其加到其所指向的页面: 最后,所有这些 PR 值被换算成百分比形式再乘上一个修正系数。 由于“没有向外链接的网页”传递出去的 PR 值会是0,这时候如果递归的话,会导致指向它的页面的 PR 值的计算结果同样为零,所以 PageRank 会赋给每个页面一个最小值。 因此,一个页面的 PR 值直接取决于指向它的的页面。如果在最初给每个网页一个随机且非零的 PR 值,经过重复计算,这些页面的 PR 值会趋向于某个定值,也就是处于收敛的状态,即最终结果。 简单来说,大多数迭代图计算模型都是基于“找到一个点的全部邻边”、“找到一个点的另一个邻点”操作。 AIGC 小课堂 在 AIGC 小课堂部分,我们会用 NebulaGraph 接入的 aibot 来讲解下文中的部分概念。你如果对 chatgpt 或者是 aibot 有兴趣,可以来 https://discuss.nebula-graph.com.cn/ 向 aibot 提出你的要求。 参考文献 Athanassoulis, M., Kester, M. S., Maas, L. M., et al. (2016). Designing Access Methods: The RUM Conjecture. In EDBT (pp. 461-466). Macko, P., Marathe, V. J., Margo, D. W., et al. (2015). Llama: Efficient Graph Analytics Using Large Multiversioned Arrays. In 2015 IEEE 31st International Conference on Data Engineering (pp. 363-374). IEEE. Wheatman, B., & Xu, H. (2018). Packed Compressed Sparse Row: A Dynamic Graph Representation. In 2018 IEEE High Performance Extreme Computing Conference (HPEC) (pp. 1-7). IEEE. Islam, A. A. R., Dai, D., & Cheng, D. (2022). VCSR: Mutable CSR Graph Format Using Vertex-Centric Packed Memory Array. In 2022 22nd IEEE International Symposium on Cluster, Cloud and Internet Computing (CCGrid) (pp. 71-80). IEEE. Firmli, S., Trigonakis, V., Lozi, J. P., et al. (2020). CSR++: A Fast, Scalable, Update-Friendly Graph Data Structure. In 24th International Conference on Principles of Distributed Systems (OPODIS'20).

优秀的个人博客,低调大师

每日一博 | React Hooks 源码深度解析

作者:京东零售 郑炳懿 前言 React Hooks是React16.8 引入的一个新特性,它允许函数组件中使用state和其他 React 特性,而不必使用类组件。Hooks是一个非常重要的概念,因为它们提供了更简单、更易于理解的React开发体验。 React Hooks的核心源码主要包括两个部分:React内部的Hook管理器和一系列预置的Hook函数。 首先,让我们看一下React内部的Hook管理器。这个管理器是React内部的一个重要机制,它负责管理组件中的所有Hook,并确保它们在组件渲染期间以正确的顺序调用。 内部Hook管理器 示例: const Hook = { queue: [], current: null, }; function useState(initialState) { const state = Hook.current[Hook.queue.length]; if (!state) { Hook.queue.push({ state: typeof initialState === 'function' ? initialState() : initialState, setState(value) { this.state = value; render(); }, }); } return [state.state, state.setState.bind(state)]; } function useHook(callback) { Hook.current = { __proto__: Hook.current, }; try { callback(); } finally { Hook.current = Hook.current.__proto__; } } function render() { useHook(() => { const [count, setCount] = useState(0); console.log('count:', count); setTimeout(() => { setCount(count + 1); }, 1000); }); } render(); 在这个示例中,Hook对象有两个重要属性:queue和current。queue存储组件中所有Hook的状态和更新函数,current存储当前正在渲染的组件的Hook链表。useState和useHook函数则分别负责创建新的Hook状态和在组件中使用Hook。 预置 Hook 函数 useState Hook 以下是useState Hook的实现示例: function useState(initialState) { const hook = updateWorkInProgressHook(); if (!hook.memoizedState) { hook.memoizedState = [ typeof initialState === 'function' ? initialState() : initialState, action => { hook.queue.pending = true; hook.queue.dispatch = action; scheduleWork(); }, ]; } return hook.memoizedState; } 上述代码实现了useState Hook,其主要作用是返回一个state和更新函数的数组,state 初始值为initialState。 在这个实现中,updateWorkInProgressHook()函数用来获取当前正在执行的函数组件的 fiber 对象并判断是否存在对应的hook。它的实现如下: function updateWorkInProgressHook() { const fiber = getWorkInProgressFiber(); let hook = fiber.memoizedState; if (hook) { fiber.memoizedState = hook.next; hook.next = null; } else { hook = { memoizedState: null, queue: { pending: null, dispatch: null, last: null, }, next: null, }; } workInProgressHook = hook; return hook; } getWorkInProgressFiber()函数用来获取当前正在执行的函数组件的fiber对象,workInProgressHook则用来存储当前正在执行的hook对象。在函数组件中,每一个useState调用都会创建一个新的 hook 对象,并将其添加到fiber对象的hooks链表中。这个hooks链表是通过fiber对象的memoizedState属性来维护的。 我们还需要注意到在useState Hook的实现中,每一个hook对象都包含了一个queue对象,用来存储待更新的状态以及更新函数。scheduleWork()函数则用来通知React调度器有任务需要执行。 在React的源码中,useState函数实际上是一个叫做useStateImpl的内部函数。 下面是useStateImpl的源码: function useStateImpl<S>(initialState: (() => S) | S): [S, Dispatch<SetStateAction<S>>] { const dispatcher = resolveDispatcher(); return dispatcher.useState(initialState); } 可以看到,useStateImpl函数的作用就是获取当前的dispatcher并调用它的useState方法,返回一个数组,第一个元素是状态的值,第二个元素是一个dispatch函数,用来更新状态。这里的resolveDispatcher函数用来获取当前的dispatcher,其实现如下: function resolveDispatcher(): Dispatcher { const dispatcher = currentlyRenderingFiber?.dispatcher; if (dispatcher === undefined) { throw new Error('Hooks can only be called inside the body of a function component. (https://fb.me/react-invalid-hook-call)'); } return dispatcher; } resolveDispatcher函数首先尝试获取当前正在渲染的fiber对象的dispatcher属性,如果获取不到则说 明当前不在组件的渲染过程中,就会抛出一个错误。 最后,我们来看一下useState方法在具体的dispatcher实现中是如何实现的。我们以useReducer的 dispatcher为例,它的实现如下: export function useReducer<S, A>( reducer: (prevState: S, action: A) => S, initialState: S, initialAction?: A, ): [S, Dispatch<A>] { const [dispatch, currentState] = updateReducer<S, A>( reducer, // $FlowFixMe: Flow doesn't like mixed types [initialState, initialAction], // $FlowFixMe: Flow doesn't like mixed types reducer === basicStateReducer ? basicStateReducer : updateStateReducer, ); return [currentState, dispatch]; } 可以看到,useReducer方法实际上是调用了一个叫做updateReducer的函数,返回了一个包含当前状态和dispatch函数的数组。updateReducer的实现比较复杂,涉及到了很多细节,这里不再展开介绍。 useEffect Hook useEffect是React中常用的一个Hook函数,用于在组件中执行副作用操作,例如访问远程数据、添加/移除事件监听器、手动操作DOM等等。useEffect的核心功能是在组件的渲染过程结束之后异步执行回调函数,它的实现方式涉及到 React 中的异步渲染机制。 以下是useEffect Hook的实现示例: function useEffect(callback, dependencies) { // 通过调用 useLayoutEffect 或者 useEffect 方法来获取当前的渲染批次 const batch = useContext(BatchContext); // 根据当前的渲染批次判断是否需要执行回调函数 if (shouldFireEffect(batch, dependencies)) { callback(); } // 在组件被卸载时清除当前 effect 的状态信息 return () => clearEffect(batch); } 在这个示例中,useEffect接收两个参数:回调函数和依赖项数组。当依赖项数组中的任何一个值发生变化时, React会在下一次渲染时重新执行useEffect中传入的回调函数。 useEffect函数的实现方式主要依赖于React中的异步渲染机制。当一个组件需要重新渲染时,React会将所有的state更新操作加入到一个队列中,在当前渲染批次结束之后再异步执行这些更新操作,从而避免在同一个渲染批次中连续执行多次更新操作。 在useEffect函数中,我们通过调用useContext(BatchContext)方法来获取当前的渲染批次,并根据shouldFireEffect方法判断是否需要执行回调函数。在回调函数执行完毕后,我们需要通过clearEffect方法来清除当前effect的状态信息,避免对后续的渲染批次产生影响。 总结 总的来说,React Hooks的实现原理并不复杂,它主要依赖于React内部的fiber数据结构和调度系统,通过这些机制来实现对组件状态的管理和更新。Hooks能够让我们在函数组件中使用状态和其他React特性,使得函数组件的功能可以和类组件媲美。 除了useState、useEffect等hook,React还有useContext等常用的Hook。它们的实现原理也基本相似,都是利用fiber架构来实现状态管理和生命周期钩子等功能。 以上是hook简单实现示例,它们并不是React中实际使用的代码,但是可以帮助我们更好地理解hook的核心实现方式。

优秀的个人博客,低调大师

jsoup 1.15.4 发布,Java HTML 解析器

jsoup 1.15.4 现已发布,其中包括了一些改进,尤其是在 pretty-printing HTML 时;以及一些错误的修正。jsoup 是一个用于处理 real-worldHTML 的 Java 库。它使用最好的 HTML5 DOM 方法和 CSS 选择器提供了一个非常方便的 API 用于提取和操作数据。 下载地址:https://jsoup.org/download 具体更新内容包括: Improvements 添加了转义 CSS 选择器(tag、ID、classes)以匹配不遵循常规 CSS 语法的元素的功能。例如,要按类名<p class="one.two">进行匹配,可以使用document.select("p.one\\.two");#838 pretty-printing 时,将<br>标签后面的文本换行。#1858 pretty-printing 时,规范化自定义标签中 self-closing 标签后面的换行符。#1852 pretty-printing 时,折叠 block和inline 标签之间的非重要空白。#1802 在[Element.forEach()]和[Node.forEachNode()]中,使用java.util.function.Consumer而不是以前的 Android ompatibility shim​​[org.jsoup.helper.Consumer]。后者已被弃用。#1870 添加了一个新方法[Document.forms()],可以方便地检索包含文档中<form>元素的List<FormElement>。 增加了一个新方法[Document.expectForm()],找到第一个匹配的[FormElement],或者blow up trying。 Bug 修复 包含[and <code>]等字符的 URL 未正确转义,并且在获取时会抛出一个MalformedURLException错误。#1873 [Element.cssSelector()]将为标签名称、ID 或类名称需要转义的元素创建无效的选择器(例如,如果类名称包含.或:)。#1742 [Element.text()]在 block 和 inline 元素之间应该有一个空格。#1877 如果一个[Node]或[Element]被自身替换,则该节点将被错误地孤立。#1843 在[newRequest()]中,之前请求的表单数据被复制到新的请求中,导致执行多步表单提交时表单数据堆积,或者数据错误发送到后面的请求。现在,[newRequest()]只复制与会话相关的设置(cookie、代理设置、用户代理等),而不是请求数据或正文。#1778 修复了[Safelist.removeAttributes()]中的问题,当使用:allpseudo-attribute时可能抛出一个ConcurrentModificationException。 给定极深嵌套的 HTML,[Element]其中的许多方法可能会由于过多的递归而抛出StackOverflowError。即:#data()、#hasText()、#parents()和#wrap(html)。#1864 Changes 弃用了未使用的[Document.normalise()]方法。规范化发生在 HTML tree构建过程中,不再作为一个独立的阶段。 更新说明:https://github.com/jhy/jsoup/releases/tag/jsoup-1.15.4

资源下载

更多资源
Nacos

Nacos

Nacos /nɑ:kəʊs/ 是 Dynamic Naming and Configuration Service 的首字母简称,一个易于构建 AI Agent 应用的动态服务发现、配置管理和AI智能体管理平台。Nacos 致力于帮助您发现、配置和管理微服务及AI智能体应用。Nacos 提供了一组简单易用的特性集,帮助您快速实现动态服务发现、服务配置、服务元数据、流量管理。Nacos 帮助您更敏捷和容易地构建、交付和管理微服务平台。

Spring

Spring

Spring框架(Spring Framework)是由Rod Johnson于2002年提出的开源Java企业级应用框架,旨在通过使用JavaBean替代传统EJB实现方式降低企业级编程开发的复杂性。该框架基于简单性、可测试性和松耦合性设计理念,提供核心容器、应用上下文、数据访问集成等模块,支持整合Hibernate、Struts等第三方框架,其适用范围不仅限于服务器端开发,绝大多数Java应用均可从中受益。

Rocky Linux

Rocky Linux

Rocky Linux(中文名:洛基)是由Gregory Kurtzer于2020年12月发起的企业级Linux发行版,作为CentOS稳定版停止维护后与RHEL(Red Hat Enterprise Linux)完全兼容的开源替代方案,由社区拥有并管理,支持x86_64、aarch64等架构。其通过重新编译RHEL源代码提供长期稳定性,采用模块化包装和SELinux安全架构,默认包含GNOME桌面环境及XFS文件系统,支持十年生命周期更新。

WebStorm

WebStorm

WebStorm 是jetbrains公司旗下一款JavaScript 开发工具。目前已经被广大中国JS开发者誉为“Web前端开发神器”、“最强大的HTML5编辑器”、“最智能的JavaScript IDE”等。与IntelliJ IDEA同源,继承了IntelliJ IDEA强大的JS部分的功能。

用户登录
用户注册