写在文章开头

测试驱动开发，也就是我们常说的TDD，按照权威资料的说法，TDD实际上不拘泥于任何形式，可以是单元测试也可以是QA等测试迁移的验收标准，总言之这种模式强调的是通过验收标准辅助开发人员理清需求，从而确保每次迭代、重构都能得到明确的反馈，确保功能落地准确性同时，还能够提升研发人员的生产效率。

而本文将从一个基础并发缓存队列，演示一下如何基于TDD模式完成功能开发、迭代、优化，希望对你有所帮助。

你好，我是 SharkChili ，禅与计算机程序设计艺术布道者，希望我的理念对您有所启发。

📝 我的公众号：写代码的SharkChili
在这里，我会分享技术干货、编程思考与开源项目实践。

🚀 我的开源项目：mini-redis
一个用于教学理解的 Redis 精简实现，欢迎 Star & Contribute：
https://github.com/shark-ctrl/mini-redis (opens new window)

👥 欢迎加入读者群
关注公众号，回复 【加群】 即可获取联系方式，期待与你交流技术、共同成长！

浅谈并发编程的测试理念

相比于常规的串行程序，并发程序的测试要相对复杂一些，因为它的执行存在更多的未知性，可能需要更长的时间才能将这些随机的未知性暴露。对于并发测试的验收标准，我们一般强调如下两个条件：

1. 安全性测试：程序不会发生任何错误的行为
2. 活跃性：某些良好的行为最终会发生

举个简单的例子，假设我们编写一个并发操作安全的队列，按照并发程序的验收标准，安全性测试可以理解为并发操作线程安全，即并发写入10个元素，最终队列中存在的元素就是10个。 同理，活跃性测试则是强调会按照正确的行为发生对应事件，例如阻塞队列容量为10，当写入10个元素后，尝试写入第11个元素的线程会阻塞：

所以对于安全性测试，我们必须学会抓住程序的不变形条件和后验条件，以我们上述说明的并发安全队列为例，则是：

1. 不变形条件：新建的队列，默认大小为0
2. 串行10个元素插入长度为10的队列，最终大小也是10，且不会发生阻塞

这里笔者之所以强调串行验收的概念，是因为针对复杂的并发程序而言，编写必要的串行测试单元，有助于发现一些据竞争之外的问题，以降低后续并发测试的排错成本。

接下来我们再来聊聊阻塞式的测试，按照juc一致性测试的说法，所有的故障都必须与明确的测试用例相关联，按照JSR 166专家组的说法：

每个测试必须等到所有创建的线程都结束才能完成

所以，涉及活跃性测试中的阻塞测试这一环，我们必须做到等待所有创建线程完成工作之后，通过某些手段去感知线程阻塞，结合java线程的特性，所有阻塞挂起的线程都可以通过interrupt方法通知其中断响应，所以对于活跃性的后验测试，我们也可以利用这一点做到：

1. 按照预期要求编写并发程序，等待所有线程运行并达到预期状态(包含阻塞态)
2. 尝试去打认定为阻塞断线程，感知中断异常确认活跃性问题

TDD模式下有界缓存队列的实现

需求说明

接下来笔者就介绍TDD开发模式下完成一个有界缓存的开发，本案例要求我们实现一个并发安全的有界缓存队列的实现要求为：

1. 可灵活指定指定缓存队列容量
2. 没达到上限时非阻塞追加写入元素
3. 成功写入一个元素，对应队列长度+1
4. 取出元素时队列长度扣减，按照先进先出原则
5. 缓存队列达到上界后线程阻塞，直到队列有空闲的空间容纳待写入的元素

业务功能拆解

需求比较简单，所以针对业务功能也没有很复杂的拆解，本质上就是需要生成一个具备如下行为的缓存队列BoundedQueue：

1. 可以指定队列长度
2. 自持查询当前队列长度以及队列是否已满
3. 支持添加元素，以追加的方式
4. 支持移除元素，从队首移除

限定条件：

1. 支持并发操作，即多个线程并发存入，最终得到元素大小就是成功存入队里的元素数
2. 队列无法容纳元素时，生产者线程会阻塞

基于上述业务需求的理解和功能拆解，我们大体得出需要实现如下几个方法：

1. 一个可以指定队列长度的构造方法
2. 一个将元素存入缓存队列的put方法，在队列未达到上限时会阻塞当前线程
3. 一个将元素取出队列的take方法，当队列没有元素的时候会阻塞当前线程

设计落地

结合对于需求的梳理，本次缓存队列的第一版本的实现，比这打算用经典的生产者消费者模式的condition变量来实现，大体来说实现的是通过非空和非满和条件控制写入元素的生产者和消费元素的take调用者正确阻塞和唤醒操作，整体实现思路为：

1. 声明缓存队列BoundedQueue，内部聚合一个链表存储元素
2. 声明生产或者消费队列时要用到的互斥锁lock，确保并发操作互斥
3. 基于lock创建notFull队列，当队列已满时阻塞生产者的等待条件队列notFull
4. 基于lock创建notEmpty队列，当队列已空时阻塞生产者的等待队列非空的条件队列notEmpty
5. 生产元素时，首先获取putLock，成功获取所后，判断队列是否已满，若已满则调用notFull的wait方法等待非满时被唤醒，若非空则写入返回
6. 消费元素时，首先获取takeLock，确保消费元素操作互斥，若为空则调用notEmpty的wait将其阻塞，等待有元素时被唤醒

对应的比这也给出基于这个生产者消费者模型，得出的经典交互流程图：

明确开发思路之后，我们就可以按照并发程序的安全性和活跃性验收标准完成驱动测试用例的设计，辅助理解并梳理并发程序的要求和验收标准，确保开发方向正确且顺序执行，不会被传统TDD所谓红绿灯步骤打断心流：

1. 安全性测试：初始化队列队列size为0
2. 安全性测试：确保空队列添加元素后，返回的size为1
3. 安全性测试：指定队列为10的情况下，10个并发线程都那正确添加成功，且最终size为10
4. 活跃性测试：指定队列为10的情况下，11个线程操作，有一个线程会阻塞

这里笔者专门抽出一个版块，着重强调一个最重要的安全测试，考虑到读者跑程序机器不确定性即在硬件条件较差的机器测试并发程序，很可能会因为创建大量线程导致并行的程序变为串行执行，所以在安全性测试这一块，笔者会考虑如下步骤完成的一个准确的并发正确性测试，对应落地步骤为：

1. 创建CyclicBarrier，当所有生产者、消费者线程就绪后并发生产者消费元素，确保所有线程就绪后，并行交替工作
2. 指定读写元素为整型，且为随机数，这样做的目的是避免编译器预先猜到校验和而去提前优化，导致安全性测试不准确，通过原子类记录生产者消费者处理元素的累加和比对一致进行安全性校验

落地步骤

因为有上述的设计方案和演示测试用例，笔者很清晰的得出下面这个核心代码片段，整体编码思路为：

1. 声明可重入锁确保并发互斥
2. 基于锁构建生产者和消费者的等待条件队列，即生产者监听notFull等待队列非满写入元素，消费者监听notEmpty等待非空消费元素
3. 写入元素函数put，上锁后判断队列是否非满，若满则阻塞，反之写入，消费者同理

对应代码示例如下，读者可结合注释理解一下实现细节

public class BoundedQueue<E> {


    private final ReentrantLock lock = new ReentrantLock();

    /**
     * 生产者写入元素后,会通知notEmpty等待队列的消费者执行消费
     */
    private final Condition notEmpty = lock.newCondition();

    /**
     * 消费者获取元素后,会通知notFull等待队列的生产者执行写入
     */
    private final Condition notFull = lock.newCondition();

    private LinkedList<E> items;

    private int capacity;

    private int size = 0;

    //初始化容量和队列(底层用链表)
    public BoundedQueue(int capacity) {
        this.capacity = capacity;
        items = new LinkedList<>();

    }

    /**
     * 上锁成功后根据是否到达上限决定写入还是阻塞
     * @param e
     * @throws InterruptedException
     */
    public void put(E e) throws InterruptedException {
        try {
            lock.lockInterruptibly();
            //避免唤醒后又满 依然执行写入的错误情况
            while (items.size() == capacity) {
                notFull.await();
            }
            //添加元素
            items.add(e);

            notEmpty.signal();
        } finally {
            lock.unlock();
        }
    }

    public E take() throws InterruptedException {
        try {
            lock.lockInterruptibly();
            //等待非空唤醒消费
            while (items.size() == 0) {
                notEmpty.await();
            }
            //获取元素
            E e = items.remove();
            //消费元素,通知生产者继续写入
            notFull.signal();
            return e;
        } finally {
            lock.unlock();
        }
    }

    public int size() {
        return items.size();
    }

}

并发无关性逻辑验证

同理顺序给出上述几个驱动的测试设计驱动阶段的测试用例落地，第一个则是基本添加功能的验收比较容易：

 //确保空队列添加元素后，返回的size为1
    @Test
    void testAddToEmptyQueue() throws InterruptedException {
        BoundedQueue queue = new BoundedQueue(1);
        queue.put(1);
        assertEquals(1, queue.size());
    }

并发写入一致性验证

第二个则是并发功能的验收，这一点可以参考并发编程的说法，采用后验条件进行验收，由需求可知正确的结构是长度为10的队列10个并发添加是可以正确添加且不阻塞的，所以我们只需通过最终的size即后验条件验收即可：

//指定队列为10的情况下，10个并发线程都那正确添加成功
    @Test
    void testAddToFullQueue() throws InterruptedException {
        BoundedQueue queue = new BoundedQueue(10);
        List<CompletableFuture<Void>> completableFutureList = IntStream.range(0, 10)
                .mapToObj(i -> CompletableFuture.runAsync(() -> {
                    try {
                        queue.put(i);
                    } catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    }
                }))
                .collect(Collectors.toList());

        completableFutureList.stream()
                .map(CompletableFuture::join) //用join阻塞获取结果
                .collect(Collectors.toList());//组成列表

        assertEquals(10, queue.size());


    }

并发阻塞验证

最后一个涉及阻塞操作，如未能正确处理可能导致后验逻辑无法执行，所以我们必须做到让阻塞的线程走到一个目标逻辑方法上。 结合并发编程基础可知，阻塞操作是可以打断的，所以我们的测试可以在等待线程阻塞后将其打断，通过判断线程的存活以及逻辑是否走到阻塞之后的代码段感知阻塞态。

对应代码如下，笔者给定时间等待线程阻塞，然后尝试将其打断确保可以顺利执行后验逻辑，如果触发中断异常则将isBlocking状态设置为true，后验逻辑就可以根据这个标识进行判断:

 //指定队列为1且被使用的情况下,另外一个线程会阻塞
    @Test
    void testAddToFullQueueWithBlocking() throws InterruptedException {
        //设置队列长度为1并写入
        AtomicBoolean isBlocking = new AtomicBoolean(false);
        BoundedQueue queue = new BoundedQueue(1);
        queue.put(1);
        //创建线程再次写入
        Thread thread = new Thread(() -> {
            try {
                queue.put(RandomUtil.randomInt(1000));
                //若走到这里说明线程没有阻塞
                isBlocking.set(false);
            } catch (InterruptedException success) {
                //若线程走到这里则说明成功被打断
                isBlocking.set(true);
            }

        });

        thread.start();
        //让当前线程等待指定线程最多1000毫秒
        thread.join(1000);
        //尝试打断线程
        ThreadUtil.sleep(1000);
        thread.interrupt();
        ThreadUtil.sleep(1000);

        //被打断的线程会死亡,isAlive会变为false
        assertFalse(thread.isAlive());
        //判断是否被阻塞,若打断则原子状态会被设置为true
        assertTrue(isBlocking.get());

    }

安全性测试

最后就是并发数据竞争写入的校验了，和笔者上述设计一致，通过CyclicBarrier确保线程并发无序执行，生产者消费者执行随机数的生产和消费并通过原子类求和，在后验逻辑中验证双方求和是否一致以做到安全性测试校验：

    @Test
    void testConcurrentSafety() throws InterruptedException, BrokenBarrierException {
        int size = Runtime.getRuntime().availableProcessors();
        //创建2N+1个循环栅栏
        CyclicBarrier barrier = new CyclicBarrier((size << 1) + 1);
        ExecutorService threadPool = Executors.newFixedThreadPool(2 * size + 1);

        //计算写入和消费元素和
        AtomicInteger putSum = new AtomicInteger();
        AtomicInteger takeSum = new AtomicInteger();
        //初始化阻塞队列
        BoundedQueue<Integer> queue = new BoundedQueue(100);
        
        //执行N次循环,对应会消费2N个循环栅栏
        for (int i = 0; i < size; i++) {
            threadPool.execute(() -> {
                try {
                    //消费启动栅栏,就绪等待
                    barrier.await();
                    int num = RandomUtil.randomInt(1000);
                    queue.put(num);
                    putSum.addAndGet(num);
                    barrier.await();
                } catch (Exception e) {
                    e.printStackTrace();
                }

            });

            threadPool.execute(() -> {
                try {
                    //消费启动栅栏,就绪等待
                    barrier.await();
                    takeSum.addAndGet(queue.take());
                    barrier.await();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            });
        }
        //消费第2N+1个循环栅栏,启动并发
        barrier.await();
        //2次消费第2N+1个循环栅栏,结束并发
        barrier.await();
        //校验和
        assertEquals(putSum.get(), takeSum.get());


    }

重构与优化

有了上述基础之后，我们后续的重构和优化就变得非常容易了，例如：笔者考虑到当前服务器内存敏感，考虑到阻塞队列容量固定，打算将有界缓存队列底层链表换为局部性友好的数组，此时对应迭代回归我们都可以用到上述的单元测试很快完成迭代后的验收：

 //初始化容量和队列(底层用数组提升执行性能)
    public BoundedQueue(int capacity) {
        this.capacity = capacity;
        items = new ArrayList<>(capacity);

    }

小结

本文详细介绍TDD与传统并发编程程序的设计与实现思路，并结合一些传统的测试技术让java并发起程序安全预期的随机并发执行，避免动态编译等语言特性所带来的误导，最后等待所有创建线程结束后针对不变形条件或者后验条件(例如本文队列中的size或者并发计算时的随机数求和)执行TDD自动化断言，完成逻辑验收。

你好，我是 SharkChili ，禅与计算机程序设计艺术布道者，希望我的理念对您有所启发。

📝 我的公众号：写代码的SharkChili
在这里，我会分享技术干货、编程思考与开源项目实践。

🚀 我的开源项目：mini-redis
一个用于教学理解的 Redis 精简实现，欢迎 Star & Contribute：
https://github.com/shark-ctrl/mini-redis (opens new window)

👥 欢迎加入读者群
关注公众号，回复 【加群】 即可获取联系方式，期待与你交流技术、共同成长！

参考

《java并发编程实战》