private AbstractWrappedQueueExt queue;
private JRubyAbstractQueueWriteClientExt inputQueueClient;
private QueueReadClientBase filterQueueClient;

@JRubyMethod(name = "open_queue")
public final IRubyObject openQueue(final ThreadContext context) throws IOException {
    try {
        queue = QueueFactoryExt.create(context, null, settings);
    } catch (final Exception ex) {
        LOGGER.error("Logstash failed to create queue.", ex);
        throw new IllegalStateException(ex);
    }
    inputQueueClient = queue.writeClient(context);
    filterQueueClient = queue.readClient();

    // 初始化性能监测指标
    ...
    return context.nil;
}

@JRubyMethod(meta = true)
public static AbstractWrappedQueueExt create(final ThreadContext context, final IRubyObject recv,
    final IRubyObject settings) throws IOException {
    final String type = getSetting(context, settings, "queue.type").asJavaString();
    if ("persisted".equals(type)) {
        // 初始化持久化队列
        ...
    } else if ("memory".equals(type)) {
        return new JrubyWrappedSynchronousQueueExt(
            context.runtime, RubyUtil.WRAPPED_SYNCHRONOUS_QUEUE_CLASS
        ).initialize(
            context, context.runtime.newFixnum(
                getSetting(context, settings, "pipeline.batch.size")
                    .convertToInteger().getIntValue()
                    * getSetting(context, settings, "pipeline.workers")
                    .convertToInteger().getIntValue()
            )
        );
    } else {
        // 异常处理
        ...
    }
}

private BlockingQueue<JrubyEventExtLibrary.RubyEvent> queue;

@JRubyMethod
@SuppressWarnings("unchecked")
public JrubyWrappedSynchronousQueueExt initialize(final ThreadContext context,
    IRubyObject size) {
    int typedSize = ((RubyNumeric)size).getIntValue();
    this.queue = new ArrayBlockingQueue<>(typedSize);
    return this;
}

@Override
protected JRubyAbstractQueueWriteClientExt getWriteClient(final ThreadContext context) {
    return JrubyMemoryWriteClientExt.create(queue);
}

@Override
protected QueueReadClientBase getReadClient() {
    // batch size and timeout are currently hard-coded to 125 and 50ms as values observed
    // to be reasonable tradeoffs between latency and throughput per PR #8707
    return JrubyMemoryReadClientExt.create(queue, 125, 50);
}

@JRubyMethod(name = "wrapped_write_client", visibility = Visibility.PROTECTED)
public final JRubyWrappedWriteClientExt wrappedWriteClient(final ThreadContext context,
    final IRubyObject pluginId) {
    return new JRubyWrappedWriteClientExt(context.runtime, RubyUtil.WRAPPED_WRITE_CLIENT_CLASS)
        .initialize(inputQueueClient, pipelineId.asJavaString(), metric, pluginId);
}

@JRubyMethod(name = {"push", "<<"}, required = 1)
public IRubyObject push(final ThreadContext context, final IRubyObject event)
    throws InterruptedException {
    final long start = System.nanoTime();
    // 计数 + 1
    incrementCounters(1L);
    // 调用 QueueWriteClient 向阻塞队列中 put 一条消息
    final IRubyObject res = writeClient.doPush(context, (JrubyEventExtLibrary.RubyEvent) event);
    // 累加耗时
    incrementTimers(start);
    return res;
}

@Override
public QueueBatch readBatch() throws InterruptedException {
    MemoryReadBatch batch = MemoryReadBatch.create(
            LsQueueUtils.drain(queue, batchSize, waitForNanos));
    startMetrics(batch);
    return batch;
}

public static LinkedHashSet<JrubyEventExtLibrary.RubyEvent> drain(
    final BlockingQueue<JrubyEventExtLibrary.RubyEvent> queue, final int count, final long nanos
) throws InterruptedException {
    int left = count;
    //todo: make this an ArrayList once we remove the Ruby pipeline/execution
    final LinkedHashSet<JrubyEventExtLibrary.RubyEvent> collection =
        new LinkedHashSet<>(4 * count / 3 + 1);
    do {
        final int drained = drain(queue, collection, left, nanos);
        if (drained == 0) {
            break;
        }
        left -= drained;
    } while (left > 0);
    return collection;
}

private static int drain(final BlockingQueue<JrubyEventExtLibrary.RubyEvent> queue,
    final Collection<JrubyEventExtLibrary.RubyEvent> collection, final int count,
    final long nanos) throws InterruptedException {
    int added = 0;
    do {
        added += queue.drainTo(collection, count - added);
        if (added < count) {
            final JrubyEventExtLibrary.RubyEvent event =
                queue.poll(nanos, TimeUnit.NANOSECONDS);
            if (event == null) {
                break;
            }
            collection.add(event);
            added++;
        }
    } while (added < count);
    return added;
}

@JRubyMethod(name = OUTPUT_METHOD_NAME)
public IRubyObject multiReceive(final IRubyObject events) {
    final RubyArray batch = (RubyArray) events;
    final int count = batch.size();
    // 增加输入计数
    eventMetricIn.increment((long) count);
    final long start = System.nanoTime();
    // 执行 output 插件处理逻辑
    doOutput(batch);
    // 增加耗时
    eventMetricTime.increment(
        TimeUnit.MILLISECONDS.convert(System.nanoTime() - start, TimeUnit.NANOSECONDS)
    );
    // 增加输出计数
    eventMetricOut.increment((long) count);
    return this;
}