| 45 | private final String queueName; |
| 46 | |
| 47 | public JCQueue(String queueName, String metricNamePrefix, int size, int overflowLimit, int producerBatchSz, |
| 48 | IWaitStrategy backPressureWaitStrategy, String topologyId, String componentId, List<Integer> taskIds, |
| 49 | int port, StormMetricRegistry metricRegistry) { |
| 50 | this.queueName = queueName; |
| 51 | this.overflowLimit = overflowLimit; |
| 52 | this.recvQueue = new MpscArrayQueue<>(size); |
| 53 | this.overflowQ = new MpscUnboundedArrayQueue<>(size); |
| 54 | |
| 55 | for (Integer taskId : taskIds) { |
| 56 | this.jcqMetrics.add(new JCQueueMetrics(metricNamePrefix, topologyId, componentId, taskId, port, |
| 57 | metricRegistry, recvQueue, overflowQ)); |
| 58 | } |
| 59 | |
| 60 | //The batch size can be no larger than half the full recvQueue size, to avoid contention issues. |
| 61 | this.producerBatchSz = Math.max(1, Math.min(producerBatchSz, size / 2)); |
| 62 | this.backPressureWaitStrategy = backPressureWaitStrategy; |
| 63 | } |
| 64 | |
| 65 | public String getQueueName() { |
| 66 | return queueName; |