java - 实现模拟线程#sleep()

Question

背景

我正在设计我的软件，以便我可以轻松地执行单元测试。我有一个 IClock 接口(interface)，除其他方法外，它有 IClock#wait(TimeUnit timeUnit，持续时间长)。此方法将暂停当前线程 timeUnit 持续时间(即 1 秒)。

IClock 接口(interface)有两种实现:

• SimulatedClock:具有手动增加存储在时钟中的时间的方法
• RealClock:引用System.currentTimeMillis()
自动增加时间

这是 IClock#wait(...) 的默认方法:

/**
     * Locks current thread for specified time
     *
     * @param timeUnit
     * @param dt
     */
    default void wait(TimeUnit timeUnit, long dt)
    {
        Lock lock = new ReentrantLock();
        scheduleIn(timeUnit, dt, lock::unlock);
        lock.lock();
    }

问题

我希望模拟单元测试的当前工作方式是

1. 开始线程
2. 等待所有线程完成或处于阻塞状态(我假设如果它们被阻塞，它们已经调用了 IClock#wait(...))
3. 如果所有线程都已完成，则结束。否则，将 SimulatedClock 时间增加一毫秒。

然而，真正发生的是:

1. 开始线程
2. 开始增加时间，即使线程还没有第一次调用 IClock#wait()。

所以，我需要做的是能够确定所有线程何时完成或阻塞。虽然这可以通过 Thread#getState() 来完成，但我宁愿采用一种更优雅的解决方案，并且可以与 ForkJoinPool 一起使用。

完整代码

GitHub

模拟时钟

package com.team2502.ezauton.utils;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;

public class SimulatedClock implements IClock
{

    private long time = 0;

    private List<Job> jobs = new ArrayList<>();

    public SimulatedClock() {}

    public void init()
    {
        init(System.currentTimeMillis());
    }

    public void init(long time)
    {
        setTime(time);
    }

    /**
     * Add time in milliseconds
     *
     * @param dt millisecond increase
     * @return The new time
     */
    public long addTime(long dt)
    {
        setTime(getTime() + dt);
        return getTime();
    }

    /**
     * Adds time with units
     *
     * @param timeUnit
     * @param value
     */
    public void addTime(TimeUnit timeUnit, long value)
    {
        addTime(timeUnit.toMillis(value));
    }

    /**
     * Add one millisecond and returns new value
     *
     * @return The new time
     */
    public long incAndGet()
    {
        return addTime(1);
    }

    /**
     * Increment a certain amount of times
     *
     * @param times
     */
    public void incTimes(long times, long dt)
    {
        long init = getTime();
        long totalDt = times * dt;
        for(int i = 0; i < times; i++)
        {
            if(!jobs.isEmpty())
            {
                addTime(dt);
            }
            else
            {
                break;
            }
        }
        setTime(init + totalDt);
    }

    /**
     * Increment a certain amount of times
     *
     * @param times
     * @return
     */
    public void incTimes(long times)
    {
        incTimes(times, 1);
    }

    @Override
    public long getTime()
    {
        return time;
    }

    public void setTime(long time)
    {
        jobs.removeIf(job -> {
            if(job.getMillis() < time)
            {
                job.getRunnable().run();
                return true;
            }
            return false;
        });

        this.time = time;
    }

    @Override
    public void scheduleAt(long millis, Runnable runnable)
    {
        if(millis < getTime())
        {
            throw new IllegalArgumentException("You are scheduling a task for before the current time!");
        }
        jobs.add(new Job(millis, runnable));
    }

    private static class Job
    {
        private final long millis;
        private final Runnable runnable;

        public Job(long millis, Runnable runnable)
        {
            this.millis = millis;
            this.runnable = runnable;
        }

        public long getMillis()
        {
            return millis;
        }

        public Runnable getRunnable()
        {
            return runnable;
        }
    }
}

模拟

package com.team2502.ezauton.command;

import com.team2502.ezauton.utils.SimulatedClock;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.TimeUnit;

public class Simulation
{

    private final SimulatedClock simulatedClock;
    private List<IAction> actions = new ArrayList<>();

    public Simulation()
    {
        simulatedClock = new SimulatedClock();
    }

    public SimulatedClock getSimulatedClock()
    {
        return simulatedClock;
    }

    public Simulation add(IAction action)
    {
        actions.add(action);
        return this;
    }

    /**
     * @param timeoutMillis Max millis
     */
    public void run(long timeoutMillis)
    {
        simulatedClock.init();

        actions.forEach(action -> new ThreadBuilder(action, simulatedClock).buildAndRun());

        simulatedClock.incTimes(timeoutMillis);

        // Need to wait until all threads are finished
        if(!ForkJoinPool.commonPool().awaitQuiescence(1, TimeUnit.SECONDS))
        {
            throw new RuntimeException("Simulator did not finish in a second.");
        }

    }

    public void run(TimeUnit timeUnit, long value)
    {
        run(timeUnit.toMillis(value));
    }
}

示例单元测试

@Test
public void testSimpleAction()
{
    AtomicBoolean atomicBoolean = new AtomicBoolean(false);
    Simulation simulation = new Simulation();
    simulation.add(new DealyedAction((TimeUnit.SECONDS, 5) -> atomicBoolean.set(true)));
    simulation.run(TimeUnit.SECONDS, 100);
    Assert.assertTrue(atomicBoolean.get());
}

Answer 1

在调用 simulatedClock.incTimes() 之前，各个线程似乎没有及时运行。

通常在多线程测试中，开始时会有某种“集合点”——允许所有线程在安全启动并运行后 checkin 。如果您预先知道有多少个线程，CountDownLatch 可以让这变得简单。

例如在 Simulation.run() 中:

    simulatedClock.init(new CountDownLatch(actions.size()));

它为以后保留对 CountDownLatch 的引用。

当每个线程到达 SimulatedClock.scheduleAt() 时，它可以将闩锁减一:

@Override
public void scheduleAt(long millis, Runnable runnable)
{
    if(millis < getTime())
    {
        throw new IllegalArgumentException("You are scheduling a task for before the current time!");
    }
    jobs.add(new Job(millis, runnable));

    countDownLatch.countDown();
}

然后 incTimes() 可以等待所有线程出现:

public void incTimes(long times, long dt)
{
    countDownLatch.await();

    long init = getTime();
    ...