一种数据库遍历工具实现

2018-06-27

基于生产消费者模式

外部需要传入查询条件和执行方法

管理者

/**
 * step 1: create or build a TableTraversalManager instance
 * <p>
 * TableTraversalManager buildedManager = TableTraversalManagerBuilder.newInstance().ddbSqlManager(sqlManagerRead).build();
 * <p>
 * <p>
 * TableTraversalManager manager = TableTraversalManager.create(sqlManagerRead);
 * <p>
 * <p>
 * TableTraversalManager specifiedParamsBuildedManager = TableTraversalManagerBuilder.newInstance().ddbSqlManager(sqlManagerRead
 * ).log(log).slaveProducerThreadCount(5).consumerThreadCount
 * (10).deliverQueueSize(1000).producerQueriesPerSecond(200).execPerSecond(100).build();
 * <p>
 * <p>
 * TableTraversalManager specifiedParamsManager = TableTraversalManager.create(log, sqlManagerRead, 5, 10, 1000, 200,
 * 100);
 * <p>
 * 
 * step 2: traversal table:<p>
        manager.traversalTable(new ConditionalTableQuery("my_table", " status = ? and deleted = ? ", Arrays.&lt;Object&gt; asList(1, 0)), exucutor);<p>
 * <p>
 * use  @see TableTraversalUtils.buildConditionalIdRangeCondition(idRange) to build your condition<p>
 * 
 * @author lingshenghang
 */
public class TableTraversalManager {

    public static final int                          DEFAULT_SLAVE_PRODUCER_THREAD_COUNT = 5;
    public static final int                          DEFAULT_CONSUMER_THREAD_COUNT       = 10;
    public static final int                          DEFAULT_DELIVER_QUEU_SIZE           = 1000;

    public static final int                          DEFAULT_PRODUCER_QUERIES_PER_SECOND = TableTraversalProducer.DEFAULT_QUERIES_PER_SECOND;
    public static final int                          DEFAULT_EXEC_PER_SECOND             = TableTraversalConsumer.DEFAULT_EXEC_PER_SECOND;

    private Logger                                   LOG;

    private TableTraversalSqlManager                 tableTraversalSqlManager;

    private BlockingQueue<TableTraversalTaskContext> deliverQueue;

    private long                                     maxCountPerBatch;
    private int                                      slaveProducerThreadCount;
    private int                                      consumerThreadCount;
    private int                                      deliverQueueSize;
    private int                                      producerQueriesPerSecond;
    private int                                      execPerSecond;
    
    private TableTraversalExecutorsFactory           executorsFactory;

    private TableTraversalManager(Logger lOG, TableTraversalSqlManager tableTraversalSqlManager,
        int slaveProducerThreadCount, int consumerThreadCount, int deliverQueueSize, int producerQueriesPerSecond,
        int execPerSecond, TableTraversalExecutorsFactory executorsFactory, long maxCountPerBatch) {
        super();
        LOG = lOG;
        this.tableTraversalSqlManager = tableTraversalSqlManager;
        this.slaveProducerThreadCount = slaveProducerThreadCount;
        this.consumerThreadCount = consumerThreadCount;
        this.deliverQueueSize = deliverQueueSize;
        this.producerQueriesPerSecond = producerQueriesPerSecond;
        this.execPerSecond = execPerSecond;
        this.executorsFactory = executorsFactory;
        this.maxCountPerBatch = maxCountPerBatch;
    }

    public static TableTraversalManager create(SqlManager sqlManager) {
        return create(LoggerFactory.getLogger(TableTraversalManager.class), sqlManager,
            DEFAULT_SLAVE_PRODUCER_THREAD_COUNT, DEFAULT_CONSUMER_THREAD_COUNT, DEFAULT_DELIVER_QUEU_SIZE,
            DEFAULT_PRODUCER_QUERIES_PER_SECOND, DEFAULT_EXEC_PER_SECOND);
    }

    public static TableTraversalManager create(Logger lOG, SqlManager sqlManager, int slaveProducerThreadCount,
        int consumerThreadCount, int deliverQueueSize, int producerQueriesPerSecond, int execPerSecond) {
        
        TableTraversalSqlManager tableTraversalSqlManager = new DdbTableTraversalSqlManager(sqlManager);
        TableTraversalExecutorsFactory executorsFactory = new DefaultTableTraversalExecutorsFactory();
        
        TableTraversalManager instance = new TableTraversalManager(lOG, tableTraversalSqlManager,
            slaveProducerThreadCount, consumerThreadCount, deliverQueueSize, producerQueriesPerSecond, execPerSecond,
            executorsFactory, TableTraversalProducer.DEFAULT_MAX_COUNT_PER_BATCH);
        instance.deliverQueue = new LinkedBlockingQueue<TableTraversalTaskContext>(instance.deliverQueueSize);

        instance.validate();
        return instance;
    }
    
    protected void validate() {
        Assert.notNull(LOG, " LOG is required ");
        Assert.notNull(tableTraversalSqlManager, " tableTraversalSqlManager is required ");
        Assert.notNull(deliverQueue, " deliverQueue is required ");
        Assert.notNull(executorsFactory, " executorsFactory is required ");

        Assert.isTrue(slaveProducerThreadCount > 0, "slaveProducerThreadCount should be positive");
        Assert.isTrue(consumerThreadCount > 0, "consumerThreadCount should be positive");
        Assert.isTrue(deliverQueueSize > 0, "deliverQueueSize should be positive");
        Assert.isTrue(producerQueriesPerSecond > 0, "producerQueriesPerSecond should be positive");
        Assert.isTrue(execPerSecond > 0, "execPerSecond should be positive");
        Assert.isTrue(maxCountPerBatch > 0L, "maxCountPerBatch should be positive");
    }
    

    public void traversalTable(final String tableName, final String condition,
        final ConditionalTableIdRangeExecutor executor) {
        traversalTable(tableName, condition, executor, null, null);
    }

    public void traversalTable(final String tableName, final String condition,
        final ConditionalTableIdRangeExecutor executor, final Long minId, final Long maxId) {
        traversalTable(new ConditionalTableIdRange(tableName, condition, minId, maxId), executor);
    }
    
    public void traversalTable(final ConditionalTableQuery query, final ConditionalTableIdRangeExecutor executor) {
        traversalTable(new ConditionalTableIdRange(query, null, null, false, false), executor);
    }

    public void traversalTable(final ConditionalTableIdRange idRange, final ConditionalTableIdRangeExecutor executor) {
    
        final CountDownLatch countDownLatch = new CountDownLatch(consumerThreadCount + 1);
        final AtomicBoolean produceFinished = new AtomicBoolean(false);

        ExecutorService masterProducerExecutor = executorsFactory.createMasterProducerExecutor();
        ExecutorService slaveProducerExecutors = executorsFactory.createSlaveProducerExecutors(slaveProducerThreadCount);
        ExecutorService consumerExecutors = executorsFactory.createConsumerExecutors(consumerThreadCount);

        try {
            final TableTraversalTaskMonitor monitor = TableTraversalTaskMonitor.create();
            final TableTraversalProducer masterProducer = TableTraversalProducer.create(LOG,
                maxCountPerBatch, producerQueriesPerSecond, slaveProducerExecutors,
                deliverQueue, tableTraversalSqlManager, monitor);

            for (int i = 0; i < consumerThreadCount; i++) {
                final TableTraversalConsumer consumer = TableTraversalConsumer.create(LOG, execPerSecond, deliverQueue,
                    monitor, String.valueOf(i), new ProduceFinishedChecker() {

                        @Override
                        public boolean checkFinished() {
                            return produceFinished.get();
                        }
                    });

                consumerExecutors.submit(new Runnable() {

                    @Override
                    public void run() {
                        try {
                            consumer.consume(executor);
                        } finally {
                            countDownLatch.countDown();
                        }
                    }
                });
            }

            masterProducerExecutor.submit(new Runnable() {

                @Override
                public void run() {
                    try {
                        masterProducer.produce(idRange);
                    } finally {
                        produceFinished.set(true);
                        countDownLatch.countDown();
                    }
                }
            });

            Long start = System.currentTimeMillis();
            LOG.info("Table traversal task start...");
            countDownLatch.await();
            LOG.info("Table traversal task finished ,elapsed time :" + (System.currentTimeMillis() - start) + "ms");
        } catch (InterruptedException e) {
            LOG.error(e.getMessage(), e);
        } finally {
            masterProducerExecutor.shutdown();
            slaveProducerExecutors.shutdown();
            consumerExecutors.shutdown();
        }
    }
    
    public static class TableTraversalManagerBuilder {

        private Logger                         LOG;
        private TableTraversalSqlManager       tableTraversalSqlManager;
        private int                            slaveProducerThreadCount;
        private int                            consumerThreadCount;
        private int                            deliverQueueSize;
        private int                            producerQueriesPerSecond;
        private int                            execPerSecond;
        private TableTraversalExecutorsFactory executorsFactory;
        private long                           maxCountPerBatch;

        private TableTraversalManagerBuilder() {
        }

        public static TableTraversalManagerBuilder newInstance() {
            return new TableTraversalManagerBuilder();
        }

        public TableTraversalManagerBuilder log(Logger log) {
            this.LOG = log;
            return this;
        }

        public TableTraversalManagerBuilder tableTraversalSqlManager(TableTraversalSqlManager tableTraversalSqlManager) {
            this.tableTraversalSqlManager = tableTraversalSqlManager;
            return this;
        }

        public TableTraversalManagerBuilder ddbSqlManager(SqlManager sqlManager) {
            this.tableTraversalSqlManager = new DdbTableTraversalSqlManager(sqlManager);
            return this;
        }

        public TableTraversalManagerBuilder slaveProducerThreadCount(int slaveProducerThreadCount) {
            this.slaveProducerThreadCount = slaveProducerThreadCount;
            return this;
        }

        public TableTraversalManagerBuilder consumerThreadCount(int consumerThreadCount) {
            this.consumerThreadCount = consumerThreadCount;
            return this;
        }

        public TableTraversalManagerBuilder deliverQueueSize(int deliverQueueSize) {
            this.deliverQueueSize = deliverQueueSize;
            return this;
        }

        public TableTraversalManagerBuilder producerQueriesPerSecond(int producerQueriesPerSecond) {
            this.producerQueriesPerSecond = producerQueriesPerSecond;
            return this;
        }

        public TableTraversalManagerBuilder execPerSecond(int execPerSecond) {
            this.execPerSecond = execPerSecond;
            return this;
        }

        public TableTraversalManagerBuilder executorsFactory(TableTraversalExecutorsFactory executorsFactory) {
            this.executorsFactory = executorsFactory;
            return this;
        }
        
        public TableTraversalManagerBuilder maxCountPerBatch(long maxCountPerBatch) {
            this.maxCountPerBatch = maxCountPerBatch;
            return this;
        }

        public TableTraversalManager build() {
            if (LOG == null) {
                LOG = LoggerFactory.getLogger(TableTraversalManager.class);
            }
            if (slaveProducerThreadCount <= 0) {
                slaveProducerThreadCount = TableTraversalManager.DEFAULT_SLAVE_PRODUCER_THREAD_COUNT;
            }
            if (consumerThreadCount <= 0) {
                consumerThreadCount = TableTraversalManager.DEFAULT_CONSUMER_THREAD_COUNT;
            }
            if (deliverQueueSize <= 0) {
                deliverQueueSize = TableTraversalManager.DEFAULT_DELIVER_QUEU_SIZE;
            }
            if (producerQueriesPerSecond <= 0) {
                producerQueriesPerSecond = TableTraversalManager.DEFAULT_PRODUCER_QUERIES_PER_SECOND;
            }
            if (execPerSecond <= 0) {
                execPerSecond = TableTraversalManager.DEFAULT_EXEC_PER_SECOND;
            }
            if (executorsFactory == null) {
                executorsFactory = new DefaultTableTraversalExecutorsFactory();
            }
            if (maxCountPerBatch <= 0L) {
                maxCountPerBatch = TableTraversalProducer.DEFAULT_MAX_COUNT_PER_BATCH;
            }

            TableTraversalManager instance = new TableTraversalManager(LOG, tableTraversalSqlManager,
                slaveProducerThreadCount, consumerThreadCount, deliverQueueSize, producerQueriesPerSecond,
                execPerSecond, executorsFactory, maxCountPerBatch);
            instance.deliverQueue = new LinkedBlockingQueue<TableTraversalTaskContext>(instance.deliverQueueSize);

            instance.validate();
            return instance;
        }

    }
    
}

public class DdbTableTraversalSqlManager implements TableTraversalSqlManager {

    private SqlManager sqlManager;

    public DdbTableTraversalSqlManager(SqlManager sqlManager) {
        this.sqlManager = sqlManager;
    }

    @Override
    public Long queryCount(String queryCountSql, List<Object> params) {
        return sqlManager.queryCount(queryCountSql, params);
    }

    @Override
    public Long queryBoundaryId(String queryBoundaryIdSql, List<Object> params) {
        return sqlManager.queryObject(queryBoundaryIdSql, new DBObjectHandler<Long>() {

            @Override
            public Long handleResultSet(ResultSet arg0) throws SQLException {
                return arg0.getLong(1);
            }
        }, params);
    }

}

public class DefaultTableTraversalExecutorsFactory implements TableTraversalExecutorsFactory {

    @Override
    public ExecutorService createMasterProducerExecutor() {
        return Executors.newSingleThreadExecutor();
    }

    @Override
    public ExecutorService createSlaveProducerExecutors(int slaveProducerThreadCount) {
        return Executors.newFixedThreadPool(slaveProducerThreadCount);
    }

    @Override
    public ExecutorService createConsumerExecutors(int consumerThreadCount) {
        return Executors.newFixedThreadPool(consumerThreadCount);
    }

}

public class TableTraversalTaskContext {

    private ConditionalTableIdRange idRange;
    private Long                   taskQuantity;

    public TableTraversalTaskContext(ConditionalTableIdRange idRange, Long taskQuantity) {
        super();
        this.idRange = idRange;
        this.taskQuantity = taskQuantity;
    }

    public ConditionalTableIdRange getIdRange() {
        return idRange;
    }

    public void setIdRange(ConditionalTableIdRange idRange) {
        this.idRange = idRange;
    }

    public Long getTaskQuantity() {
        return taskQuantity;
    }

    public void setTaskQuantity(Long taskQuantity) {
        this.taskQuantity = taskQuantity;
    }
}

生产者

public class TableTraversalProducer {

    public static final long                        DEFAULT_MAX_COUNT_PER_BATCH = 200L;

    public static final int                         DEFAULT_QUERIES_PER_SECOND  = 200;

    private Logger                                   LOG;

    private long                                     maxCountPerBatch;

    private int                                      queriesPerSecond;

    private ExecutorService                          producerExecutors;

    private BlockingQueue<TableTraversalTaskContext> deliverQueue;

    private CompletionService<Boolean>               producerCompletionService;

    private RateLimiter                              rateLimiter;

    private TableTraversalTaskMonitor                monitor;

    private AtomicInteger                            remainingTask;
    
    private TableTraversalSqlManager                 tableTraversalSqlManager;

    private TableTraversalProducer(Logger logger, long maxCountPerBatch, int queriesPerSecond,
        ExecutorService producerExecutors, BlockingQueue<TableTraversalTaskContext> deliverQueue,
        TableTraversalSqlManager tableTraversalSqlManager, TableTraversalTaskMonitor monitor) {
        LOG = logger;
        this.maxCountPerBatch = maxCountPerBatch;
        this.queriesPerSecond = queriesPerSecond;
        this.producerExecutors = producerExecutors;
        this.deliverQueue = deliverQueue;
        this.tableTraversalSqlManager = tableTraversalSqlManager;
        this.monitor = monitor;
    }

    @Deprecated
    public static TableTraversalProducer create(ExecutorService producerExecutors,
        BlockingQueue<TableTraversalTaskContext> deliverQueue, SqlManager sqlManager, TableTraversalTaskMonitor monitor) {
        return create(DEFAULT_QUERIES_PER_SECOND, producerExecutors, deliverQueue, sqlManager, monitor);
    }

    @Deprecated
    public static TableTraversalProducer create(int queriesPerSecond, ExecutorService producerExecutors,
        BlockingQueue<TableTraversalTaskContext> deliverQueue, SqlManager sqlManager, TableTraversalTaskMonitor monitor) {
        return create(LoggerFactory.getLogger(TableTraversalProducer.class), DEFAULT_MAX_COUNT_PER_BATCH,
            queriesPerSecond, producerExecutors, deliverQueue, sqlManager, monitor);
    }

    @Deprecated
    public static TableTraversalProducer create(Logger logger, long maxCountPerBatch, int queriesPerSecond,
        ExecutorService producerExecutors, BlockingQueue<TableTraversalTaskContext> deliverQueue,
        SqlManager sqlManager, TableTraversalTaskMonitor monitor) {
        DdbTableTraversalSqlManager ddbTableTraversalSqlManager = new DdbTableTraversalSqlManager(sqlManager);
        return create(logger, maxCountPerBatch, queriesPerSecond, producerExecutors, deliverQueue,
            ddbTableTraversalSqlManager, monitor);
    }
    
    public static TableTraversalProducer create(Logger logger, long maxCountPerBatch, int queriesPerSecond,
        ExecutorService producerExecutors, BlockingQueue<TableTraversalTaskContext> deliverQueue,
        TableTraversalSqlManager tableTraversalSqlManager, TableTraversalTaskMonitor monitor) {
        TableTraversalProducer instance = new TableTraversalProducer(logger, maxCountPerBatch, queriesPerSecond,
            producerExecutors, deliverQueue, tableTraversalSqlManager, monitor);
        Assert.notNull(deliverQueue, "deliverQueue must not be null");
        Assert.notNull(tableTraversalSqlManager, "tableTraversalSqlManager must not be null");
        Assert.notNull(producerExecutors, "producerExecutors must not be null");
        instance.rateLimiter = RateLimiter.create(Integer.valueOf(instance.queriesPerSecond).doubleValue());
        instance.producerCompletionService = new ExecutorCompletionService<Boolean>(instance.producerExecutors);
        instance.remainingTask = new AtomicInteger(0);
        return instance;
    }

    @Deprecated
    public void produce(String tableName, String condition) {
        produce(new ConditionalTableIdRange(tableName, condition, null, null));
    }

    @Deprecated
    public void produce(String tableName, String condition, Long minId, Long maxId) {
        produce(new ConditionalTableIdRange(tableName, condition, minId, maxId));
    }

    public void produce(final ConditionalTableIdRange idRange) {
        if (LOG != null && LOG.isInfoEnabled()) {
            LOG.info("table traversal produce start");
        }

        submitTask(new Callable<Boolean>() {

            @Override
            public Boolean call() throws Exception {
                return traversalTable(idRange, false, false);
            }
        });

        while (remainingTask.get() > 0) {
            Boolean result = false;
            try {
                result = producerCompletionService.take().get();
            } catch (Exception e) {
                if (LOG != null && LOG.isErrorEnabled()) {
                    LOG.error(e.getMessage(), e);
                }
            } finally {
                remainingTask.decrementAndGet();
                if (Boolean.TRUE.equals(result)) {
                    TableTraversalUtils.logProgress(LOG, monitor);
                }
            }
        }

        if (LOG != null && LOG.isInfoEnabled()) {
            LOG.info("table traversal produce end");
        }
    }

    private Boolean traversalTable(ConditionalTableIdRange idRange, boolean minIdConfirmed, boolean maxIdConfirmed)
                                                                                                                  throws InterruptedException {
        if (TableTraversalUtils.emptyIdRange(idRange)) {
            commitTaskFinished(0L);
        }

        Long minId = null;
        if (minIdConfirmed && idRange.isMinIdInclusive()) {
            minId = idRange.getMinId();
        } else {
            minId = queryMinId(idRange);
        }

        Long maxId = null;
        if (maxIdConfirmed && idRange.isMaxIdInclusive()) {
            maxId = idRange.getMaxId();
        } else {
            maxId = queryMaxId(idRange);
        }

        long count = 0L;
        if (minId == null || maxId == null) {
            if (LOG != null && LOG.isInfoEnabled()) {
                LOG.info(" none records in query :" + getQueryCountSql(idRange));
            }
        } else {
            count = queryCount(idRange);
        }

        if (count <= 0L) {
            commitTaskFinished(0L);
        }

        if (monitor != null && monitor.getTotalCount() == null) {
            monitor.setTotalCount(count);
        }

        boolean shouldEnqueue = count <= maxCountPerBatch;
        if (shouldEnqueue) {
            TableTraversalTaskContext context = new TableTraversalTaskContext(new ConditionalTableIdRange(idRange,
                minId, maxId, true, true), count);
            deliverQueue.put(context);
            return commitTaskFinished(count);
        } else {
            Long delta = maxId - minId + 1;
            final Long partId = minId + delta / 2;
            final ConditionalTableIdRange lowHalfRange = new ConditionalTableIdRange(idRange, minId, partId, true, false);
            final ConditionalTableIdRange highHalfRange = new ConditionalTableIdRange(idRange, partId, maxId, true, true);

            submitTask(new Callable<Boolean>() {

                @Override
                public Boolean call() throws Exception {
                    return traversalTable(lowHalfRange, true, false);
                }
            });

            submitTask(new Callable<Boolean>() {

                @Override
                public Boolean call() throws Exception {
                    return traversalTable(highHalfRange, false, true);
                }
            });

            return commitTaskFinished(0L);
        }
    }

    private void submitTask(Callable<Boolean> task) {
        remainingTask.incrementAndGet();
        producerCompletionService.submit(task);
    }

    private boolean commitTaskFinished(long produceCount) {
        if (produceCount < 0L) {
            return false;
        } else {
            if (monitor != null) {
                monitor.getProducedCount().addAndGet(produceCount);
            }
            return true;
        }
    }

    private Long queryMinId(ConditionalTableIdRange idRange) {
        return queryBoundaryId("min", idRange);
    }

    private Long queryMaxId(ConditionalTableIdRange idRange) {
        return queryBoundaryId("max", idRange);
    }

    private String getQueryCountSql(ConditionalTableIdRange idRange) {
        final StringBuilder sql = new StringBuilder();
        sql.append(" select count(*) ");
        TableTraversalUtils.buildConditionalIdRangeSql(sql, idRange);
        return sql.toString();
    }

    private Long queryCount(final ConditionalTableIdRange idRange) {

        return TableTraversalUtils.rateLimitedOperate(rateLimiter, new Callable<Long>() {

            @Override
            public Long call() throws Exception {
                return tableTraversalSqlManager.queryCount(getQueryCountSql(idRange), idRange.getParams());
            }
        });
    }

    private Long queryBoundaryId(String boundaryFunc, ConditionalTableIdRange idRange) {
        final StringBuilder sql = new StringBuilder();
        sql.append(" select ");
        sql.append(boundaryFunc);
        sql.append("(id) ");
        TableTraversalUtils.buildConditionalIdRangeSql(sql, idRange);
        final List<Object> params = idRange.getParams();
        return TableTraversalUtils.rateLimitedOperate(rateLimiter, new Callable<Long>() {

            @Override
            public Long call() throws Exception {
                return tableTraversalSqlManager.queryBoundaryId(sql.toString(), params);
            }
        });
    }

}

消费者

public class TableTraversalConsumer {

    public static final int                         DEFAULT_EXEC_PER_SECOND = 100;

    private Logger                                   LOG;

    private int                                      execPerSecond;

    private BlockingQueue<TableTraversalTaskContext> deliverQueue;

    private RateLimiter                              rateLimiter;

    private TableTraversalTaskMonitor                monitor;

    private String                                   consumerId;

    private ProduceFinishedChecker                   produceFinishedChecker;

    private TableTraversalConsumer(Logger lOG, int execPerSecond,
        BlockingQueue<TableTraversalTaskContext> deliverQueue, TableTraversalTaskMonitor monitor, String consumerId,
        ProduceFinishedChecker produceFinishedChecker) {
        LOG = lOG;
        this.execPerSecond = execPerSecond;
        this.deliverQueue = deliverQueue;
        this.monitor = monitor;
        this.consumerId = consumerId;
        this.produceFinishedChecker = produceFinishedChecker;
    }

    public static TableTraversalConsumer create(BlockingQueue<TableTraversalTaskContext> deliverQueue,
        TableTraversalTaskMonitor monitor, String consumerId, ProduceFinishedChecker produceFinishedChecker) {
        return create(DEFAULT_EXEC_PER_SECOND, deliverQueue, monitor, consumerId, produceFinishedChecker);
    }

    public static TableTraversalConsumer create(int execPerSecond,
        BlockingQueue<TableTraversalTaskContext> deliverQueue, TableTraversalTaskMonitor monitor, String consumerId,
        ProduceFinishedChecker produceFinishedChecker) {
        return create(LoggerFactory.getLogger(TableTraversalConsumer.class), execPerSecond, deliverQueue, monitor,
            consumerId, produceFinishedChecker);
    }

    public static TableTraversalConsumer create(Logger log, int execPerSecond,
        BlockingQueue<TableTraversalTaskContext> deliverQueue, TableTraversalTaskMonitor monitor, String consumerId,
        ProduceFinishedChecker produceFinishedChecker) {
        TableTraversalConsumer instance = new TableTraversalConsumer(log, execPerSecond, deliverQueue, monitor,
            consumerId, produceFinishedChecker);
        Assert.notNull(deliverQueue, "deliverQueue must not be null");
        Assert.notNull(produceFinishedChecker, "produceFinishedChecker must not be null");
        instance.rateLimiter = RateLimiter.create(Integer.valueOf(instance.execPerSecond).doubleValue());
        return instance;
    }

    public <T> void consume(ConditionalTableIdRangeExecutor executor) {
        try {
            while (!produceFinishedChecker.checkFinished() || deliverQueue.size() > 0) {
                TableTraversalTaskContext context = deliverQueue.poll(5L, TimeUnit.SECONDS);
                if (context != null) {
                    try {
                        executeTask(context, executor);
                    } catch (Exception e) {
                        LOG.error(e.getMessage(), e);
                    }
                    monitor.getConsumedCount().addAndGet(context.getTaskQuantity());
                    TableTraversalUtils.logProgress(LOG, monitor);
                }
            }
        } catch (Exception e) {
            LOG.error(e.getMessage(), e);
        }
        LOG.info("TableTraversalConsumer: " + consumerId + " finished.");
    }

    private void executeTask(final TableTraversalTaskContext context, final ConditionalTableIdRangeExecutor executor) {
        TableTraversalUtils.rateLimitedOperate(rateLimiter, new Callable<Void>() {

            @Override
            public Void call() throws Exception {
                executor.execute(context.getIdRange());
                return null;
            }
        });
    }
}