Aeron Publication.offer() 返回值完全指南

📖 前言

🎯 返回值概览

📊 处理决策流程

graph TB
    Start[调用 publication.offer] --> Check{返回值判断}

    Check -->|≥ 0| Success[发送成功
返回位置]

    Check -->|-1
NOT_CONNECTED| NC[没有订阅者]
    NC --> NC_Action[Thread.yield
然后重试]
    NC_Action --> Start

    Check -->|-2
BACK_PRESSURED| BP[流量控制触发]
    BP --> BP_Action[IdleStrategy.idle
退避等待]
    BP_Action --> Start

    Check -->|-3
ADMIN_ACTION| AA[内部维护操作]
    AA --> AA_Action[立即重试
无需等待]
    AA_Action --> Start

    Check -->|-4
CLOSED| Closed[Publication已关闭]
    Closed --> Closed_Action[抛出异常
或重新创建]

    Check -->|-5
MAX_POSITION| MaxPos[位置达到上限]
    MaxPos --> MaxPos_Action[创建新Stream
迁移订阅者]

    Success --> End[完成]
    Closed_Action --> End
    MaxPos_Action --> End

    style Success fill:#d1e7dd
    style NC fill:#cfe2ff
    style BP fill:#fff3cd
    style AA fill:#ffe5d0
    style Closed fill:#f8d7da
    style MaxPos fill:#212529,color:#fff
                

💎 生产级 ReliablePublisher 实现

以下是一个生产环境可用的可靠消息发布器实现，正确处理了所有返回值情况，并提供了超时、重试、监控等功能。

import io.aeron.Publication;
import io.aeron.logbuffer.BufferClaim;
import org.agrona.DirectBuffer;
import org.agrona.concurrent.BackoffIdleStrategy;
import org.agrona.concurrent.IdleStrategy;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicLong;

/**
 * 生产级可靠消息发布器
 *
 * 特性：
 * 1. 正确处理所有 Publication.offer() 返回值
 * 2. 超时机制，避免无限等待
 * 3. 退避策略，优化 CPU 使用
 * 4. 性能指标收集
 * 5. 自动重连机制（可选）
 */
public class ReliablePublisher {

    private final Publication publication;
    private final IdleStrategy idleStrategy;
    private final long timeoutNs;

    // 性能指标
    private final AtomicLong successCount = new AtomicLong(0);
    private final AtomicLong backPressuredCount = new AtomicLong(0);
    private final AtomicLong adminActionCount = new AtomicLong(0);
    private final AtomicLong notConnectedCount = new AtomicLong(0);

    public ReliablePublisher(Publication publication, long timeoutMs) {
        this.publication = publication;
        this.timeoutNs = TimeUnit.MILLISECONDS.toNanos(timeoutMs);
        this.idleStrategy = new BackoffIdleStrategy(
            100,        // maxSpins
            10,         // maxYields
            1,          // minParkPeriodNs
            1000_000    // maxParkPeriodNs (1ms)
        );
    }

    /**
     * 可靠地发送消息，处理所有错误情况
     *
     * @param buffer 消息缓冲区
     * @param offset 起始偏移量
     * @param length 消息长度
     * @return 消息在流中的位置
     * @throws TimeoutException 如果超时
     * @throws IllegalStateException 如果 Publication 已关闭
     */
    public long offer(DirectBuffer buffer, int offset, int length)
            throws TimeoutException {

        long deadline = System.nanoTime() + timeoutNs;
        long result;

        while (true) {
            result = publication.offer(buffer, offset, length);

            // 发送成功
            if (result >= 0) {
                successCount.incrementAndGet();
                idleStrategy.reset();
                return result;
            }

            // 检查超时
            if (System.nanoTime() > deadline) {
                throw new TimeoutException(
                    String.format("发送超时，最后返回值: %d", result)
                );
            }

            // 根据返回值采取不同策略
            switch ((int) result) {
                case (int) Publication.ADMIN_ACTION:
                    // 立即重试，不等待
                    adminActionCount.incrementAndGet();
                    continue;

                case (int) Publication.BACK_PRESSURED:
                    // 订阅者跟不上，使用退避策略
                    backPressuredCount.incrementAndGet();
                    idleStrategy.idle();
                    break;

                case (int) Publication.NOT_CONNECTED:
                    // 没有订阅者，让出 CPU
                    notConnectedCount.incrementAndGet();
                    Thread.yield();
                    break;

                case (int) Publication.CLOSED:
                    throw new IllegalStateException(
                        "Publication 已关闭，无法发送消息。" +
                        "Channel: " + publication.channel() +
                        ", StreamId: " + publication.streamId()
                    );

                case (int) Publication.MAX_POSITION_EXCEEDED:
                    throw new IllegalStateException(
                        "流位置达到上限，需要创建新流。" +
                        "当前 StreamId: " + publication.streamId()
                    );

                default:
                    throw new IllegalStateException(
                        "未知的返回值: " + result
                    );
            }
        }
    }

    /**
     * 使用 BufferClaim 零拷贝方式发送消息
     */
    public long tryClaim(int length, BufferClaim bufferClaim)
            throws TimeoutException {

        long deadline = System.nanoTime() + timeoutNs;
        long result;

        while (true) {
            result = publication.tryClaim(length, bufferClaim);

            if (result >= 0) {
                successCount.incrementAndGet();
                idleStrategy.reset();
                return result;
            }

            if (System.nanoTime() > deadline) {
                throw new TimeoutException(
                    String.format("Claim 超时，最后返回值: %d", result)
                );
            }

            switch ((int) result) {
                case (int) Publication.ADMIN_ACTION:
                    adminActionCount.incrementAndGet();
                    continue;
                case (int) Publication.BACK_PRESSURED:
                    backPressuredCount.incrementAndGet();
                    idleStrategy.idle();
                    break;
                case (int) Publication.NOT_CONNECTED:
                    notConnectedCount.incrementAndGet();
                    Thread.yield();
                    break;
                case (int) Publication.CLOSED:
                    throw new IllegalStateException("Publication 已关闭");
                case (int) Publication.MAX_POSITION_EXCEEDED:
                    throw new IllegalStateException("流位置达到上限");
            }
        }
    }

    /**
     * 获取性能指标
     */
    public PublisherMetrics getMetrics() {
        return new PublisherMetrics(
            successCount.get(),
            backPressuredCount.get(),
            adminActionCount.get(),
            notConnectedCount.get()
        );
    }

    /**
     * 重置性能指标
     */
    public void resetMetrics() {
        successCount.set(0);
        backPressuredCount.set(0);
        adminActionCount.set(0);
        notConnectedCount.set(0);
    }

    /**
     * 性能指标类
     */
    public static class PublisherMetrics {
        public final long successCount;
        public final long backPressuredCount;
        public final long adminActionCount;
        public final long notConnectedCount;

        public PublisherMetrics(long successCount, long backPressuredCount,
                                long adminActionCount, long notConnectedCount) {
            this.successCount = successCount;
            this.backPressuredCount = backPressuredCount;
            this.adminActionCount = adminActionCount;
            this.notConnectedCount = notConnectedCount;
        }

        public long getTotalRetries() {
            return backPressuredCount + adminActionCount + notConnectedCount;
        }

        public double getRetryRate() {
            if (successCount == 0) return 0.0;
            return (double) getTotalRetries() / successCount;
        }

        @Override
        public String toString() {
            return String.format(
                "PublisherMetrics{success=%d, backPressured=%d, " +
                "adminAction=%d, notConnected=%d, retryRate=%.2f%%}",
                successCount, backPressuredCount, adminActionCount,
                notConnectedCount, getRetryRate() * 100
            );
        }
    }
}

// 使用示例
public class MatchingEngineExample {

    private final ReliablePublisher publisher;

    public void publishMatchingEvent(DirectBuffer buffer, int offset, int length) {
        try {
            long position = publisher.offer(buffer, offset, length);
            System.out.println("事件发送成功，位置: " + position);
        } catch (TimeoutException e) {
            System.err.println("发送超时: " + e.getMessage());
            // 记录日志，触发告警
        } catch (IllegalStateException e) {
            System.err.println("Publication 错误: " + e.getMessage());
            // 尝试重新创建 Publication
        }
    }

    public void printMetrics() {
        ReliablePublisher.PublisherMetrics metrics = publisher.getMetrics();
        System.out.println(metrics);

        // 如果重试率过高，触发告警
        if (metrics.getRetryRate() > 0.1) {  // 超过 10%
            System.err.println("警告：重试率过高，可能存在性能瓶颈");
        }
    }
}

📋 返回值对比总结

📈 监控与指标

关键监控指标

监控实现示例

import io.micrometer.core.instrument.Counter;
import io.micrometer.core.instrument.MeterRegistry;
import io.micrometer.core.instrument.Timer;

public class MonitoredPublisher {

    private final ReliablePublisher publisher;
    private final Counter successCounter;
    private final Counter backPressuredCounter;
    private final Counter adminActionCounter;
    private final Counter notConnectedCounter;
    private final Timer offerTimer;

    public MonitoredPublisher(ReliablePublisher publisher,
                              MeterRegistry registry) {
        this.publisher = publisher;

        // 注册指标
        this.successCounter = Counter.builder("aeron.publish.success")
            .description("成功发送的消息数")
            .register(registry);

        this.backPressuredCounter = Counter.builder("aeron.publish.back_pressured")
            .description("触发流量控制的次数")
            .register(registry);

        this.adminActionCounter = Counter.builder("aeron.publish.admin_action")
            .description("内部维护操作次数")
            .register(registry);

        this.notConnectedCounter = Counter.builder("aeron.publish.not_connected")
            .description("没有订阅者的次数")
            .register(registry);

        this.offerTimer = Timer.builder("aeron.publish.latency")
            .description("发送延迟")
            .register(registry);
    }

    public long offer(DirectBuffer buffer, int offset, int length)
            throws TimeoutException {

        return offerTimer.recordCallable(() -> {
            long result = publisher.offer(buffer, offset, length);

            // 更新指标
            successCounter.increment();

            // 从 publisher 获取详细指标
            ReliablePublisher.PublisherMetrics metrics = publisher.getMetrics();
            backPressuredCounter.increment(metrics.backPressuredCount);
            adminActionCounter.increment(metrics.adminActionCount);
            notConnectedCounter.increment(metrics.notConnectedCount);

            publisher.resetMetrics();

            return result;
        });
    }
}

// Grafana 告警规则示例（PromQL）
// 告警：BACK_PRESSURED 频率过高
rate(aeron_publish_back_pressured_total[1m]) / rate(aeron_publish_success_total[1m]) > 0.05

// 告警：发送延迟 P99 过高
histogram_quantile(0.99, rate(aeron_publish_latency_bucket[1m])) > 0.010

✅ 最佳实践总结