Phaser性能测试加强版

FunTester

发布于 2024-01-17 14:07:33

4840

文章被收录于专栏：FunTesterFunTester

早前的旧文中，我分享了使用 java.util.concurrent.Phaser 在处理大量异步任务场景下的使用。其中用到了phaser类的重要特性可以灵活设置同步数量，在使用过程中注册新的同步对象。

但是在后续的使用过程中遇到的一些问题，主要有一下两点：

注册同步等待总量有上限 private static final int MAX_PARTIES = 0xffff;
功能复杂，API丰富但大部分用不到，在使用过程中经常调错API

今天终于无法忍受，特别是低一点，导致大量异步任务会丢数据。之前是按照非同步方式执行大量任务，但是今天遇到了不定任务量，一时没想到这茬，导致了半个小时数据构造化为泡影。

重写思路

一怒之下，决定自己重写一个加强版。总结了设计思路如下：

线程安全计数，用于统计完成任务数
线程安全状态技术，用于统计多少任务尚未完成
注册方法，用于增加任务统计技术
完成方法，用于减少未完成数量，增加完成任务数量
返回各类状态信息的方法

实现这样的功能，我们就得到了一个简单但加强功能的多线程同步类，用来替代 java.util.concurrent.Phaser ，我命名为 FunPhaser 。代码如下：

package com.funtester.frame  
  
import java.util.concurrent.atomic.AtomicInteger  
  
/**  
 * 自定义同步类,避免{@link java.util.concurrent.Phaser}的不足,总数量受限于65535  
 * 用于多线程任务同步,任务完成后,调用{@link #done()}方法,任务总数减少,当任务总数为0时,调用{@link #await()}方法,等待所有任务完成  
 */  
class FunPhaser extends SourceCode {  
  
    /**  
     * 任务总数索引,用于标记任务完成状态  
     * 注册增加,任务完成减少  
     */  
    AtomicInteger index  
  
    /**  
     * 任务总数,用于记录任务完成数量  
     */  
    AtomicInteger taskNum  
  
    FunPhaser() {  
        this.index = new AtomicInteger()  
        this.taskNum = new AtomicInteger()  
    }  
  
    /**  
     * 注册任务  
     * @return  
     */  
    def register() {  
        this.index.getAndIncrement()  
    }  
  
    /**  
     * 任务完成  
     * @return  
     */  
    def done() {  
        this.index.getAndDecrement()  
        this.taskNum.getAndIncrement()  
    }  
  
    /**  
     * 等待所有任务完成  
     * @return  
     */  
    def await() {  
        waitFor {index.get() == 0}  
    }  
  
    /**  
     * 获取任务完成总数  
     * @return  
     */  
    int queryTaskNum() {  
        return taskNum.get()  
    }  
  
}

源码解读

这个自定义同步类 FunPhaser 用于多线程任务同步，它避免了 java.util.concurrent.Phaser 的不足，即总数量受限于 65535。

FunPhaser 类有以下几个成员变量：

index：任务总数索引，用于标记任务完成状态。注册增加，任务完成减少。
taskNum：任务总数，用于记录任务完成数量。

FunPhaser 类提供了以下几个方法：

index 和 taskNum 是 AtomicInteger 类型的属性，用于原子地操作整数值。
FunPhaser() 是该类的构造函数，初始化了 index 和 taskNum 属性。
register() 方法用于注册任务，每次调用会增加 index 的值，表示新增一个任务。
done() 方法用于标记任务完成，每次调用会减少 index 的值并增加 taskNum 的值。
await() 方法用于等待所有任务完成。它调用了 waitFor 方法，等待 index 的值变为 0，表示所有任务已经完成。
queryTaskNum() 方法用于获取任务完成的总数，返回 taskNum 的值。

演示Demo

FunPhaser 类的使用方法如下：

import com.funtester.frame.FunPhaser;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class FunPhaserDemo {

    public static void main(String[] args) throws InterruptedException {
        // 创建FunPhaser实例
        FunPhaser phaser = new FunPhaser();

        // 创建固定大小的线程池
        ExecutorService executorService = Executors.newFixedThreadPool(5);

        // 注册10个任务
        for (int i = 0; i < 10; i++) {
            executorService.submit(new Runnable() {
                @Override
                public void run() {
                    // 注册任务
                    phaser.register();

                    // 模拟耗时操作
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }

                    // 标记任务完成
                    phaser.done();
                }
            });
        }

        // 等待所有任务完成
        phaser.await();

        // 输出已完成的任务数量
        System.out.println("已完成的任务数量: " + phaser.queryTaskNum());

        // 关闭线程池
        executorService.shutdown();
    }
}

在这个示例中，我们创建了一个FunPhaser对象，并使用固定大小为5的线程池来执行10个异步任务。每个任务在开始前调用register()方法注册，完成后调用done()方法标记。主线程通过调用await()方法等待所有任务完成，并最终输出已完成的任务数量。

自定义关键字

在自定义关键字中的使用如下：

/**  
 * 使用自定义同步器{@link FunPhaser}进行多线程同步  
 *  
 * @param f      代码块  
 * @param phaser 同步器  
 */  
public static void fun(Closure f, FunPhaser phaser) {  
    if (phaser != null) phaser.register();  
    ThreadPoolUtil.executeSync(() -> {  
        try {  
            f.call();  
        } finally {  
            if (phaser != null) {  
                phaser.done();  
                logger.info("async task {}", phaser.queryTaskNum());  
            }  
        }  
    });  
}

作为对照旧的实现代码如下：

/**  
 * 异步执行代码块,使用{@link Phaser}进行多线程同步  
 *  
 * @param f      代码块  
 * @param phaser 同步器  
 */  
public static void fun(Closure f, Phaser phaser) {  
    if (phaser != null) phaser.register();  
    ThreadPoolUtil.executeSync(() -> {  
        try {  
            f.call();  
        } finally {  
            if (phaser != null) {  
                phaser.arrive();  
                logger.info("异步任务完成 {}", phaser.getArrivedParties());  
            }  
        }  
    });  
}

这两个实现代码的功能都是相同的，都是使用同步器来进行多线程任务同步。

旧的实现代码使用的是 Phaser 类。Phaser 类是一个通用的同步器，可以用于各种多线程任务同步场景。在旧的实现代码中，我们使用 register() 方法来注册任务，使用 arrive() 方法来表示任务完成。

新的实现代码使用的是 FunPhaser 类。FunPhaser 类是一个自定义的同步器，它避免了 Phaser 类的不足，即总数量受限于 65535。在新的实现代码中，我们使用 register() 方法来注册任务，使用 done() 方法来表示任务完成。

两种实现代码的对比