Agrona 高性能工具库

编译项目

cd 09-agrona
mvn clean compile

运行示例

# 方式1: 使用完整类名
mvn exec:java -Dexec.mainClass="com.agrona.demo.agent.SimpleAgentDemo"

# 方式2: 使用配置ID
mvn exec:java@simple-agent

Agent 接口

public interface Agent {
    int doWork() throws Exception;    // 执行工作，返回完成的工作量
    String roleName();                 // Agent的角色名称
    void onStart();                    // 启动时回调
    void onClose();                    // 关闭时回调
}

AgentRunner

负责在独立线程中运行Agent，管理生命周期，应用空闲策略，处理异常。

AgentRunner runner = new AgentRunner(
    idleStrategy,      // 空闲策略
    errorHandler,      // 错误处理器
    null,              // 异常计数器（可选）
    agent              // Agent实例
);

Thread thread = AgentRunner.startOnThread(runner);

1. BusySpinIdleStrategy - 忙等待

• 最低延迟（微秒级）
• 高CPU使用率（100%）
• 适用于超低延迟场景

2. BackoffIdleStrategy - 渐进式退避

• 自旋 → 让步 → 短暂休眠 → 长时间休眠
• 平衡延迟和CPU使用
• 最常用的策略

3. SleepingIdleStrategy - 休眠策略

• 直接休眠
• 最低CPU使用
• 适用于低负载场景

JDK HashMap<Long, Object> 的隐藏成本：

// JDK HashMap - 每次put都会装箱
HashMap<Long, Order> jdkMap = new HashMap<>();
jdkMap.put(1001L, order);  // 1001L 被装箱成 Long对象

// 装箱过程（自动发生）：
Long boxed = Long.valueOf(1001L);  // 创建Long对象
jdkMap.put(boxed, order);          // 使用Long对象作为key

// 每次get也需要装箱
Order found = jdkMap.get(1002L);   // 1002L 被装箱成 Long对象

// 遍历时拆箱
for (Long id : jdkMap.keySet()) {
    long primitiveId = id;  // 拆箱！
}

存储100万个Long到Order的映射：

什么是链表法？

链表法（Separate Chaining）是一种解决哈希冲突的经典方法。每个哈希桶（bucket）存储一个链表， 当多个key的hash值相同时，将它们串成一条链表存储在同一个桶中。

工作原理：

• 计算hash：对key计算哈希值，确定桶的位置
• 定位桶：通过 hash % capacity 找到数组索引
• 遍历链表：如果桶中已有元素，遍历链表查找或插入
• 对象结构：每个节点是一个Entry对象，包含 hash, key, value, next 四个字段

JDK优化：当链表长度超过8时，JDK会将链表转换为红黑树，查找性能从O(n)提升到O(log n)

💡 链表法举例说明

场景：向一个容量为8的HashMap插入三个订单ID：1001L、1009L、1017L

步骤1：插入 1001L

• 计算哈希：hash(1001L) = 12345
• 定位桶：12345 % 8 = 1 → bucket[1]
• bucket[1]为空，创建Entry对象：Entry(hash=12345, key=1001L, value=Order@1, next=null)
• 将Entry存入bucket[1]

步骤2：插入 1009L

• 计算哈希：hash(1009L) = 12353
• 定位桶：12353 % 8 = 1 → bucket[1]（冲突！）
• bucket[1]已有Entry，创建新Entry：Entry(hash=12353, key=1009L, value=Order@2, next=null)
• 将新Entry插入链表头部：new Entry → old Entry

步骤3：插入 1017L

• 计算哈希：hash(1017L) = 12361
• 定位桶：12361 % 8 = 1 → bucket[1]（再次冲突！）
• bucket[1]已有链表，创建新Entry：Entry(hash=12361, key=1017L, value=Order@3, next=null)
• 将新Entry插入链表头部：Entry(1017L) → Entry(1009L) → Entry(1001L) → null

最终结构：

buckets数组：
[0]: null
[1]: Entry(1017L) → Entry(1009L) → Entry(1001L) → null  // 链表长度3
[2]: null
[3]: null
[4]: null
[5]: null
[6]: null
[7]: null

查找过程（查找1009L）：

• 计算哈希并定位到bucket[1]
• 遍历链表：检查Entry(1017L) → 不匹配
• 继续遍历：检查Entry(1009L) → 匹配！返回Order@2
• 总共需要：2次链表遍历（指针追踪）

⚠️ 性能问题：链表越长，查找越慢。三个元素都散落在堆内存不同位置，CPU缓存命中率低。

链表法的问题：

• 指针追踪：每个Entry节点需要next指针，增加内存占用（8字节/entry）
• 缓存不友好：链表节点散落在堆内存各处，CPU缓存行无法有效利用
• 内存碎片：每个Entry是独立对象，分配在堆的不同位置
• 多次解引用：查找需要：数组→Entry→Long对象→hashCode比较

什么是开放地址法？

开放地址法（Open Addressing）是另一种解决哈希冲突的方法。所有元素都存储在一个数组中， 当发生冲突时，通过探测序列在数组中寻找下一个空位，不使用额外的链表结构。

工作原理：

• 计算hash：对key计算哈希值，确定初始位置
• 探测空位：如果位置被占用且key不匹配，使用探测函数找下一个位置
• 线性探测：Agrona使用 index = (index + 1) & mask 线性探测下一个槽位
• 连续存储：keys数组存储原始long值，values数组存储对象引用，两个数组索引对应

关键优势：数据连续存储在数组中，CPU可以一次加载多个元素到缓存行，大幅提升访问速度

💡 开放地址法举例说明

场景：向一个容量为8的Long2ObjectHashMap插入三个订单ID：1001L、1009L、1002L

初始状态：容量=8，mask=7（用于快速取模：hash & 7 等价于 hash % 8）

步骤1：插入 1001L

• 计算哈希并混淆：hash = mix(1001L) = 0x...f9
• 定位索引：index = hash & 7 = 1
• 检查keys[1]：keys[1] == 0（空位）
• 直接存储：keys[1] = 1001L，values[1] = Order@1
• 无需探测，一次完成！

步骤2：插入 1009L

• 计算哈希并混淆：hash = mix(1009L) = 0x...01
• 定位索引：index = hash & 7 = 1
• 检查keys[1]：keys[1] == 1001L（已占用，冲突！）
• 线性探测：index = (1 + 1) & 7 = 2
• 检查keys[2]：keys[2] == 0（空位）
• 存储到index 2：keys[2] = 1009L，values[2] = Order@2
• 探测1次找到空位

步骤3：插入 1002L

• 计算哈希并混淆：hash = mix(1002L) = 0x...0a
• 定位索引：index = hash & 7 = 2
• 检查keys[2]：keys[2] == 1009L（已占用，冲突！）
• 线性探测：index = (2 + 1) & 7 = 3
• 检查keys[3]：keys[3] == 0（空位）
• 存储到index 3：keys[3] = 1002L，values[3] = Order@3
• 探测1次找到空位

最终结构：

keys数组 (long[]):         values数组 (Object[]):
[0]: 0 (空位)               [0]: null
[1]: 1001L                  [1]: Order@1  ← 直接命中
[2]: 1009L  (冲突后探测)    [2]: Order@2  ← 探测1次
[3]: 1002L  (冲突后探测)    [3]: Order@3  ← 探测1次
[4]: 0 (空位)               [4]: null
[5]: 0 (空位)               [5]: null
[6]: 0 (空位)               [6]: null
[7]: 0 (空位)               [7]: null

关键：所有数据在两个数组中连续存储，索引1、2、3在同一个CPU缓存行内！

查找过程（查找1009L）：

• 计算哈希：hash = mix(1009L)，定位到index 1
• 检查keys[1]：keys[1] == 1001L → 不匹配
• 线性探测：index = (1 + 1) & 7 = 2
• 检查keys[2]：keys[2] == 1009L → 匹配！返回values[2]
• 总共需要：2次数组访问（都在同一缓存行，延迟~4 CPU周期）

✅ 性能优势：数据连续存储，index 1、2、3会被一次性加载到CPU缓存行（64字节可容纳8个long）。 探测时无需跳转内存，延迟极低。无装箱，无链表，无指针追踪！

开放地址法的优势：

• 连续内存：keys和values各用一个数组，内存连续分配
• 缓存友好：顺序扫描时，CPU缓存行可以预加载后续数据
• 无指针追踪：不需要next指针，节省内存和CPU周期
• 线性探测：冲突时直接探测下一个位置，利用缓存局部性

缓存行利用率对比：

put操作详解

核心思想：通过hash计算初始位置，如果冲突则线性探测找到空位或相同key的位置。

关键步骤：

• ① hash计算：使用 mix(key) 函数对key进行混淆，得到更均匀的hash值
• ② 索引定位：index = hash & mask，mask = capacity - 1，等价于取模但更快
• ③ 空位检查：检查 keys[index] 是否为0（空位标记）
• ④ 直接存储：如果是空位，直接存储key和value，完成插入
• ⑤ key比较：如果位置已占用，检查是否是相同的key
• ⑥ 替换旧值：如果key相同，替换旧value，返回旧值
• ⑦ 线性探测：如果key不同（冲突），探测下一个位置 index = (index + 1) & mask
• ⑧ 循环探测：重复步骤③-⑦，直到找到空位或相同key

性能关键：由于keys数组连续存储，线性探测时CPU缓存命中率极高，平均只需1-2次探测。

get操作详解

核心思想：通过hash计算初始位置，如果key不匹配则线性探测，直到找到匹配的key或遇到空位。

关键步骤：

• ① hash计算：使用与put相同的 mix(key) 函数计算hash值
• ② 索引定位：index = hash & mask，定位到初始位置
• ③ key比较：检查 keys[index] 是否等于要查找的key
• ④ 找到返回：如果key匹配，返回 values[index]
• ⑤ 空位检查：如果key不匹配，检查是否为0（空位）
• ⑥ 未找到：如果遇到空位，说明key不存在，返回null
• ⑦ 线性探测：如果既不匹配也不为空，探测下一个位置 index = (index + 1) & mask
• ⑧ 循环探测：重复步骤③-⑦，直到找到key或遇到空位

关键逻辑：空位是查找终止的标志。因为put操作会占据所有中间位置，如果遇到空位，说明该key从未被插入过。

性能优势：由于数据连续存储，探测过程中的内存访问都在同一缓存行内，延迟极低（~4 CPU周期）。

1. Long2ObjectHashMap - long到对象的映射

// 订单ID（long）到订单对象的映射
Long2ObjectHashMap<Order> orderMap = new Long2ObjectHashMap<>();

// 添加订单（无装箱）
orderMap.put(1001L, order);

// 查询订单（无装箱）
Order found = orderMap.get(1002L);

// 高性能遍历（无装箱）
Long2ObjectHashMap.KeyIterator iter = orderMap.keySet().iterator();
while (iter.hasNext()) {
    long orderId = iter.nextLong();  // 无装箱！
    Order order = orderMap.get(orderId);
}

2. LongHashSet - long的HashSet

// 用户ID集合
LongHashSet userIds = new LongHashSet();

// 添加（无装箱）
userIds.add(123456L);

// 检查存在（无装箱）
boolean exists = userIds.contains(123456L);

3. Int2ObjectHashMap - int到对象的映射

// 商品ID（int）到商品对象的映射
Int2ObjectHashMap<Product> productMap = new Int2ObjectHashMap<>();

测试环境: JDK 11, 4核8GB, 预热10万次操作后测量

• 订单处理系统：订单ID到订单对象的映射
• 用户会话管理：用户ID到会话信息的映射
• 实时交易系统：交易ID到交易记录的映射
• 时间序列数据：时间戳到事件对象的映射
• 去重检测：使用LongHashSet存储已处理的ID
• 任何需要微秒级延迟的场景

堆内 vs 堆外内存对比

性能差异根源

• 位置管理开销：ByteBuffer需要维护position/limit，每次操作都要更新
• 边界检查开销：每次读写都要检查是否越界，无法JIT优化
• 字节序转换：需要判断和转换字节序（Big Endian / Little Endian）
• UnsafeBuffer优势：直接内存地址操作，一条CPU指令完成

代码示例：堆内 vs 堆外

// 1. 堆内缓冲区（byte数组）
byte[] byteArray = new byte[1024];
UnsafeBuffer heapBuffer = new UnsafeBuffer(byteArray);

// 写入数据（高性能）
heapBuffer.putInt(0, 42);           // offset=0, value=42
heapBuffer.putLong(4, 123456L);     // offset=4, value=123456L
heapBuffer.putStringUtf8(12, "Hi"); // offset=12, string

// 读取数据
int val = heapBuffer.getInt(0);
long id = heapBuffer.getLong(4);
String msg = heapBuffer.getStringUtf8(12);

// 2. 堆外缓冲区（DirectByteBuffer）- 零拷贝
ByteBuffer directBB = ByteBuffer.allocateDirect(1024);
UnsafeBuffer directBuffer = new UnsafeBuffer(directBB);

// 相同的API，但零拷贝发送到网络
directBuffer.putInt(0, 999);
channel.write(directBB);  // 零拷贝发送！

AtomicBuffer 代码示例

// AtomicBuffer 支持多种原子操作
AtomicBuffer buffer = new UnsafeBuffer(
    ByteBuffer.allocateDirect(1024)
);

// 1. Ordered写入（保证顺序，性能高）
buffer.putLongOrdered(0, 100L);

// 2. Volatile读写（保证可见性）
buffer.putLongVolatile(8, 200L);
long value = buffer.getLongVolatile(8);

// 3. CAS操作（无锁并发）
boolean success = buffer.compareAndSetLong(
    0,      // offset
    100L,   // expected
    200L    // update
);

// 4. getAndAdd原子操作（计数器）
long oldValue = buffer.getAndAddLong(16, 1L);

// 实际应用：序列号生成器
public class SequenceGenerator {
    private final AtomicBuffer buffer;
    private static final int OFFSET = 0;

    public long next() {
        // 原子递增，无需synchronized
        return buffer.getAndAddLong(OFFSET, 1L);
    }
}

ExpandableBuffer 使用场景

// 初始容量很小，自动扩展
ExpandableArrayBuffer buffer = new ExpandableArrayBuffer(64);

// 写入数据，容量不足时自动扩展
buffer.putInt(0, 42);        // OK，在容量内
buffer.putInt(1024, 999);    // 自动扩展到2048字节

// 适用场景：动态大小的消息
public void encodeMessage(Message msg, ExpandableArrayBuffer buffer) {
    int offset = 0;

    // 不用担心容量，自动扩展
    buffer.putInt(offset, msg.getType());
    offset += 4;

    buffer.putLong(offset, msg.getTimestamp());
    offset += 8;

    // 可变长度字符串
    offset += buffer.putStringUtf8(offset, msg.getContent());

    // 可变长度数组
    int[] data = msg.getData();
    buffer.putInt(offset, data.length);
    offset += 4;

    for (int value : data) {
        buffer.putInt(offset, value);
        offset += 4;
    }
}

零拷贝的优势

内存映射文件代码示例

// 1. 创建内存映射文件
RandomAccessFile raf = new RandomAccessFile("data.log", "rw");
FileChannel channel = raf.getChannel();

// 映射1GB文件到内存
MappedByteBuffer mappedBuffer = channel.map(
    FileChannel.MapMode.READ_WRITE,
    0,                  // position
    1024 * 1024 * 1024  // size: 1GB
);

// 2. 使用 UnsafeBuffer 包装
UnsafeBuffer buffer = new UnsafeBuffer(mappedBuffer);

// 3. 高性能读写（零拷贝）
buffer.putLong(0, System.currentTimeMillis());
buffer.putInt(8, messageId);
buffer.putStringUtf8(12, "Event data");

// 4. 读取数据（无需read系统调用）
long timestamp = buffer.getLong(0);
int id = buffer.getInt(8);
String data = buffer.getStringUtf8(12);

// OS会自动将修改刷盘，无需手动flush

测试环境: JDK 11, 4核8GB, 预热10万次操作后测量

• 网络通信：消息序列化/反序列化，零拷贝发送
• 进程间通信：共享内存IPC（MappedByteBuffer）
• 消息队列：环形缓冲区、消息存储
• 日志系统：内存映射文件持久化
• 交易系统：订单簿、市场数据存储
• 实时流处理：高吞吐量事件处理

BitUtil - 位操作工具

• 快速的对齐计算
• 2的幂次判断
• 位反转、计数

// 向上对齐到8字节
int aligned = BitUtil.align(123, 8);  // 128

// 判断是否是2的幂
boolean isPowerOfTwo = BitUtil.isPowerOfTwo(64);  // true

EpochClock / NanoClock - 时间工具

• 高性能时间戳获取
• 可模拟时间（便于测试）

// 毫秒级时间戳
long millis = epochClock.time();

// 纳秒级时间戳
long nanos = nanoClock.nanoTime();

DutyCycle - 工作周期追踪

• 追踪Agent的工作负载
• 计算工作时间占比

ManyToOneConcurrentArrayQueue - 多生产者单消费者队列

• 无锁设计
• 高吞吐量
• 适用于多线程向单线程传递消息

RingBuffer - 环形缓冲区

• 固定大小
• 零垃圾
• 支持批量操作