ConsumeDriver implements IDriver { @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(); } } ConsumeDriver(String name, Channels<T> channels, Class<? extends IConsumer<T>> consumerClass, int num,
        long consumeCycle);  ConsumeDriver(String name, Channels<T> channels, IConsumer<T> prototype, int num, long consumeCycle); private  ConsumeDriver(Channels<T> channels, int num); @Override void begin(Channels channels); @Override boolean isRunning(Channels channels); @Override void close(Channels channels);  }
@Test public void testBeginConsumeDriver() throws IllegalAccessException { 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 = (ConsumerThread[]) MemberModifier.field(ConsumeDriver.class, "consumerThreads") .get(pool); Assert.assertEquals(2, threads.length); Assert.assertTrue(threads[0].isAlive()); Assert.assertTrue(threads[1].isAlive()); }