消息隊列流處理后端微服務(wù)消息路由【免費下載鏈接】pulsarApache Pulsar - distributed pub-sub messaging system項目地址https://gitcode.com/gh_mirrors/pu/pulsar點擊查看免費下載Apache Pulsar 的命名空間主題列舉topic listing邏輯在 PIP-452 中被重新設(shè)計從硬編碼掃描元數(shù)據(jù)存儲演進為支持客戶端屬性上下文properties、可插拔擴展PulsarResourcesExtended的靈活架構(gòu)。本文將以 PIP-452 提案為主線結(jié)合當(dāng)前倉庫中的協(xié)議定義、Broker 源碼與端到端測試完整講解其設(shè)計動機、協(xié)議變更、配置方式、自定義實現(xiàn)方法及向后兼容與安全邊界幫助讀者在 Pulsar 上落地多租戶場景下按屬性篩選主題的能力。背景與動機硬編碼主題列舉的局限在 PIP-452 之前Broker 中處理CommandGetTopicsOfNamespace的邏輯是硬編碼的直接掃描元數(shù)據(jù)存儲如 ZooKeeper下命名空間節(jié)點的全部子節(jié)點。這在復(fù)雜多租戶場景下暴露了兩大問題沒有客戶端上下文No Client ContextBroker 無法區(qū)分是誰、因為什么目的在請求主題列表也就無法基于客戶端屬性這些屬性可能對應(yīng)或派生于主題屬性對結(jié)果進行過濾過濾效率低下Inefficient Filtering對于包含數(shù)百萬主題的命名空間Broker 必須先把完整主題列表加載進內(nèi)存再應(yīng)用topics_pattern正則做過濾。沒有任何機制可以把過濾下推push down到數(shù)據(jù)源側(cè)例如帶索引的數(shù)據(jù)庫。PIP-452 的解決思路是雙管齊下讓協(xié)議攜帶客戶端屬性同時把主題列舉邏輯改造成可插拔的擴展點。目標(biāo)范圍做什么與不做什么In Scope范圍內(nèi)協(xié)議為CommandGetTopicsOfNamespace增加properties字段承載客戶端側(cè)上下文Broker引入可插拔接口PulsarResourcesExtended允許自定義資源管理邏輯先從 Get Topics 請求做起Client更新 Java 客戶端在使用Regex 訂閱時把 Consumer 屬性轉(zhuǎn)發(fā)給 lookup 服務(wù)Admin API CLIREST API 與命令行支持在列舉命名空間主題時傳入propertiesConfiguration新增 Broker 配置項用于開關(guān)主題列表監(jiān)聽topic list watching。Out of Scope范圍外不支持為自定義主題列舉開啟主題列表監(jiān)聽topic list watching可作為后續(xù)工作考慮若想使用該特性必須關(guān)閉主題列表監(jiān)聽器即在broker.conf中設(shè)置enableBrokerTopicListWatcher false。這一點與測試代碼相互印證在 CustomizedPulsarResourcesExtendedTest.java 的setup()中測試顯式調(diào)用conf.setEnableBrokerTopicListWatcher(false)后才配置自定義擴展類?？傮w設(shè)計協(xié)議攜帶上下文 Broker 可插拔資源接口高層設(shè)計上PIP-452 做了兩處修改Pulsar 協(xié)議新增propertiesmap 字段Broker 側(cè)新增可插拔接口PulsarResourcesExtended其默認(rèn)實現(xiàn)保留原有行為委托給NamespaceService。請求的處理鏈路為Broker 連接處理器connection handler→NamespaceService→PulsarResourcesExtended。NamespaceService作為入口把帶屬性的列舉請求最終導(dǎo)向可插拔擴展從而為自定義主題列舉策略留出實現(xiàn)空間。協(xié)議變更PulsarApi.proto 新增 properties 字段協(xié)議定義位于 PulsarApi.proto源碼中的CommandGetTopicsOfNamespace已完整包含 PIP-452 新增的字段message CommandGetTopicsOfNamespace { enum Mode { PERSISTENT 0; NON_PERSISTENT 1; ALL 2; } required uint64 request_id 1; required string namespace 2; optional Mode mode 3 [default PERSISTENT]; optional string topics_pattern 4; optional string topics_hash 5; // Context properties from the client repeated KeyValue properties 6; }字段說明字段編號類型說明request_id1required uint64請求標(biāo)識用于關(guān)聯(lián)響應(yīng)namespace2required string目標(biāo)命名空間mode3optional Mode列舉模式默認(rèn)PERSISTENT可選NON_PERSISTENT、ALLtopics_pattern4optional string客戶端下發(fā)的正則過濾模式既有字段topics_hash5optional string用于增量同步的主題哈希既有字段properties6repeated KeyValue新增字段客戶端屬性上下文用于自定義過濾對應(yīng)的響應(yīng)消息CommandGetTopicsOfNamespaceResponsePulsarApi.proto保留了topics、filtered、topics_hash、changed等既有語義字段Broker 最終過濾邏輯不受影響。Broker 配置兩個關(guān)鍵配置項PIP-452 在 Broker 配置中引入兩個參數(shù)其定義與默認(rèn)值見 ServiceConfiguration.java# Enables watching topic add/remove events on broker side. It is separated from enableBrokerSideSubscriptionPatternEvaluation. # 是否在 Broker 側(cè)監(jiān)聽主題新增/刪除事件用于訂閱模式評估默認(rèn) true enableBrokerTopicListWatcher true # Class name for the extended Pulsar resources. # 擴展資源實現(xiàn)類名該類必須實現(xiàn) org.apache.pulsar.broker.PulsarResourcesExtended # 默認(rèn)實現(xiàn)為 DefaultPulsarResourcesExtended保留原有行為 pulsarResourcesExtendedClassName org.apache.pulsar.broker.DefaultPulsarResourcesExtended源碼層面的細(xì)節(jié)值得注意pulsarResourcesExtendedClassName的默認(rèn)值常量定義在 ServiceConfiguration.java即org.apache.pulsar.broker.DefaultPulsarResourcesExtendedenableBrokerTopicListWatcher默認(rèn)值為trueServiceConfiguration.java且屬于dynamic false的靜態(tài)配置修改需要重啟 Broker配置項文檔明確說明舊的enableBrokerSideSubscriptionPatternEvaluation不再控制主題列表監(jiān)聽行為已遷移到enableBrokerTopicListWatcher見 ServiceConfiguration.java。默認(rèn)配置文件 broker.conf 中仍保留enableBrokerSideSubscriptionPatternEvaluationtrue。配置注意事項當(dāng)前倉庫的默認(rèn) broker.conf 尚未包含這兩個新配置項它們依靠ServiceConfiguration的默認(rèn)值生效。若需要自定義主題列舉應(yīng)在broker.conf中顯式添加pulsarResourcesExtendedClassName并同時關(guān)閉主題列表監(jiān)聽enableBrokerTopicListWatcher false pulsarResourcesExtendedClassName com.example.MyCustomPulsarResourcesExtended可插拔接口PulsarResourcesExtended 與默認(rèn)實現(xiàn)接口定義接口源碼位于 PulsarResourcesExtended.java標(biāo)注為InterfaceStability.Evolving共定義三個方法package org.apache.pulsar.broker; InterfaceStability.Evolving public interface PulsarResourcesExtended { CompletableFutureListString listTopicOfNamespace(NamespaceName namespaceName, CommandGetTopicsOfNamespace.Mode mode, Nullable MapString, String properties); void initialize(PulsarService pulsarService); void close(); }listTopicOfNamespace(...)核心方法按命名空間 模式 屬性過濾返回主題名列表properties為 null 或空時不施加過濾initialize(PulsarService)Broker 啟動時調(diào)用為自定義實現(xiàn)注入PulsarService及其依賴close()Broker 關(guān)閉時調(diào)用用于釋放自定義實現(xiàn)持有的資源。默認(rèn)實現(xiàn)委托 NamespaceServiceDefaultPulsarResourcesExtended.java 是默認(rèn)實現(xiàn)其listTopicOfNamespace直接委托給pulsarService.getNamespaceService().getListOfTopics(namespaceName, mode)完全保留 PIP 之前的原生行為public class DefaultPulsarResourcesExtended implements PulsarResourcesExtended { Override public CompletableFutureListString listTopicOfNamespace(NamespaceName namespaceName, CommandGetTopicsOfNamespace.Mode mode, MapString, String properties) { return pulsarService.getNamespaceService().getListOfTopics(namespaceName, mode); } // initialize() 中持有 pulsarServiceclose() 為空實現(xiàn) }加載機制擴展實例在 Broker 啟動階段創(chuàng)建。從 PulsarService.java 可以看到啟動流程中調(diào)用loadPulsarResourcesExtended()其實現(xiàn)PulsarService.java通過Reflections.createInstance(className, PulsarResourcesExtended.class, ...)按類名反射實例化并調(diào)用initialize(this)——因此自定義類必須具備無參構(gòu)造且pulsarResourcesExtendedClassName必須指向?qū)崿F(xiàn)了該接口的類。NamespaceService 改造帶屬性過濾的主題列表入口NamespaceService是請求處理與可插拔擴展之間的橋接層源碼位于 NamespaceService.java。新增方法public CompletableFutureListString getListOfTopicsByProperties(NamespaceName namespaceName, Mode mode, MapString, String properties) { return pulsar.getPulsarResourcesExtended().listTopicOfNamespace(namespaceName, mode, properties); } public CompletableFutureListString getListOfUserTopicsByProperties(NamespaceName namespaceName, Mode mode, MapString, String properties) { return getListOfUserTopicsInternal(cacheKeyWithProperties(namespaceName, mode, properties), () - getListOfTopicsByProperties(namespaceName, mode, properties)); }getListOfTopicsByProperties把請求下放給可插拔擴展是自定義策略的鉤子getListOfUserTopicsByProperties在擴展之上疊加既有邏輯——按緩存鍵合并并發(fā)請求并對結(jié)果執(zhí)行系統(tǒng)主題過濾。請求合并與緩存鍵getListOfUserTopicsInternalNamespaceService.java利用inProgressQueryUserTopics的computeIfAbsent合并同一鍵的并發(fā)查詢只讓第一個線程真正發(fā)起底層查詢并通過thenApplyAsync(TopicList::filterSystemTopic, pulsar.getExecutor())剔除系統(tǒng)主題。緩存鍵由cacheKeyWithPropertiesNamespaceService.java生成先拼接mode :// namespaceName若 properties 非空則按鍵排序后以|keyvalue追加——排序保證了等價屬性集合產(chǎn)生相同緩存鍵避免緩存碎片化。連接處理器調(diào)用更新Broker 側(cè)處理CommandGetTopicsOfNamespace時從原來的getListOfUserTopics改為getListOfUserTopicsByPropertiesPIP-452 中internalHandleGetTopicsOfNamespace片段return getBrokerService().pulsar().getNamespaceService() .getListOfUserTopicsByProperties(namespaceName, mode, properties);同時該路徑仍受maxTopicListInFlightLimiterAsyncDualMemoryLimiter的堆內(nèi)存配額限流保護說明帶屬性的列舉與普通列舉一樣受 Broker 的 in-flight 主題列表內(nèi)存限制約束?？蛻舳俗兏黂egex 訂閱轉(zhuǎn)發(fā) Consumer 屬性LookupService 接口Java 客戶端內(nèi)部LookupService接口更新簽名接受屬性 mapCompletableFutureListString getTopicsUnderNamespace( NamespaceName namespace, Mode mode, String topicsPattern, String topicsHash, MapString, String properties );PatternMultiTopicsConsumerImpl正則訂閱消費者實現(xiàn)PatternMultiTopicsConsumerImpl會從ConsumerConfigurationData中提取屬性并傳給LookupServiceConsumerConfigurationData conf; MapString, String contextProperties conf.getProperties(); lookup.getTopicsUnderNamespace( namespace, mode, topicsPattern.pattern(), topicsHash, contextProperties // Pass consumer properties here ).thenAccept(topics - { // ... update subscriptions ... });這意味著只要你在創(chuàng)建正則訂閱 Consumer 時設(shè)置了.properties(...)這些屬性就會隨getTopicsUnderNamespace請求帶到 Broker供自定義PulsarResourcesExtended實現(xiàn)做屬性匹配——客戶端屬性對應(yīng)或派生于主題屬性的假設(shè)由此閉環(huán)。Admin API 與 CLI按屬性列舉主題PIP-452 為 REST API 和pulsar-admin增加了properties參數(shù)注意 URL 中的值需要編碼GET /admin/v2/persistent/{tenant}/{namespace}?propertiesk1v1,k2v2pulsar-admin topics list tenant/namespace -p k1v1 -p k2v2測試代碼 CustomizedPulsarResourcesExtendedTest.java 展示了 Admin API 側(cè)的用法——通過ListTopicsOptions.builder().properties(propsFromClient).build()傳入屬性過濾條件ListString list admin.topics().getList(public/default, TopicDomain.persistent, ListTopicsOptions.builder().properties(propsFromClient).build());端到端驗證從自定義實現(xiàn)到測試用例測試用例解讀CustomizedPulsarResourcesExtendedTest.java 完整演示了該特性的落地路徑在setup()中配置setEnableBrokerTopicListWatcher(false)和setPulsarResourcesExtendedClassName(CustomizedPulsarResourcesExtended.class.getName())L44-L51創(chuàng)建分區(qū)主題并給若干主題打上自定義屬性setCustomProperties或通過admin.topics().updateProperties(...)更新真實主題屬性L92-L96通過 Admin API 傳入properties {env:prod, region:us-west}列舉主題斷言只返回帶這些屬性的主題含分區(qū)-partition-0/1/2用.topicsPattern(persistent://public/default/.*).properties(propsFromClient)創(chuàng)建 Regex Consumer斷言其實際訂閱的主題集合與屬性過濾結(jié)果一致L115-L124。該測試同時覆蓋了Admin API 路徑與Regex Consumer 路徑且在同一命名空間混有無屬性主題test-topic4的屬性為envtest以驗證過濾確實生效。自定義實現(xiàn)參考測試配套的自定義實現(xiàn) CustomizedPulsarResourcesExtended.java 是一個極佳的模板它繼承DefaultPulsarResourcesExtended僅覆蓋listTopicOfNamespaceOverride public CompletableFutureListString listTopicOfNamespace(NamespaceName namespaceName, CommandGetTopicsOfNamespace.Mode mode, Nullable MapString, String properties) { if (MapUtils.isEmpty(properties)) { return super.listTopicOfNamespace(namespaceName, mode, properties); } if (enabledTopicWatcher) { return CompletableFuture.failedFuture(new IllegalStateException( Customized topic listing with properties is not supported when broker topic watcher is enabled.)); } ListString list queryTopicListByProperties(namespaceName.toString(), properties); // ... 返回過濾結(jié)果 }要點無屬性請求直接走默認(rèn)邏輯委托原生NamespaceService只有攜帶屬性時才走自定義查詢自定義查詢時若開啟了 topic watcher 會顯式報錯與 PIP 的 Out of Scope 聲明一致自定義實現(xiàn)內(nèi)部可以用MapNamespace, MapTopic, MapPropertyKey, PropertyValue這樣的內(nèi)存映射模擬數(shù)據(jù)庫索引生產(chǎn)環(huán)境中完全可以把這里的查詢替換為對帶索引存儲如數(shù)據(jù)庫的訪問實現(xiàn)真正的過濾下推。向后兼容與安全考量向后兼容協(xié)議層properties是 Protobuf 的 optional/repeated 字段舊客戶端不會發(fā)送該字段舊 Broker 會忽略它屬于非破壞性變更行為層默認(rèn)策略與現(xiàn)有行為完全一致DefaultPulsarResourcesExtended委托原生NamespaceService且Broker 始終保留最終過濾邏輯——即使自定義策略返回了主題Broker 仍會應(yīng)用客戶端請求的topics_pattern正則確保自定義策略不可能返回違反客戶端模式的主題。安全考量輸入校驗propertiesmap 是用戶可控輸入。PIP 明確要求自定義實現(xiàn)必須校驗與清洗這些輸入尤其當(dāng)它們被用于構(gòu)造數(shù)據(jù)庫查詢時需防范注入風(fēng)險授權(quán)邊界本 PIP 只控制主題的發(fā)現(xiàn)列舉并不會繞過訂閱/生產(chǎn)這些主題時的 Authorization Service——對主題的讀寫權(quán)限校驗依然由既有的授權(quán)體系負(fù)責(zé)。小結(jié)PIP-452 以協(xié)議攜帶屬性 Broker 可插拔擴展的組合拳把 Apache Pulsar 命名空間主題列舉從硬編碼掃描升級為可定制、可下推過濾的能力同時通過默認(rèn)實現(xiàn)與 Broker 最終過濾保證 100% 向后兼容。對開發(fā)者而言落地路徑清晰關(guān)閉enableBrokerTopicListWatcher→ 實現(xiàn)并注冊PulsarResourcesExtended→ 在 Regex Consumer 或 Admin API 中攜帶屬性即可。相關(guān)源碼與測試均可在倉庫中查閱協(xié)議定義見 PulsarApi.proto接口與默認(rèn)實現(xiàn)見 PulsarResourcesExtended.java 與 DefaultPulsarResourcesExtended.java入口邏輯見 NamespaceService.java端到端示例見 CustomizedPulsarResourcesExtendedTest.java。贊分享消息隊列流處理后端微服務(wù)消息路由【免費下載鏈接】pulsarApache Pulsar - distributed pub-sub messaging system項目地址https://gitcode.com/gh_mirrors/pu/pulsar點擊查看免費下載相關(guān)推薦Apache Pulsar 可插拔 Entry Filter 機制解析基于 PIP-105 的 Dispatcher 消息過濾框架Apache Pulsar 可插拔 Entry Filter 機制解析基于 PIP 105 的 Dispatcher 消息過濾框架 導(dǎo)讀 PIP 105Su消息隊列流處理后端微服務(wù)消息路由Apache Pulsar 命名空間復(fù)制職責(zé)拆分PIP-321 與 allowed-clusters 機制深度解析Apache Pulsar 命名空間復(fù)制職責(zé)拆分PIP 321 與 allowed clusters 機制深度解析 PIP 321Split the res消息隊列后端Apache Pulsar PIP-191 深度解析基于屬性分組的批量消息 Entry Filter 過濾方案Apache Pulsar PIP 191 深度解析基于屬性分組的批量消息 Entry Filter 過濾方案 導(dǎo)讀 PIP 191 是 Apache Pul消息隊列流處理后端微服務(wù)消息路由上一篇淘寶淘金幣自動化腳本每天節(jié)省20分鐘的終極解決方案下一篇LMDeploy 部署 CogVLM / CogVLM2 多模態(tài)模型模型準(zhǔn)備、離線推理與 PyTorch 引擎實現(xiàn)剖析創(chuàng)作聲明：本文部分內(nèi)容由AI輔助生成（AIGC），僅供參考