@Override
    public void begin(Channels channels) {
        if (running) {
            return;
        }
        lock.lock();
        try {
            this.allocateBuffer2Thread();
            for (ConsumerThread consumerThread : consumerThreads) {
                consumerThread.start();
            }
            running = true;
        } finally {
            lock.unlock();
        }
    }
@Test
    public void testBeginConsumeDriver() {
        Channels<SampleData> channels = new Channels<SampleData>(2, 100, new SimpleRollingPartitioner<SampleData>(), BufferStrategy.BLOCKING);
        ConsumeDriver<SampleData> pool = new ConsumeDriver<SampleData>("default", channels, new SampleConsumer(), 2, 20);
        pool.begin(channels);

        ConsumerThread[] threads = Whitebox.getInternalState(pool, "consumerThreads");
        Assertions.assertEquals(2, threads.length);
        Assertions.assertTrue(threads[0].isAlive());
        Assertions.assertTrue(threads[1].isAlive());
    }