擴(kuò)展實(shí)戰(zhàn)指南：Awesome 第三方庫清單、日志與可觀測性適配器及自定義 Pub/Sub)
Watermill 生態(tài)擴(kuò)展實(shí)戰(zhàn)指南Awesome 第三方庫清單、日志與可觀測性適配器及自定義 Pub/Sub【免費(fèi)下載鏈接】watermillBuilding event-driven applications the easy way in Go.項(xiàng)目地址: https://gitcode.com/GitHub_Trending/wa/watermillWatermill 的核心設(shè)計(jì)是官方內(nèi)核 可插拔適配器框架只定義消息模型與路由器而具體的消息中間件Pub/Sub、日志、追蹤全部通過接口接入。官方將社區(qū)中未經(jīng)其維護(hù)的優(yōu)質(zhì)第三方庫集中收錄于 Awesome Watermill 清單即本倉庫docs/content/docs/awesome.md按示例項(xiàng)目、Pub/Sub 適配器、日志適配器、可觀測性與其他工具分類。閱讀本文后你將掌握該清單的完整構(gòu)成、第三方庫與 Watermill 的接入接口原理并能依據(jù)倉庫內(nèi)的通用測試套件評(píng)估一個(gè)第三方適配器的成熟度甚至獨(dú)立實(shí)現(xiàn)并貢獻(xiàn)你自己的適配器。官方與第三方生態(tài)的邊界Awesome 清單的定位Awesome 清單的第一段文字就劃清了責(zé)任邊界下列庫并非由 Three Dots Labs 維護(hù)官方無法提供支持、也不保證它們一定工作正常使用前需要自行調(diào)研Do your own research。這正是它被稱為 Selected unofficial libraries 的原因——官方收錄它們只是因?yàn)槟憧赡軙?huì)覺得有用而不是背書。該清單本身是一個(gè)可維護(hù)的文檔文件如果你知道其他值得收錄的庫或者你本人就是某個(gè)適配器的作者可以通過編輯 docs/content/docs/awesome.md 把它加入清單。整個(gè)清單分為五類Examples社區(qū)編寫的完整示例項(xiàng)目Pub/Subs針對(duì)各類消息中間件的第三方適配器Logging將 Watermill 日志接入 logrus / zap / zerolog 的適配器Observability接入 OpenCensus / OpenTelemetry 的追蹤與指標(biāo)適配器Other代碼生成、工具集等周邊設(shè)施。清單末尾還特別引導(dǎo)讀者如果想了解如何實(shí)現(xiàn)自己的 Pub/Sub 適配器參見 Implementing custom Pub/Sub。理解接入點(diǎn)第三方庫如何與 Watermill 銜接要評(píng)估任何一個(gè)第三方庫先要理解 Watermill 為它們定義的插槽。核心接入點(diǎn)全部在 message/pubsub.go 中任何消息中間件適配器本質(zhì)上都是這兩個(gè)接口的實(shí)現(xiàn)// Publisher is the emitting part of a Pub/Sub. type Publisher interface { // Publish publishes provided messages to the given topic. // Publish can be synchronous or asynchronous - it depends on the implementation. // Most publisher implementations dont support atomic publishing of messages. // Publish does not work with a single Context. Use the Context() method of each message instead. // Publish must be thread safe. Publish(topic string, messages ...*Message) error // Close should flush unsent messages if publisher is async. Close() error } // Subscriber is the consuming part of the Pub/Sub. type Subscriber interface { // Subscribe returns an output channel with messages from the provided topic. // The channel is closed after Close() is called on the subscriber. // To receive the next message, Ack() must be called on the received message. // If message processing fails and the message should be redelivered Nack() should be called instead. Subscribe(ctx context.Context, topic string) (-chan *Message, error) // Close closes all subscriptions with their output channels and flushes offsets etc. when needed. Close() error }從接口注釋可以提煉出第三方適配器必須遵守的關(guān)鍵契約Publish必須線程安全且不保證原子性——批量發(fā)布多條消息時(shí)一旦某條失敗后續(xù)消息不會(huì)繼續(xù)發(fā)布見 docs/content/docs/pub-sub.md 的 Publishing multiple messages 一節(jié)同步/異步由實(shí)現(xiàn)決定異步發(fā)布器必須在Close()時(shí)沖刷未發(fā)送的消息忘記關(guān)閉發(fā)布器可能丟失消息Ack/Nack 是 Subscriber 的職責(zé)正確實(shí)現(xiàn)應(yīng)在收到上一條消息的Ack/Nack后再消費(fèi)下一條且必須等 Watermill 處理完消息、返回 Ack 之后才向底層 broker 提交 offset否則進(jìn)程在消息處理完成前崩潰會(huì)丟消息另有可選的 SubscribeInitializer 接口含SubscribeInitialize(topic string) error用于在消費(fèi)前初始化訂閱非強(qiáng)制實(shí)現(xiàn)。對(duì)日志類第三方庫而言接入點(diǎn)則是 log.go 中定義的LoggerAdapter接口type LoggerAdapter interface { Error(msg string, err error, fields LogFields) Info(msg string, fields LogFields) Debug(msg string, fields LogFields) Trace(msg string, fields LogFields) With(fields LogFields) LoggerAdapter }Watermill 自身提供了寫入標(biāo)準(zhǔn)輸出的 StdLoggerAdapterNewStdLogger(debug, trace bool)與靜默的NopLogger第三方日志適配器的工作就是把LoggerAdapter的方法調(diào)用橋接到 logrus、zap、zerolog 各自的 API 上。逐類盤點(diǎn) Awesome 清單中的第三方庫以下內(nèi)容完整覆蓋 docs/content/docs/awesome.md 中收錄的每一類庫。需要說明的是清單原文以外部倉庫鏈接形式給出此處按名稱與功能分類整理便于按圖索驥。示例項(xiàng)目Examples清單收錄了三個(gè)社區(qū)示例用于學(xué)習(xí)如何在真實(shí)項(xiàng)目中組合使用 Watermillgolang-taipei-watermill-exampleWatermill 的完整示例項(xiàng)目Kafka-PubSub以 Kafka 為消息中間件的示例go-example-financing一個(gè)金融場景融資的 Go 示例。這些項(xiàng)目與倉庫內(nèi)官方維護(hù)的_examples/目錄互補(bǔ)——官方示例覆蓋了 基礎(chǔ)應(yīng)用、路由器、CQRS、指標(biāo) 以及各類 Pub/Sub 接入示例社區(qū)示例則展示更多真實(shí)業(yè)務(wù)組合。Pub/Sub 適配器第三方這是清單中體量最大的一類。每個(gè)適配器都實(shí)現(xiàn)了上文所述的message.Publisher/message.Subscriber接口把 Watermill 統(tǒng)一的消息模型映射到特定中間件。清單收錄了以下第三方適配器目標(biāo)中間件適配器定位AMQP 1.0watermill-amqp10支持 AMQP 1.0 協(xié)議區(qū)別于官方內(nèi)置的 AMQP/RabbitMQ 適配器Apache Pulsarwatermill-pulsar云原生流式消息平臺(tái) PulsarApache RocketMQwatermill-rocketmq阿里巴巴開源的分布式消息中間件CockroachDBwatermill-crdb基于分布式數(shù)據(jù)庫 CockroachDB 的持久化 Pub/SubEnsignwatermill-ensignRotational 的事件基礎(chǔ)設(shè)施服務(wù)GoogleCloud Pub/Sub HTTP Pushwatermill-googlecloud-http以 HTTP Push 方式接入 GoogleCloud Pub/SubMongoDBwatermill-mongodb以 MongoDB 作為消息存儲(chǔ)MQTTwatermill-mqtt物聯(lián)網(wǎng)場景常用的輕量級(jí)消息協(xié)議NSQwatermill-nsq分布式實(shí)時(shí)消息平臺(tái) NSQRedis Zsetwatermill-rediszset基于 Redis 有序集合的輕量 Pub/SubSQLitewatermill-comfymill以 SQLite 為后端的單機(jī)持久化從這些條目可以看出生態(tài)的多樣性除了傳統(tǒng)消息隊(duì)列還可以用數(shù)據(jù)庫MongoDB、CockroachDB、SQLite或 Redis 結(jié)構(gòu)實(shí)現(xiàn)發(fā)布/訂閱。官方內(nèi)置支持的中間件清單見 docs/content/pubsubs/_index.mdKafka、AMQP、GoChannel、Redis Stream、GoogleCloud、NATS、SQL 等第三方清單正是對(duì)它的有力補(bǔ)充。日志適配器Logging清單按日志庫分三組每組都有多個(gè)可選適配器logruswatermill-logrus-adapter與walrus兩個(gè)適配器zap兩個(gè)同名watermillzap適配器來自不同作者zerologzerowater、watermillzlog、zerolog-watermill-adapter三個(gè)適配器。選擇時(shí)可以根據(jù)團(tuán)隊(duì)已有日志棧決定。接入方式統(tǒng)一為構(gòu)造一個(gè)實(shí)現(xiàn) LoggerAdapter 的適配器實(shí)例再將其傳入需要日志的組件構(gòu)造函數(shù)——例如 gochannel.NewGoChannel 的第二個(gè)參數(shù)logger watermill.LoggerAdapter傳nil時(shí)內(nèi)部自動(dòng)回退到watermill.NopLogger{}。日志適配器的With(fields LogFields)方法用于攜帶上下文字段如pubsub_uuid便于關(guān)聯(lián)同一組件實(shí)例產(chǎn)生的所有日志?？捎^測性O(shè)bservability這一組把 Watermill 的組件接入業(yè)界主流觀測棧OpenCensuswatermill-opencensus與ocwatermill兩個(gè)適配器OpenTelemetrywatermill-opentelemetry、watermill-opentelemetry-go-extra、watermill-opentelemetry另一作者以及針對(duì) AMQP 的otel-watermill-amqp和針對(duì) GoChannel 的watermill-otel-tracable-gochannel。值得注意 AMQP 與 GoChannel 都有專門的追蹤適配器說明消息鏈路追蹤需要針對(duì)傳輸層實(shí)現(xiàn)細(xì)節(jié)做處理例如把 trace 上下文注入/提取到消息的 Metadata。倉庫內(nèi)官方的觀測能力集中在 components/metrics 組件基于 Prometheus 的指標(biāo)構(gòu)建器、Handler 裝飾器等第三方可觀測性庫則主要面向分布式追蹤兩者可以搭配使用。其他工具Othergo-watermill-templateAsyncAPI 社區(qū)提供的代碼生成模板可從 AsyncAPI 規(guī)范生成 Watermill 相關(guān)代碼watermillx一套 Watermill 擴(kuò)展工具集protoc-gen-event從 Protobuf 定義生成事件代碼的插件與倉庫內(nèi) CQRS 組件的 Protobuf marshaler 思路一脈相承。如何評(píng)估第三方適配器用通用測試套件做體檢既然官方聲明不保證第三方庫可用讀者就必須自己驗(yàn)證。Watermill 為此提供了標(biāo)準(zhǔn)工具——通用 Pub/Sub 測試套件位于 pubsub/tests/test_pubsub.go。任何自稱生產(chǎn)可用的 Pub/Sub 實(shí)現(xiàn)都應(yīng)能通過它測試套件本身也常被第三方適配器的作者直接復(fù)用。入口函數(shù)簽名如下func TestPubSub( t *testing.T, features Features, pubSubConstructor PubSubConstructor, consumerGroupPubSubConstructor ConsumerGroupPubSubConstructor, )它內(nèi)部自動(dòng)運(yùn)行一批覆蓋基礎(chǔ)與邊界場景的測試TestPublishSubscribe基礎(chǔ)收發(fā)、TestConcurrentSubscribe50 個(gè)并發(fā)訂閱者、TestConcurrentSubscribeMultipleTopics、TestResendOnErrorNack 后重投遞、TestNoAck未 Ack 不推送下一條、TestContinueAfterSubscribeClose、TestConcurrentClose、TestContinueAfterErrors、TestPublishSubscribeInOrder、TestPublisherClose、TestTopic、TestMessageCtx、TestSubscribeCtx、TestNewSubscriberReceivesOldMessages與TestReconnect外加消費(fèi)者組測試TestConsumerGroups。另有 TestPubSubStressTest 可通過環(huán)境變量STRESS_TEST_COUNT控制壓力輪數(shù)默認(rèn) 10 輪。Features結(jié)構(gòu)體是評(píng)估適配器能力邊界的關(guān)鍵也是閱讀第三方庫文檔時(shí)的對(duì)照表字段含義ConsumerGroups是否支持消費(fèi)者組ExactlyOnceDelivery是否支持恰好一次投遞GuaranteedOrder是否保證消息順序GuaranteedOrderWithSingleSubscriber僅單個(gè)訂閱者時(shí)是否保證順序Persistent消息是否持久化GoChannel 不支持RestartServiceCommand用于測試斷線重連的重啟 broker 的命令RequireSingleInstance是否要求單實(shí)例工作如 GoChannelNewSubscriberReceivesOldMessages新訂閱者能否收到歷史消息如 KafkaContextPreserved是否保留消息發(fā)布時(shí)的 Context對(duì)照倉庫內(nèi)官方適配器的特性表可以直觀感受特性矩陣的寫法例如 docs/content/pubsubs/kafka.md 標(biāo)明 Kafka 適配器支持消費(fèi)者組、持久化與保證順序需配合分區(qū)鍵但不支持恰好一次投遞而 docs/content/pubsubs/gochannel.md 的 GoChannel 恰好相反——不支持消費(fèi)者組與持久化。評(píng)估第三方適配器時(shí)也應(yīng)要求作者提供同樣格式的特性表。自行實(shí)現(xiàn)并貢獻(xiàn)一個(gè)適配器如果清單中沒有你需要的中間件Implementing custom Pub/Sub 給出了標(biāo)準(zhǔn)路徑只需實(shí)現(xiàn)message.Publisher與message.Subscriber兩個(gè)接口然后用通用測試套件驗(yàn)證。結(jié)合該文檔與源碼實(shí)現(xiàn)時(shí)不要遺漏以下檢查清單日志提供清晰、分級(jí)的日志消息復(fù)用LoggerAdapter的 Error/Info/Debug/Trace 級(jí)別可替換的消息 Marshaler消息序列化策略應(yīng)可配置、可替換參考 Kafka 適配器kafka.Marshaler的設(shè)計(jì)健壯的Close()發(fā)布器與訂閱器的 Close 必須冪等在發(fā)布器/訂閱器被阻塞如等待 Ack時(shí)仍能正確關(guān)閉在訂閱器輸出通道無人讀取時(shí)也能正確關(guān)閉完整的 Ack/Nack消費(fèi)到的消息必須同時(shí)支持Ack()與Nack()Nack 后的重投遞Nack()后消息應(yīng)能重新投遞這正是 at-least-once 語義的體現(xiàn)參見 docs/content/docs/pub-sub.md跑通通用測試使用 pubsub/tests/test_pubsub.go 中的通用測試套件與壓力測試聲明實(shí)現(xiàn)的Features測試排查技巧見 troubleshooting性能優(yōu)化文檔完備提供 GoDoc、Markdown 文檔對(duì)應(yīng) docs/content/pubsubs/ 目錄下的中間件文檔格式與 Getting Started 示例對(duì)應(yīng) _examples/pubsubs/ 目錄的格式。以倉庫內(nèi)置的 GoChannel 實(shí)現(xiàn) pubsub/gochannel/pubsub.go 為最小參考它的Config只有四個(gè)字段輸出通道緩沖OutputChannelBuffer、內(nèi)存持久化Persistent、發(fā)布阻塞直到訂閱者 Ack 的BlockPublishUntilSubscriberAck、保留 Context 的PreserveContext構(gòu)造函數(shù)NewGoChannel(config, logger)返回*GoChannel實(shí)現(xiàn)了完整的 Publisher/Subscriber。第三方適配器的接入形態(tài)與之完全一致——只是把內(nèi)存 channel 換成外部中間件的連接。接入到路由器后發(fā)布側(cè)調(diào)用Publish(topic, messages...)消費(fèi)側(cè)通過Subscribe(ctx, topic)拿到-chan *message.Message配合Ack()/Nack()即可納入 Watermill 的路由與中間件體系。完成實(shí)現(xiàn)后歡迎向官方提交 Pull Request或把自己的庫補(bǔ)充進(jìn) docs/content/docs/awesome.md 的 Awesome 清單幫助后來的使用者發(fā)現(xiàn)它?！久赓M(fèi)下載鏈接】watermillBuilding event-driven applications the easy way in Go.項(xiàng)目地址: https://gitcode.com/GitHub_Trending/wa/watermill創(chuàng)作聲明：本文部分內(nèi)容由AI輔助生成（AIGC），僅供參考 Watermill 生態(tài)擴(kuò)展實(shí)戰(zhàn)指南：Awesome 第三方庫清單、日志與可觀測性適配器及自定義 Pub/Sub
Watermill 生態(tài)擴(kuò)展實(shí)戰(zhàn)指南Awesome 第三方庫清單、日志與可觀測性適配器及自定義 Pub/Sub【免費(fèi)下載鏈接】watermillBuilding event-driven applications the easy way in Go.項(xiàng)目地址: https://gitcode.com/GitHub_Trending/wa/watermillWatermill 的核心設(shè)計(jì)是官方內(nèi)核 可插拔適配器框架只定義消息模型與路由器而具體的消息中間件Pub/Sub、日志、追蹤全部通過接口接入。官方將社區(qū)中未經(jīng)其維護(hù)的優(yōu)質(zhì)第三方庫集中收錄于 Awesome Watermill 清單即本倉庫docs/content/docs/awesome.md按示例項(xiàng)目、Pub/Sub 適配器、日志適配器、可觀測性與其他工具分類。閱讀本文后你將掌握該清單的完整構(gòu)成、第三方庫與 Watermill 的接入接口原理并能依據(jù)倉庫內(nèi)的通用測試套件評(píng)估一個(gè)第三方適配器的成熟度甚至獨(dú)立實(shí)現(xiàn)并貢獻(xiàn)你自己的適配器。官方與第三方生態(tài)的邊界Awesome 清單的定位Awesome 清單的第一段文字就劃清了責(zé)任邊界下列庫并非由 Three Dots Labs 維護(hù)官方無法提供支持、也不保證它們一定工作正常使用前需要自行調(diào)研Do your own research。這正是它被稱為 Selected unofficial libraries 的原因——官方收錄它們只是因?yàn)槟憧赡軙?huì)覺得有用而不是背書。該清單本身是一個(gè)可維護(hù)的文檔文件如果你知道其他值得收錄的庫或者你本人就是某個(gè)適配器的作者可以通過編輯 docs/content/docs/awesome.md 把它加入清單。整個(gè)清單分為五類Examples社區(qū)編寫的完整示例項(xiàng)目Pub/Subs針對(duì)各類消息中間件的第三方適配器Logging將 Watermill 日志接入 logrus / zap / zerolog 的適配器Observability接入 OpenCensus / OpenTelemetry 的追蹤與指標(biāo)適配器Other代碼生成、工具集等周邊設(shè)施。清單末尾還特別引導(dǎo)讀者如果想了解如何實(shí)現(xiàn)自己的 Pub/Sub 適配器參見 Implementing custom Pub/Sub。理解接入點(diǎn)第三方庫如何與 Watermill 銜接要評(píng)估任何一個(gè)第三方庫先要理解 Watermill 為它們定義的插槽。核心接入點(diǎn)全部在 message/pubsub.go 中任何消息中間件適配器本質(zhì)上都是這兩個(gè)接口的實(shí)現(xiàn)// Publisher is the emitting part of a Pub/Sub. type Publisher interface { // Publish publishes provided messages to the given topic. // Publish can be synchronous or asynchronous - it depends on the implementation. // Most publisher implementations dont support atomic publishing of messages. // Publish does not work with a single Context. Use the Context() method of each message instead. // Publish must be thread safe. Publish(topic string, messages ...*Message) error // Close should flush unsent messages if publisher is async. Close() error } // Subscriber is the consuming part of the Pub/Sub. type Subscriber interface { // Subscribe returns an output channel with messages from the provided topic. // The channel is closed after Close() is called on the subscriber. // To receive the next message, Ack() must be called on the received message. // If message processing fails and the message should be redelivered Nack() should be called instead. Subscribe(ctx context.Context, topic string) (-chan *Message, error) // Close closes all subscriptions with their output channels and flushes offsets etc. when needed. Close() error }從接口注釋可以提煉出第三方適配器必須遵守的關(guān)鍵契約Publish必須線程安全且不保證原子性——批量發(fā)布多條消息時(shí)一旦某條失敗后續(xù)消息不會(huì)繼續(xù)發(fā)布見 docs/content/docs/pub-sub.md 的 Publishing multiple messages 一節(jié)同步/異步由實(shí)現(xiàn)決定異步發(fā)布器必須在Close()時(shí)沖刷未發(fā)送的消息忘記關(guān)閉發(fā)布器可能丟失消息Ack/Nack 是 Subscriber 的職責(zé)正確實(shí)現(xiàn)應(yīng)在收到上一條消息的Ack/Nack后再消費(fèi)下一條且必須等 Watermill 處理完消息、返回 Ack 之后才向底層 broker 提交 offset否則進(jìn)程在消息處理完成前崩潰會(huì)丟消息另有可選的 SubscribeInitializer 接口含SubscribeInitialize(topic string) error用于在消費(fèi)前初始化訂閱非強(qiáng)制實(shí)現(xiàn)。對(duì)日志類第三方庫而言接入點(diǎn)則是 log.go 中定義的LoggerAdapter接口type LoggerAdapter interface { Error(msg string, err error, fields LogFields) Info(msg string, fields LogFields) Debug(msg string, fields LogFields) Trace(msg string, fields LogFields) With(fields LogFields) LoggerAdapter }Watermill 自身提供了寫入標(biāo)準(zhǔn)輸出的 StdLoggerAdapterNewStdLogger(debug, trace bool)與靜默的NopLogger第三方日志適配器的工作就是把LoggerAdapter的方法調(diào)用橋接到 logrus、zap、zerolog 各自的 API 上。逐類盤點(diǎn) Awesome 清單中的第三方庫以下內(nèi)容完整覆蓋 docs/content/docs/awesome.md 中收錄的每一類庫。需要說明的是清單原文以外部倉庫鏈接形式給出此處按名稱與功能分類整理便于按圖索驥。示例項(xiàng)目Examples清單收錄了三個(gè)社區(qū)示例用于學(xué)習(xí)如何在真實(shí)項(xiàng)目中組合使用 Watermillgolang-taipei-watermill-exampleWatermill 的完整示例項(xiàng)目Kafka-PubSub以 Kafka 為消息中間件的示例go-example-financing一個(gè)金融場景融資的 Go 示例。這些項(xiàng)目與倉庫內(nèi)官方維護(hù)的_examples/目錄互補(bǔ)——官方示例覆蓋了 基礎(chǔ)應(yīng)用、路由器、CQRS、指標(biāo) 以及各類 Pub/Sub 接入示例社區(qū)示例則展示更多真實(shí)業(yè)務(wù)組合。Pub/Sub 適配器第三方這是清單中體量最大的一類。每個(gè)適配器都實(shí)現(xiàn)了上文所述的message.Publisher/message.Subscriber接口把 Watermill 統(tǒng)一的消息模型映射到特定中間件。清單收錄了以下第三方適配器目標(biāo)中間件適配器定位AMQP 1.0watermill-amqp10支持 AMQP 1.0 協(xié)議區(qū)別于官方內(nèi)置的 AMQP/RabbitMQ 適配器Apache Pulsarwatermill-pulsar云原生流式消息平臺(tái) PulsarApache RocketMQwatermill-rocketmq阿里巴巴開源的分布式消息中間件CockroachDBwatermill-crdb基于分布式數(shù)據(jù)庫 CockroachDB 的持久化 Pub/SubEnsignwatermill-ensignRotational 的事件基礎(chǔ)設(shè)施服務(wù)GoogleCloud Pub/Sub HTTP Pushwatermill-googlecloud-http以 HTTP Push 方式接入 GoogleCloud Pub/SubMongoDBwatermill-mongodb以 MongoDB 作為消息存儲(chǔ)MQTTwatermill-mqtt物聯(lián)網(wǎng)場景常用的輕量級(jí)消息協(xié)議NSQwatermill-nsq分布式實(shí)時(shí)消息平臺(tái) NSQRedis Zsetwatermill-rediszset基于 Redis 有序集合的輕量 Pub/SubSQLitewatermill-comfymill以 SQLite 為后端的單機(jī)持久化從這些條目可以看出生態(tài)的多樣性除了傳統(tǒng)消息隊(duì)列還可以用數(shù)據(jù)庫MongoDB、CockroachDB、SQLite或 Redis 結(jié)構(gòu)實(shí)現(xiàn)發(fā)布/訂閱。官方內(nèi)置支持的中間件清單見 docs/content/pubsubs/_index.mdKafka、AMQP、GoChannel、Redis Stream、GoogleCloud、NATS、SQL 等第三方清單正是對(duì)它的有力補(bǔ)充。日志適配器Logging清單按日志庫分三組每組都有多個(gè)可選適配器logruswatermill-logrus-adapter與walrus兩個(gè)適配器zap兩個(gè)同名watermillzap適配器來自不同作者zerologzerowater、watermillzlog、zerolog-watermill-adapter三個(gè)適配器。選擇時(shí)可以根據(jù)團(tuán)隊(duì)已有日志棧決定。接入方式統(tǒng)一為構(gòu)造一個(gè)實(shí)現(xiàn) LoggerAdapter 的適配器實(shí)例再將其傳入需要日志的組件構(gòu)造函數(shù)——例如 gochannel.NewGoChannel 的第二個(gè)參數(shù)logger watermill.LoggerAdapter傳nil時(shí)內(nèi)部自動(dòng)回退到watermill.NopLogger{}。日志適配器的With(fields LogFields)方法用于攜帶上下文字段如pubsub_uuid便于關(guān)聯(lián)同一組件實(shí)例產(chǎn)生的所有日志?？捎^測性O(shè)bservability這一組把 Watermill 的組件接入業(yè)界主流觀測棧OpenCensuswatermill-opencensus與ocwatermill兩個(gè)適配器OpenTelemetrywatermill-opentelemetry、watermill-opentelemetry-go-extra、watermill-opentelemetry另一作者以及針對(duì) AMQP 的otel-watermill-amqp和針對(duì) GoChannel 的watermill-otel-tracable-gochannel。值得注意 AMQP 與 GoChannel 都有專門的追蹤適配器說明消息鏈路追蹤需要針對(duì)傳輸層實(shí)現(xiàn)細(xì)節(jié)做處理例如把 trace 上下文注入/提取到消息的 Metadata。倉庫內(nèi)官方的觀測能力集中在 components/metrics 組件基于 Prometheus 的指標(biāo)構(gòu)建器、Handler 裝飾器等第三方可觀測性庫則主要面向分布式追蹤兩者可以搭配使用。其他工具Othergo-watermill-templateAsyncAPI 社區(qū)提供的代碼生成模板可從 AsyncAPI 規(guī)范生成 Watermill 相關(guān)代碼watermillx一套 Watermill 擴(kuò)展工具集protoc-gen-event從 Protobuf 定義生成事件代碼的插件與倉庫內(nèi) CQRS 組件的 Protobuf marshaler 思路一脈相承。如何評(píng)估第三方適配器用通用測試套件做體檢既然官方聲明不保證第三方庫可用讀者就必須自己驗(yàn)證。Watermill 為此提供了標(biāo)準(zhǔn)工具——通用 Pub/Sub 測試套件位于 pubsub/tests/test_pubsub.go。任何自稱生產(chǎn)可用的 Pub/Sub 實(shí)現(xiàn)都應(yīng)能通過它測試套件本身也常被第三方適配器的作者直接復(fù)用。入口函數(shù)簽名如下func TestPubSub( t *testing.T, features Features, pubSubConstructor PubSubConstructor, consumerGroupPubSubConstructor ConsumerGroupPubSubConstructor, )它內(nèi)部自動(dòng)運(yùn)行一批覆蓋基礎(chǔ)與邊界場景的測試TestPublishSubscribe基礎(chǔ)收發(fā)、TestConcurrentSubscribe50 個(gè)并發(fā)訂閱者、TestConcurrentSubscribeMultipleTopics、TestResendOnErrorNack 后重投遞、TestNoAck未 Ack 不推送下一條、TestContinueAfterSubscribeClose、TestConcurrentClose、TestContinueAfterErrors、TestPublishSubscribeInOrder、TestPublisherClose、TestTopic、TestMessageCtx、TestSubscribeCtx、TestNewSubscriberReceivesOldMessages與TestReconnect外加消費(fèi)者組測試TestConsumerGroups。另有 TestPubSubStressTest 可通過環(huán)境變量STRESS_TEST_COUNT控制壓力輪數(shù)默認(rèn) 10 輪。Features結(jié)構(gòu)體是評(píng)估適配器能力邊界的關(guān)鍵也是閱讀第三方庫文檔時(shí)的對(duì)照表字段含義ConsumerGroups是否支持消費(fèi)者組ExactlyOnceDelivery是否支持恰好一次投遞GuaranteedOrder是否保證消息順序GuaranteedOrderWithSingleSubscriber僅單個(gè)訂閱者時(shí)是否保證順序Persistent消息是否持久化GoChannel 不支持RestartServiceCommand用于測試斷線重連的重啟 broker 的命令RequireSingleInstance是否要求單實(shí)例工作如 GoChannelNewSubscriberReceivesOldMessages新訂閱者能否收到歷史消息如 KafkaContextPreserved是否保留消息發(fā)布時(shí)的 Context對(duì)照倉庫內(nèi)官方適配器的特性表可以直觀感受特性矩陣的寫法例如 docs/content/pubsubs/kafka.md 標(biāo)明 Kafka 適配器支持消費(fèi)者組、持久化與保證順序需配合分區(qū)鍵但不支持恰好一次投遞而 docs/content/pubsubs/gochannel.md 的 GoChannel 恰好相反——不支持消費(fèi)者組與持久化。評(píng)估第三方適配器時(shí)也應(yīng)要求作者提供同樣格式的特性表。自行實(shí)現(xiàn)并貢獻(xiàn)一個(gè)適配器如果清單中沒有你需要的中間件Implementing custom Pub/Sub 給出了標(biāo)準(zhǔn)路徑只需實(shí)現(xiàn)message.Publisher與message.Subscriber兩個(gè)接口然后用通用測試套件驗(yàn)證。結(jié)合該文檔與源碼實(shí)現(xiàn)時(shí)不要遺漏以下檢查清單日志提供清晰、分級(jí)的日志消息復(fù)用LoggerAdapter的 Error/Info/Debug/Trace 級(jí)別可替換的消息 Marshaler消息序列化策略應(yīng)可配置、可替換參考 Kafka 適配器kafka.Marshaler的設(shè)計(jì)健壯的Close()發(fā)布器與訂閱器的 Close 必須冪等在發(fā)布器/訂閱器被阻塞如等待 Ack時(shí)仍能正確關(guān)閉在訂閱器輸出通道無人讀取時(shí)也能正確關(guān)閉完整的 Ack/Nack消費(fèi)到的消息必須同時(shí)支持Ack()與Nack()Nack 后的重投遞Nack()后消息應(yīng)能重新投遞這正是 at-least-once 語義的體現(xiàn)參見 docs/content/docs/pub-sub.md跑通通用測試使用 pubsub/tests/test_pubsub.go 中的通用測試套件與壓力測試聲明實(shí)現(xiàn)的Features測試排查技巧見 troubleshooting性能優(yōu)化文檔完備提供 GoDoc、Markdown 文檔對(duì)應(yīng) docs/content/pubsubs/ 目錄下的中間件文檔格式與 Getting Started 示例對(duì)應(yīng) _examples/pubsubs/ 目錄的格式。以倉庫內(nèi)置的 GoChannel 實(shí)現(xiàn) pubsub/gochannel/pubsub.go 為最小參考它的Config只有四個(gè)字段輸出通道緩沖OutputChannelBuffer、內(nèi)存持久化Persistent、發(fā)布阻塞直到訂閱者 Ack 的BlockPublishUntilSubscriberAck、保留 Context 的PreserveContext構(gòu)造函數(shù)NewGoChannel(config, logger)返回*GoChannel實(shí)現(xiàn)了完整的 Publisher/Subscriber。第三方適配器的接入形態(tài)與之完全一致——只是把內(nèi)存 channel 換成外部中間件的連接。接入到路由器后發(fā)布側(cè)調(diào)用Publish(topic, messages...)消費(fèi)側(cè)通過Subscribe(ctx, topic)拿到-chan *message.Message配合Ack()/Nack()即可納入 Watermill 的路由與中間件體系。完成實(shí)現(xiàn)后歡迎向官方提交 Pull Request或把自己的庫補(bǔ)充進(jìn) docs/content/docs/awesome.md 的 Awesome 清單幫助后來的使用者發(fā)現(xiàn)它?！久赓M(fèi)下載鏈接】watermillBuilding event-driven applications the easy way in Go.項(xiàng)目地址: https://gitcode.com/GitHub_Trending/wa/watermill創(chuàng)作聲明：本文部分內(nèi)容由AI輔助生成（AIGC），僅供參考