Class JCQueue

storm-client/src/jvm/org/apache/storm/utils/JCQueue.java:32–411

import org.slf4j.LoggerFactory;

@SuppressWarnings("checkstyle:AbbreviationAsWordInName")
public class JCQueue implements Closeable {
    private static final Logger LOG = LoggerFactory.getLogger(JCQueue.class);
    private final ExitCondition continueRunning = () -> true;
    private final List<JCQueueMetrics> jcqMetrics = new ArrayList<>();
    private final MpscArrayQueue<Object> recvQueue;
    // only holds msgs from other workers (via WorkerTransfer), when recvQueue is full
    private final MpscUnboundedArrayQueue<Object> overflowQ;
    private final int overflowLimit; // ensures... overflowCount <= overflowLimit. if set to 0, disables overflow limiting.
    private final int producerBatchSz;
    private final DirectInserter directInserter = new DirectInserter(this);
    private final ThreadLocal<BatchInserter> thdLocalBatcher = new ThreadLocal<BatchInserter>(); // ensure 1 instance per producer thd.
    private final IWaitStrategy backPressureWaitStrategy;
    private final String queueName;

    public JCQueue(String queueName, String metricNamePrefix, int size, int overflowLimit, int producerBatchSz,
                   IWaitStrategy backPressureWaitStrategy, String topologyId, String componentId, List<Integer> taskIds,
                   int port, StormMetricRegistry metricRegistry) {
        this.queueName = queueName;
        this.overflowLimit = overflowLimit;
        this.recvQueue = new MpscArrayQueue<>(size);
        this.overflowQ = new MpscUnboundedArrayQueue<>(size);

        for (Integer taskId : taskIds) {
            this.jcqMetrics.add(new JCQueueMetrics(metricNamePrefix, topologyId, componentId, taskId, port,
                    metricRegistry, recvQueue, overflowQ));
        }

        //The batch size can be no larger than half the full recvQueue size, to avoid contention issues.
        this.producerBatchSz = Math.max(1, Math.min(producerBatchSz, size / 2));
        this.backPressureWaitStrategy = backPressureWaitStrategy;
    }

    public String getQueueName() {
        return queueName;
    }

    @Override
    public void close() {
        for (JCQueueMetrics jcQueueMetric : jcqMetrics) {
            jcQueueMetric.close();
        }
    }

    /**
     * Non blocking. Returns immediately if Q is empty. Returns number of elements consumed from Q.
     */
    public int consume(JCQueue.Consumer consumer) {
        return consume(consumer, continueRunning);
    }

    /**
     * Non blocking. Returns immediately if Q is empty. Runs till Q is empty OR exitCond.keepRunning() return false. Returns number of
     * elements consumed from Q.
     */
    public int consume(JCQueue.Consumer consumer, ExitCondition exitCond) {
        try {
            return consumeImpl(consumer, exitCond);