写点什么

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

作者:FunTester
  • 2024-01-03
    河北
  • 本文字数:3027 字

    阅读完需:约 10 分钟

最近在使用 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),在这些线程中执行了传入的任务(runnableclosure),同时通过 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):创建了一个名为 DaemonStateAtomicBoolean 类型的变量,用于控制 daemon() 方法是否执行的状态。

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

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

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

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

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

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

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


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


如果想等待的话,可以使用一下方法:


waitFor {      VirtualThreadTool.queue.size() == 0  }
复制代码

总结

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


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


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


发布于: 刚刚阅读数: 4
用户头像

FunTester

关注

公众号:FunTester,800篇原创,欢迎关注 2020-10-20 加入

Fun·BUG挖掘机·性能征服者·头顶锅盖·Tester

评论

发布
暂无评论
利用虚拟线程重写自定义异步功能_FunTester_InfoQ写作社区