自動化 Kafka 主題創(chuàng)建和刪除對于使用事件驅(qū)動架構(gòu)的開發(fā)人員至關(guān)重要。 隨著 Kafka 部署的擴展，無論您是處理不斷增長的系統(tǒng)還是實施基礎(chǔ)設(shè)施即代碼，手動管理都變得不切實際。本指南演示了如何使用 JavaScript 和 GitHub Actions 自動化此過程，并將其無縫集成到您的 DevOps 工作流程中。

手動 Kafka 主題管理的挑戰(zhàn)

Kafka 主題是活動組織的基礎(chǔ)。 隨著您的架構(gòu)擴展，您將需要不同的主題：

• 對不同的事件類型進行分類
• 存儲過濾后的事件子集
• 維護轉(zhuǎn)換后的事件版本

隨著系統(tǒng)的增長，手動主題管理變得復(fù)雜且容易出錯。每個新服務(wù)可能需要新主題，每個環(huán)境（開發(fā)、登臺、生產(chǎn)）都需要自己一致的配置集。 如果沒有自動化，這種手動開銷會減慢開發(fā)速度并增加配置錯誤。

自動化主題管理的好處

自動化 Kafka 主題創(chuàng)建和刪除具有顯著優(yōu)勢：

1. 所有環(huán)境中一致的主題配置。
2. 減少主題管理中的人為錯誤。
3. 主題配置的版本控制。
4. 簡化的 DevOps 工作流程。
5. 將主題更改集成到 CI/CD 管道中。

選擇正確的工具

本教程使用 JavaScript 作為其自動化解決方案。雖然其他語言也是可能的，但 JavaScript 具有以下優(yōu)點：

• 比 Bash 更簡單的腳本。
• 利用現(xiàn)有團隊 JavaScript 專業(yè)知識。
• 通過 npm 豐富的軟件包生態(tài)系統(tǒng)。
• Kafka 操作的簡潔異步/等待語法。

此解決方案作為 Node.js 應(yīng)用程序?qū)崿F(xiàn)，作為 Kubernetes 作業(yè)運行，非常適合使用 VPC 可訪問的 Kafka 集群的團隊。

設(shè)置應(yīng)用程序

您需要一個 Node.js 項目。使用 npm init -y 創(chuàng)建一個。 如果未安裝 Node.js 和 npm，請從http://m.miracleart.cn/link/0d78f6439e652fdbf801d103430d2e12下載。

在您的項目目錄中，運行 npm install kafkajs 安裝 Kafka JavaScript 客戶端。

實施解決方案

我們的自動化包括三個核心組件：

1. 指定要創(chuàng)建或刪除的主題的 JSON 配置文件。
2. Node.js 應(yīng)用程序與 Kafka 交互來執(zhí)行這些操作。
3. GitHub Actions 工作流程在 JSON 配置文件更改時觸發(fā)自動化。

讓我們構(gòu)建每個組件。

配置文件設(shè)置

應(yīng)用程序讀取列出要創(chuàng)建或刪除的主題的 JSON 文件。 這允許通過 GitHub 拉取請求進行版本控制的更改。 合并 PR 后，代碼會讀取更新的文件并相應(yīng)地創(chuàng)建/刪除主題。

創(chuàng)建topics.json:

{
  "create": [],
  "delete": []
}

此結(jié)構(gòu)提供主題管理操作的版本控制記錄。 create 數(shù)組列出了要創(chuàng)建的主題，而 delete 列出了要刪除的主題。

構(gòu)建主題管理 API

創(chuàng)建api.js:

async function createTopics(topics, kafkaAdmin) {
  if (topics.length > 0) {
    await kafkaAdmin.createTopics({
      topics: topics.map((topic) => ({
        topic,
        numPartitions: 1,        // Adjust as needed
        replicationFactor: 3,    // Adjust as needed
        configEntries: [
          { name: "min.insync.replicas", value: "2" }  // Adjust as needed
        ],
      })),
    });
  }
}

async function deleteTopics(topics, kafkaAdmin) {
  if (topics.length > 0) {
    await kafkaAdmin.deleteTopics({ topics: topics });
  }
}

module.exports = { createTopics, deleteTopics };

該模塊導(dǎo)出用于創(chuàng)建和刪除 Kafka 主題的函數(shù)。 根據(jù)您的需求調(diào)整配置值（分區(qū)、復(fù)制因子、min.insync.replicas）。

創(chuàng)建主應(yīng)用程序

創(chuàng)建index.js:

const { Kafka } = require("kafkajs");
const { createTopics, deleteTopics } = require("./api");
const topics = require("../topics.json");

// ... (Kafka connection details using environment variables) ...

const kafka = new Kafka({
  // ... (Kafka connection configuration) ...
});

const admin = kafka.admin();

admin.connect().then(async () => {
  const existingTopics = await admin.listTopics();

  // Create only non-existent topics
  const newTopics = topics.create.filter((x) => !existingTopics.includes(x));
  await createTopics(newTopics, admin);

  // Delete only existing topics
  const deletionTopics = topics.delete.filter((x) => existingTopics.includes(x));
  await deleteTopics(deletionTopics, admin);

  await admin.disconnect();
});

該應(yīng)用程序處理 Kafka 連接、讀取主題配置并協(xié)調(diào)主題創(chuàng)建/刪除。 它包括防止重復(fù)操作的保護措施并確保干凈的連接處理。

使用 GitHub Actions 進行自動化部署

使用 GitHub Actions 將主題管理應(yīng)用程序集成到 CI/CD 管道中。 每當(dāng)主題配置發(fā)生變化時，這就會自動執(zhí)行。

添加.github/workflows/kafka.yml：

# ... (GitHub Actions workflow YAML configuration -  This section needs significant adaptation for a real-world scenario and would require setting up AWS credentials, ECR, and Kubernetes cluster details. The provided example is incomplete and illustrative only.) ...

此工作流程自動執(zhí)行構(gòu)建、容器化（使用 Docker）、推送到容器注冊表（如 Amazon ECR）以及作為作業(yè)部署到 Kubernetes。

結(jié)論

本指南演示了使用 JavaScript 和 GitHub Actions 自動創(chuàng)建和刪除 Kafka 主題。 這種方法使用 JSON 配置文件和 GitHub Actions，提供了一個可擴展的解決方案，用于隨著系統(tǒng)的增長管理 Kafka 主題。 將主題管理集成到 CI/CD 管道中可確保整個 Kafka 部署中的主題配置一致且可靠。 這將手動主題管理轉(zhuǎn)變?yōu)榕c現(xiàn)代 DevOps 實踐保持一致的簡化、自動化流程。 提供的代碼可作為定制和適應(yīng)您的特定要求的基礎(chǔ)。

以上是如何使用 GitHub Actions 和 JavaScript 自動創(chuàng)建和刪除 Kafka 主題的詳細內(nèi)容。更多信息請關(guān)注PHP中文網(wǎng)其他相關(guān)文章！