利用虚拟线程重写自定义异步功能

最近在使用JDK 21的虚拟线程功能，感觉对于性能测试来说，还是非常值得推广的。通过之前文章介绍，相比各位也有所了解了，这里跳过Java虚拟线程的介绍了。

在官方文档中，虚拟线程其中一个适用场景就是处理多个小异步任务时，本着随用随创建，用完即销毁的理念，不要进行过的的多线程管理和多线程同步设计。

这一点说完是否有些似曾相识，跟Golang应用关键字?go?非常一致，可以说一模一样了。我感觉这个非常适合处理异步任务，所以对原来的自定义异步关键字进行了新版本的开发。旧版本的功能也是根据?go?关键字功能进行开发的。

方案设计

下面分享方案设计的要点

1. 没有采用无限创建虚拟线程的方式，还是用了一个最大并行虚拟线程数量限制

2. 使用任务队列设计，使用了线程安全队列，存储待执行的任务

3. 设计了同款daemon线程，功能与上篇自定义异步文章类似，功能从任务队列中获取并执行任务

4. 在通用的工具类中自定义关键字方法，功能向任务队列中添加任务

代码实现

任务队列

    /**
     * 待执行任务队列,最大容量为MAX_WAIT_TASK
     */
    static LinkedBlockingQueue<Closure> queue = new LinkedBlockingQueue(MAX_WAIT_TASK)

这段代码是在Java中创建了一个静态的待执行任务队列，使用了?LinkedBlockingQueue?类型，并命名为?queue。在创建队列时，使用了?MAX_WAIT_TASK?常量来指定队列的最大容量。

根据代码片段提供的信息，这个队列?queue?的元素类型是?Closure，这可能是一个自定义类型或者来自某个框架或库的特定类。LinkedBlockingQueue?是 Java 中的一个线程安全的队列实现，它使用链表实现了一个阻塞队列，在队列已满或为空时，会对添加或获取元素的操作进行阻塞，直到条件满足。

这段代码创建了一个具有最大容量为?MAX_WAIT_TASK?的阻塞队列，用于存储待执行的任务（Closure?类型的任务）。队列的容量限制可以确保队列不会无限增长，防止内存溢出或其他资源问题。当往队列中添加元素时，如果队列已满，则添加操作会被阻塞，直到有空间可用。

添加任务方法：

  
/**  
 * 添加任务  
 * @param closure  
 * @return  
 */  
static def add(Closure closure) {  
    queue.add(closure)  
}

执行方法

这里写了两个方法，一个执行?java.lang.Runnable?，另外一个执行?groovy.lang.Closure。

/**  
 * 执行任务  
 * @param runnable  
 * @return  
 */  
static def execute(Runnable runnable) {  
    daemon()  
    Thread.startVirtualThread {  
        index.getAndIncrement()  
        SourceCode.noError {  
            runnable.run()  
        }  
        index.getAndDecrement()  
    }  
}  
  
/**  
 * 执行任务  
 * @param closure 任务闭包  
 * @return  
 */  
static def execute(Closure closure) {  
    daemon()  
    Thread.startVirtualThread {  
        index.getAndIncrement()  
        SourceCode.noError {  
            closure()  
        }  
        index.getAndDecrement()  
    }  
}

这段代码片段展示了两个重载的?execute()?方法，用于执行任务。这些方法主要负责启动线程执行任务，并且对执行任务的计数进行增减操作。

1. execute(Runnable runnable)?方法：接受一个?Runnable?参数，该方法会在内部调用?daemon()?方法，确保守护线程已经启动。然后，使用?Thread.startVirtualThread?启动一个虚拟线程，对?index?进行增减操作，并执行传入的?runnable.run()。

2. execute(Closure closure)?方法：接受一个闭包（Closure）作为参数。与前一个方法类似，它也会调用?daemon()?方法以确保守护线程已经启动。然后，使用?Thread.startVirtualThread?启动一个虚拟线程，对?index?进行增减操作，并执行传入的?closure()。

这两个方法的共同点是它们都启动了一个虚拟线程（Virtual Thread），在这些线程中执行了传入的任务（runnable?或?closure），同时通过?index.getAndIncrement()?和?index.getAndDecrement()?对执行任务的计数进行了管理。

daemon线程

  
/**  
 * daemon线程状态,保障只执行一次  
 * @param closure  
 * @return  
 */  
static AtomicBoolean DaemonState = new AtomicBoolean(false)  
  
/**  
 * 最大并发执行任务数量  
 */  
static int MAX_THREAD = 10  
  
/**  
 * 执行daemon线程,保障main方法结束后关闭线程池  
 * @return  
 */  
static def daemon() {  
    def set = DaemonState.getAndSet(true)  
    if (set) return  
    new Thread(new Runnable() {  
  
        @Override  
        void run() {  
            SourceCode.noError {  
                while (ThreadPoolUtil.checkMain()) {  
                    while (index.get() < MAX_THREAD) {  
                        def poll = queue.poll(100, TimeUnit.MILLISECONDS)  
                        if (poll != null) {  
                            execute(poll)  
                        } else {  
                            break  
                        }  
                    }  
                    sleep(0.3)  
                }  
            }  
        }  
    }, "FV").start()  
}

这段代码的功能是创建一个名为?daemon()?的方法，它涉及了一些多线程处理和任务执行控制的逻辑。

1. AtomicBoolean DaemonState = new AtomicBoolean(false)：创建了一个名为?DaemonState?的?AtomicBoolean?类型的变量，用于控制?daemon()?方法是否执行的状态。

2. static int MAX_THREAD = 10：定义了一个整数常量?MAX_THREAD，表示最大并发执行任务数量。

3. daemon()?方法：这是一个多线程的方法，用于执行后台守护线程任务。这个方法通过?DaemonState?的状态控制确保只执行一次。具体实现逻辑如下：

   • 首先，使用?DaemonState.getAndSet(true)?方法检查?DaemonState?的状态，如果已经为?true，则直接返回，确保方法只执行一次。

   • 然后，创建一个新的线程，该线程实现了一个?Runnable?接口，在?run()?方法中执行具体的任务逻辑。

   • 在?run()?方法中，通过?ThreadPoolUtil.checkMain()?方法检查主线程状态，然后进入一个循环。在循环中，检查当前线程池中任务执行的数量，如果小于?MAX_THREAD，则从?queue?中获取任务并执行。

   • queue.poll(100, TimeUnit.MILLISECONDS)?从任务队列?queue?中获取任务，设置了超时时间为 100 毫秒，如果获取到任务则执行?execute(poll)?方法，否则跳出内部循环。

   • 外部循环控制着守护线程的执行条件，使用?sleep(0.3)?控制循环的时间间隔，确保不会过于频繁地检查任务队列。

这里复用了检查main线程的方法，没有进行兜底执行逻辑，所以可能会因为main线程结束过早，导致任务队列积压任务未被执行。我们有增加这个功能也是保持了虚拟线程非线程的思想，这一点跟?go?也保持了一致。

如果想等待的话，可以使用一下方法：

waitFor {  
    VirtualThreadTool.queue.size() == 0  
}

总结

一个简单的异步任务执行框架就完成了，各路大神已经测试过Java虚拟线程和Golang语言的?goroutine?性能，我就不画蛇添足了。

虚拟线程提供了更轻量级的并发模型，能够有效地管理大规模的并发操作，提升应用程序的性能。在性能测试阶段，可以利用虚拟线程模拟并发场景，评估系统在高并发负载下的表现，检测潜在的性能瓶颈，并进行性能优化。

Java虚拟线程拥有广阔的应用前景，但就目前进展上业务服务还需要时间，但是对于性能测试来讲，已经可以提前下手了。

最后：?下方这份完整的软件测试视频教程已经整理上传完成，需要的朋友们可以自行领取【保证100%免费】

软件测试面试文档

我们学习必然是为了找到高薪的工作，下面这些面试题是来自阿里、腾讯、字节等一线互联网大厂最新的面试资料，并且有字节大佬给出了权威的解答，刷完这一套面试资料相信大家都能找到满意的工作。

行动吧，在路上总比一直观望的要好，未来的你肯定会感谢现在拼搏的自己！如果想学习提升找不到资料，没人答疑解惑时，请及时加入群：1150305204，里面有各种测试开发资料和技术可以一起交流哦。