Aeron C API 开发指南：构建依赖、配置方式与发布订阅编程实战

消息队列后端通信【免费下载链接】aeronEfficient reliable UDP unicast, UDP multicast, and IPC message transport项目地址https://gitcode.com/gh_mirrors/ae/aeron点击查看免费下载本指南以 Aeron 仓库中 aeron-client/src/main/c/README.md 为骨架系统讲解 Aeron C 客户端库的构建产物与平台依赖、环境变量与程序化两种配置途径并结合仓库内的头文件与示例程序给出基于libaeron编写 UDP 单播 / IPC 发布订阅程序的完整路径。读完本文你将掌握aeron_context_t/aeron_t/aeron_publication_t/aeron_subscription_t的核心用法、AERON_*系列环境变量的语义以及从构建到运行调试的全流程。一、C API 概览单一头文件承载的完整接口Aeron 的 C API 源码位于仓库 aeron-client/src/main/c 目录。与 Java API 分散的类文档不同C API 的全部接口声明与用法文档集中在一个头文件aeronc.h 中约 2800 行包含了aeron_context_t、aeron_t等客户端核心结构aeron_publication_t/aeron_exclusive_publication_t发布端接口offer、try_claim、offerv等aeron_subscription_t/aeron_image_t订阅端接口poll、controlled_poll、block_poll等aeron_fragment_assembler_t等消息重组工具计数器Counters、版本查询、错误处理、时钟、CnC 文件访问等辅助 API。因此编写 C 客户端时只需#include aeronc.h即可获得完整能力。C API 与 C 驱动Media Driver的系统级测试位于 aeron_c_system_test.cpp可作为端到端行为验证的参考。C API 的更多使用样例分布在 aeron-samples/src/main/c 目录下。二、构建产物与平台依赖2.1 构建产物位置C 客户端通过 CMake 构建。构建过程会生成客户端动态库并放置在以下位置${CMAKE_CURRENT_BINARY_DIR}/lib/libaeron.so Linux ${CMAKE_CURRENT_BINARY_DIR}/lib/libaeron.dylibmacOS构建配置见 aeron-client/src/main/c/CMakeLists.txt。该文件定义了动态库目标aeron与静态库目标aeron_static对应别名aeron::aeron与aeron::aeron_static并声明头文件安装目录include/aeron。若设置AERON_INSTALL_TARGETSCMake 会将库安装到lib、头文件安装到include/aeron方便通过find_package集成。2.2 Linux 依赖Linux 上构建和运行需要以下系统库由target_link_libraries显式链接C Library随构建系统自带-lpthreadpthread 线程库客户端 Conductor 线程与并发原语依赖它-ldl动态链接加载库用于在运行时按名称加载 Idle Strategy 等策略这也是静态库链接选项-rdynamic存在的原因见 CMakeLists.txt 注释 Because dlsym() is used to load strategies.-lm数学库。此外 CMake 会自动探测若干可选能力并开启对应编译宏libbsd提供arc4random时定义HAVE_ARC4RANDOM否则告警提示安装 libbsd/dev/urandom定义HAVE_DEV_URANDOMfallocate/posix_fallocate/F_PREALLOCATE定义HAVE_FALLOCATE等用于日志文件的快速预分配posix_memalign、reallocf等。在 aarch64 架构上还会额外链接atomic库。发布构建会追加-DDISABLE_BOUNDS_CHECKS以消除边界检查开销。2.3 Windows 依赖Windows 平台要求Windows 版本Vista 或更高编译器MSVC v141Visual Studio 2017或更高。CMake 在 MSVC 下启用CMAKE_WINDOWS_EXPORT_ALL_SYMBOLS与BUILD_SHARED_LIBS并链接wsock32、ws2_32、Iphlpapi、Psapi等 Winsock 与系统库同时探测WSAPOLL原型以定义HAVE_WSAPOLL。三、配置方式程序化设置与环境变量C API 的配置有两条途径二者等价互补程序化配置通过aeron_context_set_*系列函数在代码中设置aeron_context_t字段环境变量配置通过AERON_*环境变量在进程启动前设置。3.1 环境变量与 Java 属性的对应关系C API 的环境变量直接对应 Java API 的aeron.*系统属性规则为将属性名中的.替换为_并大写。例如设置环境变量AERON_DIR等价于 Java API 设置aeron.dir设置AERON_DRIVER_TIMEOUT等价于 Java 的aeron.driver.timeout。各环境变量宏在 aeronc.h 中定义环境变量的实际读取逻辑位于 aeron_context.c 的aeron_context_init()中第 121 行起依次getenv解析。常用变量汇总如下环境变量宏定义aeronc.h对应 Java 属性语义AERON_DIRAERON_DIR_ENV_VARaeron.dirMedia Driver 与客户端通信所用的顶层 Aeron 目录CnC 文件所在目录AERON_DRIVER_TIMEOUTAERON_DRIVER_TIMEOUT_ENV_VARaeron.driver.timeout驱动超时毫秒默认 10000AERON_CLIENT_NAMEAERON_CLIENT_NAME_ENV_VARaeron.client.name客户端名称长度上限 100 字符AERON_COUNTER_MAX_CLIENT_NAME_LENGTHAERON_CLIENT_RESOURCE_LINGER_DURATIONAERON_CLIENT_RESOURCE_LINGER_DURATION_ENV_VARaeron.client.resource.linger.duration资源驻留时长纳秒默认 3 秒防止资源释放过早AERON_CLIENT_IDLE_SLEEP_DURATIONAERON_CLIENT_IDLE_SLEEP_DURATION_ENV_VARaeron.client.idle.sleep.duration空闲休眠时长纳秒默认 16ms也是默认sleep-ns策略的初始参数AERON_CLIENT_PRE_TOUCH_MAPPED_MEMORYAERON_CLIENT_PRE_TOUCH_MAPPED_MEMORY_ENV_VARaeron.client.pre.touch.mapped.memory是否预触碰映射内存bool默认 falseAERON_AGENT_ON_START_FUNCTIONAERON_AGENT_ON_START_FUNCTION_ENV_VAR—每个 Agent 启动时调用的函数名3.2 程序化配置 API与上述环境变量一一对应aeronc.h提供了aeron_context_set_*/aeron_context_get_*成对函数int aeron_context_set_dir(aeron_context_t *context, const char *value); const char *aeron_context_get_dir(aeron_context_t *context); int aeron_context_set_driver_timeout_ms(aeron_context_t *context, uint64_t value); uint64_t aeron_context_get_driver_timeout_ms(aeron_context_t *context); int aeron_context_set_keepalive_interval_ns(aeron_context_t *context, uint64_t value); int aeron_context_set_resource_linger_duration_ns(aeron_context_t *context, uint64_t value); int aeron_context_set_idle_sleep_duration_ns(aeron_context_t *context, uint64_t value); int aeron_context_set_idle_strategy(aeron_context_t *context, const char *value); int aeron_context_set_pre_touch_mapped_memory(aeron_context_t *context, bool value); int aeron_context_set_client_name(aeron_context_t *context, const char *value); int aeron_context_set_error_handler(aeron_context_t *context, aeron_error_handler_t handler, void *clientd);需要注意aeron_context_init()会优先读取环境变量覆盖默认值之后代码中再调用aeron_context_set_*则可在进程内做最终覆盖。除环境变量外还可用aeron_properties_buffer_load()、aeron_properties_file_load()、aeron_properties_http_load()从字符串、属性文件或 HTTP URL 加载namevalue配置并写入进程环境。3.3 默认值与参数校验从 aeron_context.c 第 34-39 行的默认值宏可以看到客户端侧关键默认值#define AERON_CONTEXT_DRIVER_TIMEOUT_MS_DEFAULT (10 * 1000L) // 10 秒 #define AERON_CONTEXT_KEEPALIVE_INTERVAL_NS_DEFAULT (500 * 1000 * 1000LL) // 500ms #define AERON_CONTEXT_RESOURCE_LINGER_DURATION_NS_DEFAULT (3 * 1000 * 1000 * 1000LL) // 3 秒 #define AERON_CONTEXT_IDLE_SLEEP_DURATION_NS_DEFAULT (16 * 1000 * 1000LL) // 16ms #define AERON_CONTEXT_PRE_TOUCH_MAPPED_MEMORY_DEFAULT (false)环境变量解析失败如AERON_DRIVER_TIMEOUT不是合法整数、AERON_CLIENT_RESOURCE_LINGER_DURATION无法按纳秒时长解析时aeron_context_init()会返回-1并可通过aeron_errcode()/aeron_errmsg()获取错误详情。客户端名称超过AERON_COUNTER_MAX_CLIENT_NAME_LENGTH100也会被拒绝。四、最小可运行示例发布 / 订阅生命周期仓库中的 basic_publisher.c 与 basic_subscriber.c 是最直观的入门样例完整展示了客户端标准生命周期。以下流程基于这两个文件提炼。4.1 客户端初始化三件套无论发布还是订阅都必须依次完成aeron_context_t *context NULL; aeron_t *aeron NULL; // 1. 创建并初始化 context此处可 set_dir / set_* 定制也可依赖 AERON_* 环境变量 if (aeron_context_init(context) 0) { /* 处理错误aeron_errmsg() 查看详情 */ } // 2. 基于 context 创建客户端。注意context 将被客户端独占不可复用于其他客户端 if (aeron_init(aeron, context) 0) { /* 处理错误 */ } // 3. 启动客户端可能派生 Client Conductor 线程 if (aeron_start(aeron) 0) { /* 处理错误 */ }若希望以“应用线程驱动 Conductor”的方式运行避免额外线程可先aeron_context_set_use_conductor_agent_invoker(context, true)随后在业务循环中调用aeron_main_do_work(aeron)与aeron_main_idle_strategy(aeron, work_count)。4.2 异步添加资源轮询驱动的非阻塞模型Aeron C API 采用async poll模式创建发布端与订阅端调用aeron_async_add_*发起请求后反复调用对应的_poll函数直到返回 1完成。basic_publisher 中的典型写法aeron_async_add_publication_t *async NULL; aeron_publication_t *publication NULL; if (aeron_async_add_publication(async, aeron, channel, stream_id) 0) { /* 错误 */ } while (NULL publication) { if (aeron_async_add_publication_poll(publication, async) 0) { /* 错误 */ } sched_yield(); }其中channel为通道 URI 字符串stream_id为流 ID。以 IPC 为例aeron.ipc以 UDP 单播为例aeron:udp?endpointlocalhost:40123。订阅端对应aeron_async_add_subscription(...)/aeron_async_add_subscription_poll(...)注册订阅时需同时提供 image 可用/不可用回调if (aeron_async_add_subscription( async, aeron, channel, stream_id, on_available_image_handler, /* 收到 image 时回调 */ NULL, on_unavailable_image_handler, /* image 消失时回调 */ NULL) 0) { /* 错误 */ }同一模式还适用于aeron_async_add_exclusive_publication、aeron_async_add_counter、aeron_async_add_static_counter以及aeron_async_remove_*资源移除。发起中的操作若不再需要可调用对应的aeron_async_add_*_cancel取消。4.3 发布消息offer 与 try_claim发布端核心是aeron_publication_offer()Exclusive 发布端为aeron_exclusive_publication_offerint64_t result aeron_publication_offer( publication, (const uint8_t *)message, message_len, NULL, NULL); if (result 0) { /* 成功result 为新的流位置 */ } else if (AERON_PUBLICATION_BACK_PRESSURED result) { /* 订阅者背压稍后重试 */ } else if (AERON_PUBLICATION_NOT_CONNECTED result) { /* 尚无连接的订阅者瞬时状态订阅者随时可能加入 */ } else if (AERON_PUBLICATION_ADMIN_ACTION result) { /* 因日志轮转等管理操作失败下次重试大概率成功 */ } else if (AERON_PUBLICATION_CLOSED result) { /* 发布端已关闭 */ }负返回值对应的语义常量在 aeronc.h 中定义为AERON_PUBLICATION_NOT_CONNECTED (-1)、AERON_PUBLICATION_BACK_PRESSURED (-2)、AERON_PUBLICATION_ADMIN_ACTION (-3)、AERON_PUBLICATION_CLOSED (-4)、AERON_PUBLICATION_MAX_POSITION_EXCEEDED (-5)、AERON_PUBLICATION_ERROR (-6)。其中MAX_POSITION_EXCEEDED表示达到 term buffer 长度 × 2^31 字节的流位置上限此时应关闭发布端并新建或增大 term buffer 长度。需要零拷贝写入时使用aeron_publication_try_claim()aeron_buffer_claim_commit()aeron_buffer_claim_t buffer_claim; if (aeron_publication_try_claim(publication, length, buffer_claim) 0L) { /* 直接向 buffer_claim.data 写入消息内容 */ aeron_buffer_claim_commit(buffer_claim); }注意 claim 长度不能超过最大负载长度max_payload_length即 MTU 减帧头且若 claim 持有超过aeron.publication.unblock.timeout驱动会认定发布线程已死并强制解除 claim见 aeronc.h 中aeron_publication_try_claim的注释。4.4 接收消息poll fragment assembler订阅端默认按**片段fragment**交付小于 MTU 的消息是一个完整片段大于 MTU 的消息会以多个片段到达。若应用需要始终收到完整消息应使用aeron_fragment_assembler_t组装。basic_subscriber 的做法是aeron_fragment_assembler_t *fragment_assembler NULL; /* 创建组装器组装完成的消息交给 poll_handler */ if (aeron_fragment_assembler_create(fragment_assembler, poll_handler, subscription) 0) { /* 错误 */ } while (is_running()) { int fragments_read aeron_subscription_poll( subscription, aeron_fragment_assembler_handler, fragment_assembler, DEFAULT_FRAGMENT_COUNT_LIMIT); if (fragments_read 0) { /* 错误 */ } aeron_idle_strategy_sleeping_idle((void *)idle_duration_ns, fragments_read); }片段处理器签名如下typedef void (*aeron_fragment_handler_t)( void *clientd, const uint8_t *buffer, size_t length, aeron_header_t *header);处理器中可通过aeron_header_values()读取帧头字段session_id、stream_id、term_id等通过aeron_header_position()获取读取进度。对需要精细控制流位置的应用可改用aeron_subscription_controlled_poll()处理器返回AERON_ACTION_ABORT、AERON_ACTION_BREAK、AERON_ACTION_COMMIT、AERON_ACTION_CONTINUE四种动作之一。4.5 资源清理程序结束按逆序释放aeron_publication_close(publication, NULL, NULL); /* 或 aeron_subscription_close(...) */ aeron_close(aeron); aeron_context_close(context); aeron_fragment_assembler_delete(fragment_assembler);aeron_*_close均支持传入可选的aeron_notification_t完成回调因关闭动作可能发生在独立线程。客户端若因驱动超时被自动关闭可用aeron_is_closed(aeron)判断注意必须在aeron_close之前调用。五、进阶从 API 到底层实现的源码对照客户端与驱动的通信aeron_context_init()在 aeron_context.c 中会分配命令缓冲并初始化 MPSC 环形缓冲aeron_mpsc_rb_init客户端通过 CnCCommand and Control文件与 Media Driver 交换命令。CnC 文件相关结构aeron_cnc_t、aeron_cnc_constants_t及aeron_cnc_init()等函数同样定义在 aeronc.h 末尾可直接读取驱动的计数器、错误日志与丢包报告。环境变量工具跨平台读写环境变量由 util/aeron_env.c 提供Windows 用_putenv_sPOSIX 用setenv/unsetenv。错误处理线程本地错误码与错误消息通过aeron_errcode()/aeron_errmsg()获取全局打印可通过AERON_FPRINTF宏与aeron_set_fprintf_handler()重定向。版本信息aeron_version_full()、aeron_version_major()等函数返回当前构建的版本与 Git SHA。协议层参考帧头与通道 URI 解析分别位于 protocol/aeron_udp_protocol.h 与 uri/aeron_uri.h。六、构建与运行建议用 CMake 构建客户端库与样例样例由 aeron-samples/src/main/c/CMakeLists.txt 管理确认libaeron.so/libaeron.dylib生成位置符合预期运行前先启动 Media DriverC 驱动入口见 aeronmd.c或使用aeron-samples/scripts/media-driver脚本若驱动使用非默认目录通过AERON_DIR环境变量或在代码中aeron_context_set_dir()指定同一目录确保客户端与驱动对齐程序启动后可通过aeron_is_driver_active()校验驱动存活或通过aeron_cnc_*系列接口直接观测驱动状态。需要注意的是C API 的配置命名、默认值与底层行为均以当前仓库源码为准若与旧版本或 Java API 文档存在差异应以 aeronc.h 与 aeron_context.c 中的实际实现为最终依据。赞分享消息队列后端通信【免费下载链接】aeronEfficient reliable UDP unicast, UDP multicast, and IPC message transport项目地址https://gitcode.com/gh_mirrors/ae/aeron点击查看免费下载相关推荐TDengine 数据订阅编程接口实战TMQ 消费者 API、配置参数与多语言开发指南TDengine 数据订阅编程接口实战TMQ 消费者 API、配置参数与多语言开发指南 导读 本文以 TDengine 开源时序数据库的数据订阅TMQTD数据库时序数据库大数据物联网云原生FrankenPHP 内置 Mercure Hub 实时推送完整指南配置、订阅与发布实战FrankenPHP 内置 Mercure Hub 实时推送完整指南配置、订阅与发布实战 本文档围绕 FrankenPHP 项目内置的 Mercure htt后端node-redis Pub/Sub 实战指南订阅、发布、退订与 Buffer 模式详解node redis Pub/Sub 实战指南订阅、发布、退订与 Buffer 模式详解 node redis 的 Pub/Sub发布/订阅API 是构建后端数据库客户端缓存上一篇Buzz终极隐私保护的离线语音转文字工具彻底告别云端依赖下一篇Ubisoft La Forge Animation Dataset性能测试基准模型Zero-Velocity与Interpolation对比创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考