public Object consume() throws InterruptedException {
        int index = consumed - base;
        if (index < 0) {
            logger.debug("negative index {} for base {}.", index, base);
            index += size;
            if (index < 0) {
                logger.error("remove id out of window.");
                return null;
            }
        }
        Object result = buffer.get(index);
        if (result == null) {
            synchronized (consumerLock) {
                while (result == null) {
                    consumerLock.wait();
                    result = buffer.get(index);
                    logger.debug("cell {} is empty. base {}.", index, base);
                }
            }
        }
        if (!buffer.compareAndSet(index, result, null)) {
            logger.error("index {} changed  to {}", index, buffer.get(index));
        }
        logger.debug("consumed id {} of message {}.", consumed++, result);
        if (used.getAndDecrement() == size) {
            synchronized (dispatcherLock) {
                dispatcherLock.notify();
            }
        }
        return result;
    }