缓存KV存储消息队列流处理后端【免费下载链接】hazelcastHazelcast is a unified real-time data platform combining stream processing with a fast data store, allowing customers to act instantly on>项目地址https://gitcode.com/gh_mirrors/ha/hazelcast点击查看免费下载导读本文基于 Hazelcast 仓库中的设计文档 docs/design/sql/11-engine-merge.md 展开系统梳理 Hazelcast SQL 历史上 IMDG 与 Jet 两套执行引擎并存的成因、能力差距、底层架构差异Tasklet/Exec、网络、队列、背压、负载均衡、内存管理并给出引擎合并的决策依据与实施路径。读完本文你将理解 Hazelcast SQL 最终统一到 Jet 执行引擎的技术逻辑、两套引擎各自的核心机制含源码级佐证以及轻量任务light job与计划缓存等关键优化点如何落地。背景为什么会有两套 SQL 执行引擎Hazelcast 历史上存在两个彼此独立的执行引擎IMDG 引擎位于 IMDG 模块内聚焦小规模、短平快的查询特点是查询往返roundtrip开销低、有计划缓存、支持动态参数。Jet 引擎位于 Jet 模块内聚焦大规模批处理与流式查询特点是支持流处理、持久化任务、故障容错与丰富的外部连接器。两者采用非常相似的、基于 DAG 的执行模型但在大量细节上各不相同。双引擎并存的成因主要有二IMDG 与 Jet 曾是两款独立产品IMDG 需要一套不依赖 Jet 也能运行的引擎两套引擎的目标场景不同——IMDG 面向小查询Jet 面向大批次与流式查询。然而双引擎并行的代价是明显的维护成本翻倍。例如为查询增加LIMIT支持两套引擎各自都要实现一遍尽管设计思路几乎相同同时还会产生功能缺口——某些特性只在一个引擎中可用。合并之前查询的解析与执行流程是先由 IMDG 引擎解析并尝试执行如果 IMDG 引擎无法运行再交给 Jet 引擎。这种两段式的尝试模式既增加链路开销也让行为不一致的风险扩散到每条查询。两套引擎的功能清单对比设计文档按仅在 IMDG 实现仅在 Jet 实现两引擎共有三个维度梳理了功能分布这直接度量了若移除某一引擎所需补齐的工作量。仅在 IMDG 引擎实现的功能IMap 的高效读取可利用索引LIMIT/OFFSETORDER BY当时仅能借助有序索引实现快速的查询往返fast query roundtrip计划缓存plan cache动态参数查询中的?占位符。仅在 Jet 引擎实现的功能非 IMap 连接器Kafka、CSV/JSON/Avro本地或远程文件含 Hadoop、S3、Amazon、GCP迁移容忍migration-tolerant的 IMap 读取流式支持结果无界的流式查询聚合函数与GROUP BY且当时无需内存管理IMDG 侧为 WIP出于内存顾虑只允许按索引列分组数据生成器generate_series、generate_stream富化连接enrichment join可将 IMap 与除 IMap 之外的任何对象连接IMap 中的 JSON 序列化CREATE MAPPING同时也让 IMDG 能操作空映射持久化任务CREATE JOB任务管理suspend、resume、restart故障容错exactly-once、at-least-once与自动扩缩算子并行度operator parallelism任务指标job metricsManagement Center 中的任务可视化information_schema、SHOW MAPPINGS、SHOW JOBSINSERT INTO与SINK INTO连接器IMap、Kafka、文件对丢包具有韧性网络丢包可被检测并报告失败。两引擎共有无需双份实现API 层包括 JDBC 以及任意 HZ 客户端中的 SQL API支持的数据类型简单算子与函数即计划中仅作为投影projection存在的算子如、ABS、SIN但不包括SUM、TUMBLE这类算子上述函数的类型强制转换例如TINYINT REAL的结果类型或INT VARCHAR应报的错误方言dialect引号风格、大小写敏感性、字面量语法等。从清单可以直观看出Jet 引擎的覆盖面明显更广尤其在容错、流处理与连接器生态方面。这也是最终决策补齐 Jet、移除 IMDG的重要依据。引擎架构对比Worker 模型Jet 的基本执行单元Tasklet / ProcessorJet 中的基本执行单元是Tasklet即 Processor均为单线程。它们运行在两类线程上协作线程cooperative threads数量固定每个协作线程可运行任意数量的协作 tasklet。协作 tasklet 有非常严格的要求显著增加了开发难度——不允许阻塞且单次调用不能耗时过长。但其优势在于可充当轻量级线程切换开销极低。专用线程dedicated threads用于使用阻塞 API 的 tasklet尤其是各类连接器。IMDG 的基本执行单元ExecIMDG 引擎中基本执行单元是Exec被包装在QueryFragmentExecutable中由调度器提交到ForkJoinPoolFJP执行。当 Exec 完成或需要更多数据时它会终止本次调度当下游节点通知有新数据时再被重新调度。当时并没有限制单次调度执行时长的机制但文档指出若存在大量待执行 fragment可以很容易地加入时长上限以改善 CPU 共享。值得注意的是从当前仓库源码来看IMDG 侧这套Exec/QueryFragmentExecutable执行机制已不再出现在hazelcast与hazelcast-sql模块中SQL 的统一执行实现集中在hazelcast-sql/src/main/java/com/hazelcast/jet/sql/impl/包下如 PlanExecutor.java、SqlPlanImpl.java这从源码结构上印证了文档中合并到 Jet 引擎的结论已经落地。引擎架构对比网络与消息传递JetSenderTasklet / ReceiverTaskletJet 中处理器把条目写入队列队列分为本地队列与远程队列远程队列连接着SenderTasklet它负责序列化条目并把一批条目合并为一个 packet直接发送到Connection对象由成员的网络线程处理在目标成员上网络线程把 packet 放入ReceiverTasklet的队列典型实现为MPSCQueue即多生产者单消费者队列ReceiverTasklet反序列化条目再写入目标处理器的输入队列。发送与接收 tasklet 都是协作式的。为避免网络上的丢包或重排Jet 使用底层Connection对象一个通常不会引发集群拓扑变化的连接抖动也会导致查询失败但如果任务是容错的则可以从最近的快照重启并继续运行。IMDGFragment / Sender-Receiver 边IMDG 中查询计划被拆分为一个或多个查询 fragmentfragment 之间通过带唯一 ID 的边edge相连每个 fragment 有一个或多个输入且严格只有一个输出输入可以是具体的数据源如IMap也可以是接收算子receiver operator输出要么是发送算子sender operator要么是用户游标user cursor从 DAG 视角看fragment 是顶点一对 sender/receiver 算子构成一条边。IMDG 同样使用底层Connection对象发送消息绕开了 invocation 子系统。它对 packet 重排有韧性但对 packet 丢失目前没有韧性——丢包可能导致查询卡住或消耗大量内存文档认为该问题可以修复。引擎架构对比队列与背压队列Jet 使用单消费者、单生产者的无等待wait-free并发队列。这是协作式处理的硬性要求——其他队列类型可能因等待锁而阻塞。背压机制两引擎本质一致两套引擎的背压思想高度相似发送方持有预算budget接收方周期性更新预算。Jet 本地背压处理器之间使用固定容量队列放不进去就退避back offJet 跨成员背压采用类似 TCPRWIN的做法——接收方定期告知自己处理了多少条目发送方不得发送超过该数量所有任务的 rwin 值被打包在一个紧凑结构中、随单个 packet 一起发送IMDG 背压一对 sender/receiver 事先约定初始字节数即信用额度 credits。每发出一个消息sender 就扣减可用信用信用归零时 sender 停止发送。Receiver 周期性地回复一条特殊的flow_control消息来增加 sender 的信用额度。引擎架构对比负载均衡与内存管理负载均衡Jet 不做负载均衡分区倾斜partition skew可能造成单个查询内资源利用不均tasklet 到线程的分配是确定性的、轮询式的round-robin因此两条查询会确定性地被分配到不同线程但在特定情况下可能出现重要 tasklet 都堆在一个线程、其余线程闲置的失衡大小不同的两个任务小查询 vs 大查询交错运行表现良好文档认为出问题的场景罕见。协作线程模型还容易放大实现不符合协作规则的缺陷好在 SQL 所需的绝大多数 processor 已在 Core Jet 中实现。IMDG 引擎则直接利用ForkJoinPool的负载均衡能力不受上述问题困扰文档指出若再加入单次处理条目数上限的机制也不会引入负载不均问题。内存管理截至文档撰写时两套引擎均未实现内存管理但两引擎的内存管理设计在概念上将高度相似——这也为合并降低了额外的架构成本。引擎架构对比空闲 CPU 负载与 Windows 性能空闲 CPU 负载问题JetJet 协作线程以无限循环运行被分配的 tasklet若所有 tasklet 都无进展线程会短暂休眠但只要众多 tasklet 中有一个取得了进展下一轮迭代就会重新调用所有tasklet。这导致即使没有任务运行也存在 CPU 开销通常每个核 1%但当存在成百上千个空闲流式任务时会显著升高另一个表现是轻流式负载下 CPU 使用率偏高——文档给出的经验值是最大吞吐的 20% 时 CPU 占用约 80%批处理任务不受影响因为它们跑在最大吞吐。缓解手段增大退避超时以流式任务延迟升高为代价超时过大还会降低批处理任务的吞吐实现更智能的 tasklet 调度文档坦承该方向尚无研究。IMDG 引擎因为有新数据才向 FJP 重新调度任务不存在此问题。Windows 上的低性能问题Jet协作线程的退避使用LockSupport.parkNanos()而在 Windows 上它的最小休眠时间约为15ms某些条件下为 1ms即便代码只请求 50µs。这大幅降低了吞吐与延迟。因此文档明确Jet 引擎完全不支持将 Windows 用于生产环境。这一条也是推动合并后统一引擎形态必须正视的部署约束。性能对比同一基准下的双引擎实测文档在当前版本性能对比一节给出了一个专门设计的基准测试数据规模100 万条1M条目公平性处理Jet 本地并行度设为 1因为 IMDG 引擎没有算子并行度分区数设为 2Jet 将 map 读取卸载到分区线程能利用比 IMDG 当前实现更多的并行度运行环境两台 AWSc5.xlarge成员节点。实测结果文档记录集群/引擎类型耗时IMDG847msJet290ms文档同时给出重要免责声明该基准不直接度量两引擎的纯处理性能因为两引擎读取 IMap 的方式差异很大但它足以表明两引擎处于同一量级same arena。基准测试完整代码文档原文保留备查import com.hazelcast.client.HazelcastClient; import com.hazelcast.client.config.ClientConfig; import com.hazelcast.core.HazelcastInstance; import com.hazelcast.jet.Jet; import com.hazelcast.jet.JetInstance; import com.hazelcast.jet.core.DAG; import com.hazelcast.jet.core.Vertex; import com.hazelcast.jet.core.processor.Processors; import com.hazelcast.jet.core.processor.SourceProcessors; import com.hazelcast.jet.function.RunnableEx; import com.hazelcast.jet.pipeline.Pipeline; import com.hazelcast.jet.pipeline.Sinks; import com.hazelcast.jet.pipeline.Sources; import com.hazelcast.map.IMap; import java.util.HashMap; import java.util.Map; import static com.hazelcast.jet.core.Edge.between; import static java.util.concurrent.TimeUnit.NANOSECONDS; public class JetImdgBenchmark { private static int numItems; private static int jetParallelism; private static int warmUpIterations; private static int measuredIterations; private static HazelcastInstance hzInst; public static void main(String[] args) { if (args.length ! 5) { System.err.println(Usage:); System.err.println( JetImdgBenchmark jet|imdg numItems jetParallelism warmUpIterations measuredIterations); System.exit(1); } boolean isJet jet.equalsIgnoreCase(args[0]); numItems Integer.parseInt(args[1]); jetParallelism Integer.parseInt(args[2]); warmUpIterations Integer.parseInt(args[3]); measuredIterations Integer.parseInt(args[4]); if (isJet) { testJet(); } else { testImdg(); } } public static void testJet() { JetInstance inst Jet.newJetClient(); try { hzInst inst.getHazelcastInstance(); prepareData(); DAG dag new DAG(); Vertex src dag.newVertex(src, SourceProcessors.readMapP(m)) .localParallelism(jetParallelism); Vertex sink dag.newVertex(sink, Processors.noopP()); dag.edge(between(src, sink).distributeTo(hzInst.getCluster().getMembers().iterator().next().getAddress())); Pipeline p Pipeline.create(); p.readFrom(Sources.map(m)) .writeTo(Sinks.noop()); runTest(() - inst.newJob(p).join()); } finally { inst.shutdown(); } } public static void testImdg() { ClientConfig config new ClientConfig() .setClusterName(jet); hzInst HazelcastClient.newHazelcastClient(config); try { prepareData(); runTest(() - hzInst.getSql().execute(select * from m).iterator().forEachRemaining(r - { })); } finally { hzInst.shutdown(); } } private static void runTest(RunnableEx test) { for (int i 0; i warmUpIterations; i) { System.out.println(warmup i); long start System.nanoTime(); test.run(); long elapsed System.nanoTime() - start; System.out.println(### warmup iteration i took NANOSECONDS.toMillis(elapsed)); } long totalTime 0; for (int i 0; i measuredIterations; i) { System.out.println(real i); long start System.nanoTime(); test.run(); long elapsed System.nanoTime() - start; totalTime elapsed; System.out.println(### real iteration i took NANOSECONDS.toMillis(elapsed)); } System.out.println(### avg time NANOSECONDS.toMillis(totalTime / measuredIterations)); } private static void prepareData() { IMapLong, Long imap hzInst.getMap(m); MapLong, Long tmpMap new HashMap(); int batchSize 100_000; for (long i 0; i numItems; ) { for (int j 0; j batchSize; j, i) { tmpMap.put(i, i); } imap.putAll(tmpMap); System.out.println(i); tmpMap.clear(); } if (imap.size() ! numItems) { throw new AssertionError(); } } }命令行用法代码内嵌提示JetImdgBenchmark jet|imdg numItems jetParallelism warmUpIterations measuredIterations第一个参数选jet或imdg分别走 DAG/Pipeline 路径与hzInst.getSql().execute(...)路径numItems指定数据量文档基准为 1_000_000jetParallelism指定 Jet 本地并行度基准中取 1 以对齐 IMDG 无算子并行度的现状warmUpIterations与measuredIterations分别控制预热与正式测量轮次。短查询性能合并前必须补齐的短板Jet 引擎历史上聚焦大批次与流处理任务初始化时间不是关注点。要让极短任务如只返回 1 行或 100 行的查询的性能追平 IMDG文档列出了四项必须完成的工作完成轻量任务light job原型把init、execute、complete三个操作合并为一个。这并不简单——会引入各种新的竞态races但可行可以只对非容错任务启用。增加 light job 模式不把任务元数据持久化到任何 IMap不使用JetClassLoader协调器除了向成员转发工作外不需要任何远程调用。允许任意成员协调任务而不只限于 master 成员——这样从非 master 成员提交的查询就不必多一跳转发到 master。加入计划缓存与动态参数支持。从当前仓库源码看这些规划大部分已落地JobCoordinationService中已存在submitLightJob、joinLightJob、terminateLightJob、getLightJobIds等完整方法见 JobCoordinationService.java计划缓存也已实现为独立组件 PlanCache.java并实现了CachedPlanInvalidationCallback接口用于缓存失效回调。文档预计这些改动落地后短查询性能将与 IMDG 相当。结论与决策保留 Jet移除 IMDG文档的最终结论可以概括为三点两引擎高度相似功能上都能支撑全部必需特性只是在不同场景下特性表现各异没有不可逾越的障碍未发现任何固有特性会阻止把全部功能实现到某一个引擎上决策方向明确由于 Jet 引擎当前功能更多尤其是故障容错、成熟度更高提案是向 Jet 引擎补齐缺失功能并移除 IMDG 引擎。从当前仓库的模块结构看这一决策已经执行hazelcast-sql模块中 SQL 的解析、优化与执行实现全部位于com.hazelcast.jet.sql.impl包下例如 CalciteSqlOptimizerImpl.java 以 Calcite 为优化器基础PlanExecutor.java 统一执行SelectPlan、DmlPlan、CreateMappingPlan、CreateJobPlan、IMapSelectPlan等各类计划而 IMDG 侧独立的Exec/QueryFragmentExecutable执行链已不在源码中出现。文档中列举的 Jet 侧能力CREATE MAPPING、SHOW MAPPINGS、information_schema、持久化任务、INSERT INTO/SINK INTO等如今都在 PlanExecutor.java 中对应的方法execute(CreateMappingPlan...)、execute(ShowStatementPlan...)、execute(CreateJobPlan...)、execute(DmlPlan...)等中得到统一承载。延伸阅读本文所依据的设计文档为 docs/design/sql/11-engine-merge.md它是docs/design/sql/系列设计文档中的一环。该系列从不同侧面完整刻画了 Hazelcast SQL 的架构演进可与本文互相印证01-type-system.md类型系统设计02-operator-interface.md算子接口03-network-protocol.mdSQL 网络协议04-parallel-execution.md并行执行模型。若要继续深入代码建议从以下入口出发SqlServiceImpl.javaSQL 服务入口CalciteSqlOptimizerImpl.java基于 Calcite 的 SQL 优化器实现SqlPlanImpl.java各类计划Select/Dml/CreateJob/CreateMapping 等的抽象PlanExecutor.java统一执行入口JobCoordinationService.javalight job 的提交、join、终止与查询逻辑PlanCache.javaSQL 计划缓存实现。理解引擎合并这条主线是读懂 Hazelcast SQL 从双引擎试跑走向统一 Jet 执行引擎的关键——它决定了今天你在 Hazelcast 中执行 SQL 时背后运行的是同一套具备流处理、容错与丰富连接器能力的执行基础设施。赞分享缓存KV存储消息队列流处理后端【免费下载链接】hazelcastHazelcast is a unified real-time data platform combining stream processing with a fast data store, allowing customers to act instantly on>项目地址https://gitcode.com/gh_mirrors/ha/hazelcast点击查看免费下载相关推荐Hazelcast Mustang SQL 引擎并行执行环境设计解析从 SEDA 线程池到 ForkJoinPool 的演进Hazelcast Mustang SQL 引擎并行执行环境设计解析从 SEDA 线程池到 ForkJoinPool 的演进 导读 Hazelcast 的 S缓存KV存储消息队列流处理后端Faster-Whisper 本地语音识别比 Whisper 快 4 倍5 分钟完成部署Faster Whisper 本地语音识别比 Whisper 快 4 倍5 分钟完成部署 OpenAI 原版 whisper 是语音识别的头部选择但转录一人工智能语音音频本地部署secretsdump.py 使用教程Windows密码哈希提取工具详解secretsdump.py 使用教程Windows密码哈希提取工具详解 项目概述 secretsdump.py 是 Impacket 库中 secretsdAI Agent强化学习大模型模型评测上一篇如何选择AtomicQueue队列类型从AtomicQueue到OptimistAtomicQueue的完整比较指南 下一篇ServerPackCreator5分钟从模组包到服务器包的魔法转换器创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考