1. 程式人生 > >如何在Rails應用程式中使用Kafka?

如何在Rails應用程式中使用Kafka?

背景介紹

有那麼一段時間,我們的系統需要用到分散式流式處理和訊息系統,而 Apache Kafka 似乎成了我們建立業務關鍵型應用程式的堅實基礎。它可用於很多場景下,比如產品更新管道、訂單跟蹤、實時使用者通知、商戶賬單等。

接下來的故事講述了我們如何將 Kafka 引入到我們的 Rails 單體程式碼庫中,內容包括技術細節、我們面臨的挑戰以及我們在此過程中所做的技術決策。

眼前的挑戰

第一個問題是 Kafka 只提供了相對較底層的抽象。雖然這具有一定的優勢,但同時也意味著客戶端開發者需要面對更多的 API,需要處理更多的細節,實現一個 Kafka 客戶端也因此變成了一項艱鉅的任務。

作為一個基於 Ruby 的專案,我們嘗試了各種使用 Ruby 開發的 Kafka 客戶端,但總是碰到一些難以診斷的錯誤。 Ruby 缺乏併發原語,要寫出一個高效的客戶端並不容易。

我們通過多種方式來歸避這些問題:通過獨立服務來隱藏底層的複雜性,只為客戶端提供最小化的 API 集合。這個服務可以使用 Ruby 以外的語言開發,所以我們就可以用上久經驗證的 librdkafka,我們在其他的 Python 和 Go 應用程式中也使用過這個庫。

於是,我們開發了 Rafka——位於 Kafka 前端的代理服務,並通過簡單的語義和 API 把它暴露出來。它提供了合理的預設配置,為使用者隱藏了很多繁雜的細節。我們選擇了 Go 語言,因為它已經有一個健壯的基於 librdkafka 的 Kafka 客戶端,並提供了必要的工具來實現我們需要的功能。

為了避免讓客戶端的開發變複雜,我們選擇使用 Redis 協議的一個子集。我們所要做的只是在 Ruby 的 Redis 客戶端之上新增一個層。

幾天後,我們便有了一個使用 Ruby 開發的客戶端,打包成一個名為 rafka-rb 的 gem,其中包含了消費者和生產者。

有了 Rafka 及其配套的 Ruby 客戶端,我們的服務和 Rails 應用程式就可以輕鬆地從 Kafka 讀取資料和往 Kafka 寫入資料。

大部分開發人員的時間都花在了我們的 Rails 主應用程式上,因此,能夠在應用程式內輕鬆使用 Kafka 消費者和生產者就變得非常重要。接下來就是讓 Rails 開發人員直接用上 Kafka 消費者和生產者。

在 Rails 應用程式中傳送資料

將生產者整合到現有的應用程式中其實很簡單,因為即使需要使用多個主題,也只需要一個生產者。

因此,我們使用了單個生產者例項,並在應用程式初始化的時候建立它,整個程式碼庫都使用這個例項:

# config/initializers/kafka_producer.rb
Skroutz.kafka_producer = Rafka::Producer.new(...)

傳送訊息非常簡單:

Skroutz.kafka_producer.produce("greetings", "Hello there!")

在 Rails 應用程式中讀取資料

使用消費者就有點不一樣了,因為消費訊息需要長時間執行。接下來,我們將看到如何在 Rails 程式碼庫中通過 Rafka 來使用 Kafka 消費者。

文末提供了相關元件原始碼的連結。

消費者是普通的 Ruby 物件,它們的類是在 Rails 應用程式中定義的。它們繼承了 KafkaConsumer 抽象類,這個抽象類集成了用於統計的 statsd 和用於錯誤跟蹤的 Sentry,在將來可能還會整合其他東西。它們的類名以“Consumer”作為字尾,相應的檔案按照 Rails 慣例來命名。

典型的消費者看起來如下:

在這裡,每個消費者都使用了 Rafka :: Consumer 例項。

在寫好新的消費者之後,需要在配置檔案中啟用它:

  • name: “price_drops”
    scale: 2
    按照 Rails 慣例,消費者的名字來自類名。

關鍵是,所有消費者例項基本上都是獨立的 Kafka 消費者,它們同屬於一個消費者群組。

在部署時,Capistrano 會讀取配置檔案,並在伺服器上建立適當的消費者例項。

這些就是開發和部署消費者所要做的事情。

下一個問題來了:如何將消費者作為長時間執行的程序?

長時間執行的消費者程序

在實現了消費者之後,下一步就是執行它們。

我們使用了一個名為 KafkaConsumerWorker 的類,這個類封裝了消費者物件,並讓它們成為長時間執行的程序。下面給出了這個類的簡化版程式碼:

KafkaConsumerWorker 不斷呼叫底層消費者的 #process 方法來迴圈處理訊息。它還提供了優雅的退出功能。它還將消費者與 systemd 整合在一起,用以提供健壯性、活躍度檢查、可見性和監控能力。

在不需要直接與 KafkaConsumerWorker 發生互動的情況下進行開發或除錯也很容易:

consumer = PriceDropsConsumer.new(...)
worker = KafkaConsumerWorker.new(consumer)
# start work loop
worker.work

下一步是使用 systemd 啟動 KafkaConsumerWorker。我們使用了一個簡單的 systemd 服務檔案:

每個消費者例項使用包含消費者名稱和例項編號(例如 price_drops:1)的字串作為標識,該例項編號作為模板引數(%i 部分)傳遞給 systemd。這樣我們就可以使用相同的服務檔案生成不同的消費者例項。

將消費者與 systemd 整合意味著我們可以使用消費者內建的很多功能:

消費者管理命令(start、stop、restart、status)
在消費者發生異常時發出告警
可見性:每個消費者的狀態(工作中、等待作業、已關閉)、其當前偏移量 / 主題 / 分割槽(使用 sd_notify(3))
自動重啟失效的消費者
通過 systemd watchdog 計時器自動重啟被掛起的消費者
簡化的日誌:我們只需將日誌打到 stdout/stderr,systemd 負責處理其餘部分
通過檢查消費者例項,我們可以得到非常有用的資訊輸出,在出現問題時,這些資訊可用於除錯問題:

最後要解決的問題是,為了重啟消費者,systemd 需要呼叫哪個命令。這個命令實際上是一個普通的 rake 任務,它將消費者設定為 worker 並執行它。

與其他元件類似,該任務的相關程式碼也放在 Rails 程式碼庫中:

部署

由於我們使用 Capistrano 進行部署,所以添加了一個 Capistrano 任務,負責停止和啟動消費者。它的簡化版本如下:

kafkactl 是一個包裝指令碼,負責執行必要的 systemctl 命令。

當有人部署應用程式時,Capistrano 會讀取 YAML 配置檔案並建立消費者:

在部署好消費者後,我們檢視 Grafana 儀表盤,確保一切正常,同時我們也會檢視 Slack,確保沒有觸發任何告警。

整體架構

我們的 Kafka/Rails 整合基礎架構包含以下元件:

Rafka:具有簡單語義和最小化 API 集合的 Kafka 代理服務
rafka-rb:Rafka 的 Ruby 客戶端
KafkaConsumer:一個 Ruby 抽象類,具體的消費者實現類會繼承這個類
KafkaConsumerWorker:一個 Ruby 類,用於將消費者作為長時間執行的程序
kafka:consumer:執行消費者例項的 rake 任務
kafka_consumers.yml:一個配置檔案,用於控制哪些消費者應該在生產環境中執行以及使用多少個例項
[email protected]:通過呼叫 rake 任務生成消費者的 systemd 服務檔案
它們之間互動如下圖所示:

從圖中可以看到,這些元件是正交分佈的,無論是用於除錯、測試還是原型設計,它們中的每一個都可以與其他元件的分開使用。

監控

因為很多消費者需要執行關鍵任務,所以必須對它們進行充分的監控。

監控發生在各個層面,每個消費者都提供瞭如下特性:

當消費者失效時,Icinga 發出告警(通過 systemd)
當發生異常時發出 Sentry 事件

統計:作業程序時間和消費者吞吐量(已處理訊息數 / 秒)

當消費者消費速度落後時(通過 Burrow 和 Grafana)
這些功能主要得益於我們使用了通用的消費者基礎架構。

未來展望

我們非常喜歡通過這種方式與 Kafka 進行互動,並且收到了非常積極的反饋。

通過幾個簡單的步驟就能開發和部署好消費者,這極大提升了開發團隊的效率,而且,我們能夠以一致和高效的方式基於 Kafka 開發應用程式。

將來,我們希望將本文中描述的所有元件開源出來,讓其他組織也能從中受益。

最後,我們計劃向 Rafka 和消費者 / 生產者基礎架構中新增更多功能,包括:

批處理功能
多主題消費者
基於 KSQL 的原語(聚合、連線等)
消費者鉤子(hook)
歡迎工作一到五年的Java工程師朋友們加入Java程式設計師開發: 854393687
群內提供免費的Java架構學習資料(裡面有高可用、高併發、高效能及分散式、Jvm效能調優、Spring原始碼,MyBatis,Netty,Redis,Kafka,Mysql,Zookeeper,Tomcat,Docker,Dubbo,Nginx等多個知識點的架構資料)合理利用自己每一分每一秒的時間來學習提升自己,不要再用"沒有時間“來掩飾自己思想上的懶惰!趁年輕,使勁拼,給未來的自己一個交代!