消息队列后端通信【免费下载链接】aeronEfficient reliable UDP unicast, UDP multicast, and IPC message transport项目地址https://gitcode.com/gh_mirrors/ae/aeron点击查看免费下载本指南以 aeron-samples/README.md 为骨架系统讲解 Aeron 官方示例与工具集的用法先介绍如何启动媒体驱动Media Driver再逐一演示 BasicSubscriber、BasicPublisher、RateSubscriber、StreamingPublisher 等消息收发样例接着深入 AeronStat、ErrorStat、LossStat、BacklogStat、LogInspector 等监控诊断工具最后覆盖 Archive 录播样例、嵌入式吞吐/延迟测试以及跨进程网络测试。读完本文你将掌握在真实环境中搭建 Aeron 收发链路、观测系统状态并做性能基准测试的完整实战能力。一、运行前置条件先启动媒体驱动Media DriverAeron 采用客户端 媒体驱动的进程模型客户端应用通过共享内存中的命令与控制CnC文件与媒体驱动通信实际的数据收发、重传、流量控制等工作全部由媒体驱动完成。因此运行任何样例之前必须先在独立的控制台中启动媒体驱动命令如下来自 aeron-samples/README.mdaeron-samples/scripts/media-driver optional properties file aeron-samples/scripts/low-latency-media-driver optional properties filemedia-driver脚本本质上等价于执行 run-java 启动io.aeron.driver.MediaDriver即标准配置的媒体驱动low-latency-media-driver会额外加载low-latency.properties配置见 low-latency-media-driver适合对时延敏感的场景。optional properties file可以是一个 Java properties 文件用于覆盖媒体驱动的各项参数如线程模式、MTU、socket 缓冲区等不传则使用默认配置。所有运行脚本都依赖 java-common 中定义的环境必须设置JAVA_HOME环境变量否则脚本会报错退出脚本从仓库根目录的 version.txt 读取版本号然后加载aeron-all/build/libs/aeron-all-${VERSION}.jar作为 classpath默认附带了面向高性能的 JVM 参数-XX:TrustFinalNonStaticFields、-XX:GuaranteedSafepointInterval300000降低安全点频率、-XX:UseParallelGC以及为最新 JDK 准备的--add-opens模块开关额外 JVM 参数可通过JVM_OPTS环境变量注入例如JVM_OPTS-Daeron.sample.channel... ./script。嵌入式媒体驱动模式从源码看多数样例类如 BasicPublisher.java 与 BasicSubscriber.java都读取SampleConfiguration.EMBEDDED_MEDIA_DRIVER对应-Daeron.sample.embeddedMediaDrivertrue系统属性。设置为true时应用会通过MediaDriver.launchEmbedded()在同一进程内启动媒体驱动并把aeronDirectoryName指向该驱动从而无需先启动独立媒体驱动进程。默认值为false即默认依赖外部媒体驱动。二、SampleConfiguration所有样例共享的配置中心所有样例的默认参数都定义在 SampleConfiguration.java 中并且全部支持通过 Java 系统属性覆盖。核心参数如下表系统属性默认值说明aeron.sample.channelaeron:udp?endpointlocalhost:20121默认样例使用的 UDP 通道 URIaeron.sample.streamId1001默认流 IDaeron.sample.ping.channelaeron:udp?endpointlocalhost:20123Ping 样例发送通道aeron.sample.pong.channelaeron:udp?endpointlocalhost:20124Pong 样例响应通道aeron.sample.ping.streamId1002Ping 流 IDaeron.sample.pong.streamId1003Pong 流 IDaeron.sample.messages10000000发送/测试的消息条数aeron.sample.messageLength32消息长度字节aeron.sample.warmup.messages10000预热消息数aeron.sample.warmup.iterations10预热迭代次数aeron.sample.frameCountLimit10每次 poll 最多处理的消息数aeron.sample.lingerTimeout0发送完成后的驻留时间毫秒aeron.sample.idleStrategyorg.agrona.concurrent.BusySpinIdleStrategy空闲策略类名aeron.sample.embeddedMediaDriverfalse是否内嵌媒体驱动aeron.sample.exclusive.publicationsfalse是否使用独占 Publicationaeron.sample.randomMessageLengthfalse是否随机消息长度例如要修改 BasicSubscriber 的通道与流 ID可以这样启动aeron-samples/scripts/basic-subscriber -Daeron.sample.channelaeron:udp?endpointlocalhost:5555 -Daeron.sample.streamId20注意JAVA_OPTS不会直接生效脚本只透传JVM_OPTS与命令行位置参数。若通过脚本传系统属性请使用JVM_OPTS环境变量例如JVM_OPTS-Daeron.sample.channelaeron:udp?endpointlocalhost:5555 -Daeron.sample.streamId20 ./aeron-samples/scripts/basic-subscriber三、入门收发样例BasicPublisher 与 BasicSubscriber这是官方文档列出的第一组样例用于快速验证 Aeron 端到端消息通路。BasicPublisherBasicPublisher.java 的职责向指定通道和流 ID 发布固定数量的消息每条消息之间暂停 1 秒发送完成后驻留LINGER_TIMEOUT_MS毫秒给可能发生丢失的订阅方留出通过 NAK 请求重传并恢复数据的时间。核心代码逻辑为通过Aeron.connect(ctx)连接媒体驱动aeron.addPublication(CHANNEL, STREAM_ID)创建发布端将Hello World! i写入UnsafeBuffer调用publication.offer(buffer, 0, length)发送根据返回值判断发送结果返回正值发送成功已推进到的位置Publication.BACK_PRESSURED背压导致 offer 失败Publication.NOT_CONNECTED尚无订阅方连接Publication.ADMIN_ACTION系统管理操作期间失败Publication.CLOSEDPublication 已关闭Publication.MAX_POSITION_EXCEEDED达到最大位置。每次 offer 后调用publication.isConnected()检测是否有活动订阅方。运行方式aeron-samples/scripts/basic-publisher对应脚本 basic-publisher 实际执行io.aeron.samples.BasicPublisher主类。BasicSubscriberBasicSubscriber.java 的职责订阅指定通道与流 ID将收到的每条消息或消息片段以 ASCII 形式打印出来。它注册了availableImageHandler与unavailableImageHandler回调来打印映像Image的可用/不可用事件然后通过SamplesUtil.subscriberLoop(fragmentHandler, FRAGMENT_COUNT_LIMIT, running)进入轮询循环每次 poll 最多处理FRAGMENT_COUNT_LIMIT默认 10条消息。运行方式aeron-samples/scripts/basic-subscriber对应脚本 basic-subscriber 实际执行io.aeron.samples.BasicSubscriber主类。该应用只处理非分片消息如需处理大消息分片重组README 与类注释都指向带FragmentAssembler的变体实现。体验步骤终端 Aaeron-samples/scripts/media-driver终端 Baeron-samples/scripts/basic-subscriber终端 Caeron-samples/scripts/basic-publisher观察终端 B 打印出 Hello World! 0、Hello World! 1……终端 C 打印 Offering i/N - yay!。四、速率观测样例RateSubscriber 与 StreamingPublisherRateSubscriber一个按速率打印消息接收情况的订阅方。它借助 ImageRateReporter 统计单位时间内接收的消息数与吞吐并通过ImageRateSubscriber主类运行适合观察持续流的接收速率。StreamingPublisher以尽可能快的方式流式发布消息并实时展示发布速率与 RateSubscriber 搭配可以直观地看到 Aeron 在 UDP 通道上的线速收发能力。两者对应的脚本为 rate-subscriber 与 streaming-publisher。此外仓库还提供streaming-exclusive-publisherC 语言实现与streaming-exclusive-publisher相关 Java 变体以及 C 语言的 rate_subscriber.c 与 streaming_publisher.c方便对照跨语言行为。五、Ping/Pong延迟测试工具Ping/Pong 是经典的往返延迟RTT测量模型PingPing 侧发送端。脚本 ping 执行io.aeron.samples.Ping默认参数为 100 万条消息、消息长度 32 字节并开启-Dagrona.disable.bounds.checkstrue关闭边界检查、-Daeron.pre.touch.mapped.memorytrue预触达映射内存、-Daeron.sample.exclusive.publicationstrue使用独占发布端PongPong 侧响应端。脚本 pong 执行io.aeron.samples.Pong将 Ping 发来的消息原样回送同样开启上述高性能选项。二者使用独立的通道localhost:20123/localhost:20124与流 ID1002 / 1003在 Ping 侧用 HdrHistogram 统计 RTT 分布。Java 实现位于io.aeron.samples.Ping/io.aeron.samples.PongC 语言对应物为 cping.c 与 cpong.c。运行方法先启动媒体驱动再分别启动pong与ping观察 Ping 侧输出的延迟直方图。六、性能基准测试嵌入式与跨进程变体官方文档说明仓库还附带了一组性能测试既可以在同一进程内运行方便起见无需媒体驱动也可以跨进程运行并且每种都提供了吞吐量与延迟测量的变体embedded测试倾向于在同一进程内运行。例如 EmbeddedThroughput.java吞吐量、EmbeddedPingPong.java延迟、EmbeddedIpcThroughput.java、EmbeddedExclusiveThroughput.java、EmbeddedExclusiveSpiedThroughput.java、EmbeddedBufferClaimIpcThroughput.java、EmbeddedExclusiveVectoredIpcThroughput.java 等media面向 IPC共享内存或 UDP网络的变体需要在独立媒体驱动下跨进程运行。以 embedded-throughput 脚本为例它执行io.aeron.samples.EmbeddedThroughput并预设了一组对吞吐友好的参数可以直接照抄到自己的基准测试中exec ${DIR}/run-java \ -Djava.net.preferIPv4Stacktrue \ -Dagrona.disable.bounds.checkstrue \ -Daeron.sample.messageLength32 \ -Daeron.sample.messages500000000 \ -Daeron.term.buffer.sparse.filefalse \ -Daeron.mtu.length8k \ -Daeron.socket.so_sndbuf2m \ -Daeron.socket.so_rcvbuf2m \ -Daeron.rcv.initial.window.length2m \ ${JVM_OPTS} io.aeron.samples.EmbeddedThroughput $这些参数的含义aeron.mtu.length8k提高 MTU 以减少报文数量aeron.socket.so_sndbuf/so_rcvbuf2m加大 socket 缓冲区aeron.rcv.initial.window.length2m扩大接收窗口以容忍更多在途数据aeron.term.buffer.sparse.filefalse关闭稀疏 term 文件。从 EmbeddedPingPong.java 的源码可以看到延迟测试的实现要点它使用ThreadingMode.DEDICATED专用线程模式conductor 用BackoffIdleStrategysender/receiver 用NoOpIdleStrategy纯忙等牺牲 CPU 换取最低延迟消息用UnsafeBuffer直接分配并缓存行对齐BitUtil.CACHE_LINE_LENGTH结果存入 HdrHistogram 的Histogram同时支持WARMUP_NUMBER_OF_ITERATIONS次预热迭代。C 侧对应实现见 PingPong.cpp、ExclusivePingPong.cpp、Throughput.cpp、ExclusiveThroughput.cppC 语言侧有 ping_pong_raw.c 等。七、监控与诊断工具AeronStat / ErrorStat / LossStat / BacklogStat / LogInspector官方文档明确列出以下监控与诊断工具工具脚本用途AeronStataeron-stat打印媒体驱动正在使用的计数器counter的标签与数值ErrorStaterror-stat打印媒体驱动观测到的去重后的错误distinct errorsLossStatloss-stat按流打印丢包报告BacklogStatbacklog-stat打印各流的流位置报告给出每个流的处理积压backlog指示LogInspectorlog-inspector诊断工具打印指定流的日志缓冲区log buffer内容用于调试AeronStat 详解AeronStat.java 的原理媒体驱动会在共享内存中维护一个命令与控制CnC文件布局见CncFileDescriptorAeronStat 读取该文件并通过CountersReader打印所有计数器默认每秒刷新一次并持续监听watch模式也支持命令行过滤参数java -cp aeron-samples/build/libs/samples.jar io.aeron.samples.AeronStat type[1-9] identity12345支持的过滤键包括type计数器类型、identity系统计数器 ID 或位置计数器的注册 ID、session、stream、channel。其中计数器类型划分0 为系统计数器1–5、9、10、11 为流位置与指示器如PUBLISHER_POS、SENDER_LIMIT、RECEIVER_POS、PER_IMAGE等6–7 为通道端点状态SEND_CHANNEL_STATUS、RECEIVE_CHANNEL_STATUS。这些类型常量定义在io.aeron.driver.status包中如PublisherPos.PUBLISHER_POS_TYPE_ID、PublisherLimit.PUBLISHER_LIMIT_TYPE_ID运行时可通过delay与watch参数控制刷新间隔与持续监听行为。LogInspector 与 LogInspectorCliLogInspector 用于离线检查某个流日志缓冲区的完整内容帧头、位置、数据对排查发布端/接收端位置不一致、帧损坏等问题很有帮助仓库还提供命令行交互版本 LogInspectorCli.java 与对应的 log-inspector-cli 脚本。C 语言侧对应工具为 driver_tool.c 与 aeron_stat.c可用 CMake 构建后运行。八、Aeron Archive 录播样例官方文档指出在 aeron-samples/scripts/archive/ 子目录及 Java 包中可以找到基于 Archive 的流录制与回放样例包括录播吞吐测试、录制发布/回放订阅、ReplayMerge 与持久订阅Persistent Subscription等。目录中同时提供了.cmdWindows与无后缀Unix两套脚本以及三个 properties 模板standard-archive.properties、high-throughput-archive.properties、lightweight-archive.properties。录播吞吐测试Embedded Throughput Samplesembedded-recording-throughput录制一批消息然后询问是否重复测试多次重复可让系统预热。文档建议调整写入同步级别以权衡持久化与性能aeron.archive.file.sync.level0普通写入 OS 页缓存由后台刷盘aeron.archive.file.sync.level1强制脏数据页落盘aeron.archive.file.sync.level2强制脏数据页与文件元数据落盘。 级别越高崩溃恢复时数据越安全但吞吐/时延开销越大。embedded-replay-throughput先录制一批消息再在新的流上回放同样支持重复回放以预热。文档建议尝试不同的消息长度与线程配置。这两个样例默认会把 Archive 目录建在临时文件系统上因此强烈建议通过 properties 文件显式指定aeron.archive.dir指向快速存储如 NVMe/内存盘。参考 standard-archive.properties 的内容aeron.archive.dir../../build/archive aeron.archive.threading.modeSHARED aeron.archive.file.sync.level0 aeron.archive.file.io.max.length1m aeron.spies.simulate.connectiontrue aeron.threading.modeSHARED aeron.term.buffer.sparse.filetrue录制发布 回放订阅的完整流程来自 archive/README.md 的标准操作步骤启动归档媒体驱动自己的控制台./archiving-media-driver config properties file启动带录制功能的发布端./recorded-basic-publisher config properties file启动普通订阅方使发布端连接上并开始录制cd .. ./basic-subscriber config properties file启动请求回放的订阅方消费已录制流./replay-basic-subscriber config properties file可选运行 AeronStat 观察状态重点关注录制流对应的rec-pos录制位置计数器./aeron-stat检查错误./error-stat注意archiving-media-driver、recorded-basic-publisher、replay-basic-subscriber等脚本位于aeron-samples/scripts/archive/目录而basic-subscriber、aeron-stat、error-stat位于上一级目录aeron-samples/scripts/执行时需按上述相对路径或使用绝对路径。ReplayMerge直播 回放的合并ReplayMerge 适用于 MDCMulti-Destination Cast或组播发布场景先将发布流录制到 Archive同时让订阅方在直播流与回放流之间无缝合并适合边录边看、断点续播的实时体验。典型步骤终端一./archiving-media-driver终端二用 MDC 动态控制通道发布并录制export JVM_OPTS-Daeron.sample.channelaeron:udp?controllocalhost:20550|control-modedynamic|aliasreplay-merge-sample ./recorded-basic-publisher终端三启动与发布通道匹配的普通订阅方export JVM_OPTS-Daeron.sample.channelaeron:udp?controllocalhost:20550 ../basic-subscriber稍等片刻再启动 ReplayMerge 订阅方aeron.sample.channel与发布通道保持一致export JVM_OPTS-Daeron.sample.channelaeron:udp?controllocalhost:20550 ./replay-merge-subscriber持久订阅Persistent Subscription持久订阅允许订阅方在中间离线一段时间后从 Archive 中恢复错过的消息实现不会丢消息的订阅。步骤同样可用JVM_OPTS传系统属性定制配置./archiving-media-driver./recorded-basic-publisher../basic-subscriber确保录制进行让消息发送一段时间后启动./persistent-subscriber它会在回放位置之后从录制流继续消费。其他 Archive 样例目录下还有recording-replicator录制副本复制用于跨节点/跨区域复制录制数据、segment-inspector检查 Archive 分段文件内容等脚本以及 archive 样例源码 与aeron-system-tests/src/test/java/io/aeron/archive下的系统测试可供深入学习。九、其他实用脚本与扩展阅读raw 目录send-receive-udp-ping/send-receive-udp-pong系列脚本演示在未使用 Aeron 的情况下直接用 UDP 收发报文配合 ping_pong_raw.c用于理解底层 socket 行为与对比基线cluster 目录提供basic-auction-cluster、basic-auction-client等集群样例启动脚本与 namespace 管理脚本配套教程见 Cluster-Tutorial.asciidocresponse 目录response_client.c/response_server.c与 Java 的ResponseClient/ResponseServer演示了 Aeron 的请求-响应Response Channels模式echo / stress / security 包分别提供 echo 回显、压力测试与安全样例FileSender / FileReceiver演示通过 Aeron IPC/UDP 传输文件的场景C 样例BasicPublisher.cpp、BasicSubscriber.cpp、Ping.cpp、Pong.cpp、RateSubscriber.cpp、StreamingPublisher.cpp等位于 aeron-samples/src/main/cpp可用 CMake 构建见 CMakeLists.txt 与 aeron-samples/src/main/c/CMakeLists.txt与 Java 样例行为一一对应。十、小结Aeron Samples 是一个覆盖面完整的官方演练场从media-driver启动、BasicPublisher/BasicSubscriber入门到Ping/Pong延迟测试、embedded-*吞吐基准再到AeronStat/ErrorStat/LossStat/BacklogStat/LogInspector监控诊断以及 Archive 录制/回放/持久订阅/ReplayMerge 的高级场景均可在 aeron-samples 模块内开箱即用地复现。所有样例共享SampleConfiguration的默认参数与系统属性覆盖机制配合JVM_OPTS注入参数你可以快速在自己的环境下调整通道、流 ID、消息量与消息长度把官方样例改造成属于自己的压测与观测工具。赞分享消息队列后端通信【免费下载链接】aeronEfficient reliable UDP unicast, UDP multicast, and IPC message transport项目地址https://gitcode.com/gh_mirrors/ae/aeron点击查看免费下载相关推荐Garnet 性能基准测试实战Resp.benchmark 吞吐与延迟测试工具完全指南Garnet 性能基准测试实战Resp.benchmark 吞吐与延迟测试工具完全指南 Garnet 项目内置了专用的 RESP 基准测试工具 Resp.be缓存KV存储后端超全面Temporal性能测试指南从吞吐量到延迟优化实战超全面Temporal性能测试指南从吞吐量到延迟优化实战 想要确保你的Temporal工作流服务在生产环境中稳定运行性能测试是关键环节这篇完整指南将带你深后端工作流自动化任务调度从延迟到吞吐量Centrifugo双协议实战指南——WebSocket与GRPC深度测评从延迟到吞吐量Centrifugo双协议实战指南——WebSocket与GRPC深度测评 Centrifugo是一款开源的可扩展实时消息服务器能够通过Web消息队列后端通信上一篇3分钟掌握Mermaid用代码绘制专业图表的高效工具下一篇Android应用隐私合规检查UltimateAndroidReference中的完整工具指南创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考