### Question:
CarreraConsumer extends BaseCarreraConsumerPool { public synchronized void startConsume(MessageProcessor processor, int concurrency) { baseStartConsume(processor, concurrency, Collections.EMPTY_MAP); } CarreraConsumer(CarreraConfig config); synchronized void startConsume(MessageProcessor processor, int concurrency); synchronized void startConsume(MessageProcessor processor, int concurrency, Map<String, Integer> extraConcurrency); synchronized void startConsume(BatchMessageProcessor processor, int concurrency); synchronized void startConsume(BatchMessageProcessor processor, int concurrency, Map<String, Integer> extraConcurrency); @Deprecated List<ConsumeStats> getConsumeStats(); @Deprecated List<ConsumeStats> getConsumeStats(String topic); @Deprecated void close(); @Override synchronized void stop(); static final String CARRERA_CONSUMER_TYPE; }### Answer:
@Test public void testLowLevelPoolCarreraConsumerException() throws InterruptedException { CountDownLatch latch = new CountDownLatch(1); lowLevelPoolCarreraConsumer.startConsume(new MessageProcessor() { @Override public Result process(Message message, Context context) { System.out.println("receive msg:" + message); return Result.SUCCESS; } }, 10); latch.await(); }
@Test public void testLowLevelCarreraConsumer() throws InterruptedException { lowLevelCarreraConsumer.startConsume(new MessageProcessor() { @Override public Result process(Message message, Context context) { System.out.println("receive msg:" + context.toString() + message.toString()); return Result.SUCCESS; } }); }
@Test public void testSimpleCarreraConsumer() throws InterruptedException{ simpleCarreraConsumer.startConsume(new MessageProcessor() { @Override public Result process(Message message, Context context) { System.out.println("receive msg:" + context.toString() + message.toString()); return Result.SUCCESS; } }); }