訂閱 Kafka 的活動

透過本教學，您可以學習如何使用 IBM Cloud® Code Engine CLI 訂閱 Kafka 事件。

在分散式環境中，您通常希望應用程式或工作對從其他元件 (通常稱為事件生產者) 產生的訊息 (事件) 做出反應。 透過 Code Engine，您的應用程式或工作可透過訂閱事件產生者，接收感興趣的事件。 應用程式會透過 POST HTTP 請求接收事件資訊，而工作則透過環境變數接收事件資訊。

Kafka 事件生產者會監看新訊息出現在 Kafka 實例中。 當您針對一組主題建立 Code Engine Kafka 訂閱時，您的應用程式或工作會針對出現在其中一個主題中的每一則新訊息收到個別事件。

雖然您可以使用任何 Kafka 實例，但本教學中的範例均採用 IBM® Event Streams for IBM Cloud® 服務。Event Streams 是一項 IBM 事件串流服務，專門用於處理 Kafka 事件。 如需有關此服務的更多資訊，請參閱 Event Streams 的文件。

開始之前

所有 Code Engine 使用者均須擁有一個「隨用隨付」帳戶。 指導教學可能會產生成本。 請使用「成本估算器」，根據您的預估使用量生成成本估算。 如需更多資訊，請參閱 Code Engine 價格資訊。

設定 Kafka 事件生產者

您可以設定您的 Kafka 訊息發送端，使其將訊息傳送至 Code Engine Kafka 的事件訂閱。 利用您的 Code Engine Kafka 事件訂閱，在收到 Kafka 訊息時觸發應用程式或工作。

若要開始使用，請為事件串流服務 建立 Event Streams 服務實例。 雖然您可以使用主控台或 CLI，但下列步驟說明如何使用 CLI 來設定 Event Streams 事件生產者。

  1. 建立 Event Streams的服務實例。 Event Streams CLI 服務的名稱是 messagehub。 在此範例中，建立名為 myeventstream 的 Event Streams 服務實例。

    ibmcloud resource service-instance-create myeventstream messagehub lite us-south

  2. 建立服務金鑰以提供認證給服務實例。

    ibmcloud resource service-key-create myeventstream-key Manager --instance-name myeventstream

    輸出範例

    Creating service key of service instance myeventstream under account <user_account>...
OK
Service key crn:v1:bluemix:public:messagehub:us-south:a/e43abfcbd191404cb17ef650e9681dd3:c0736069-3f4a-438a-b614-6846877d692d:resource-key:4c8edfdb-abcd-abcd-abcd-abcdabcdabcd was created.
Name:          myeventstream-key
ID:            crn:v1:bluemix:public:messagehub:us-south:a/e43abfcbd191404cb17ef650e9681dd3:c0736069-3f4a-438a-b614-6846877d692d:resource-key:4c8edfdb-abcd-abcd-abcd-abcdabcdabcd
Created At:    Mon Mar 21 18:36:09 UTC 2022
State:         active
Credentials:
            api_key:                  abcdeH9tu3qE5Sn8VbJfcDEWtjR_l0iPisB3abcdefgh
            apikey:                   abcdeH9tu3qE5Sn8VbJfcDEWtjR_l0iPisB3abcdefgh
            iam_apikey_description:   Auto-generated for key crn:v1:bluemix:public:messagehub:us-south:a/e43abfcbd191404cb17ef650e9681dd3:c0736069-3f4a-438a-b614-6846877d692d:resource-key:4c8edfdb-abcd-abcd-abcd-abcdabcdabcd
            iam_apikey_name:          myeventstream-key
            iam_role_crn:             crn:v1:bluemix:public:iam::::serviceRole:Manager
            iam_serviceid_crn:        crn:v1:bluemix:public:iam-identity::a/e43abfcbd191404cb17ef650e9681dd3::serviceid:ServiceId-3e99caa5-b174-4f04-9845-5c5d783b8bc7
            instance_id:              c0736069-3f4a-438a-b614-6846877d692d
            kafka_admin_url:          https://abcdabcdabcdabcd.svc07.us-south.eventstreams.cloud.ibm.com
            kafka_brokers_sasl:       [broker-1-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093 broker-2-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093 broker-5-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093 broker-3-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093 broker-4-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093 broker-0-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093]
            kafka_http_url:           https://abcdabcdabcdabcd.svc07.us-south.eventstreams.cloud.ibm.com
            password:                 abcdeH9tu3qE5Sn8VbJfcDEWtjR_l0iPisB3abcdefgh
            user:                     token

    記下服務金鑰的 user、password 及 kafka-brokers_sasl 值。 當您設定 Code Engine Kafka 訂閱時，需要此資訊。 password 和 apikey 的值在 Event Streams 服務實例的服務金鑰中是相同的。 您也可以使用 ibmcloud resource service-key myeventstream-key 指令來擷取服務金鑰資訊。

  3. 起始設定相對於 Event Streams 服務實例的 Event Streams 外掛程式。

    ibmcloud es init --instance-name myeventstream

  4. 建立 Event Streams 主題。

    ibmcloud es topic-create kafka-topic1

設定 Code Engine 範例應用程式以產生 Kafka 訊息

對於本指導教學，請設定 Code Engine 應用程式，以充當 Kafka 訊息的事件生產者。 此 kafka-sender-app 應用程式的目的是連接至 Event Streams 實例，並產生 (傳送) Kafka 訊息至訊息接收端 (Kafka 消費者)。 這個產生 Kafka 訊息事件的應用程式會使用 Code Engine Kafka 傳送端範例應用程式 來傳送 Kafka 訊息。 此範例傳送端映像檔需要 BROKERS 環境變數及包含 password 認證的密鑰。

  1. 使用 Code Engine Kafka 範例所需的認證來建立密鑰。 例如，建立 kafka-subscription-secret 密鑰，以包含 Kafka 傳送端範例應用程式及 Kafka 事件訂閱 (使用 Kafka 接收端範例) 所需的認證。 範例 Kafka 傳送端應用程式及 Code Engine Kafka 事件訂閱需要這些認證，才能與 Event Streams的服務實例進行通訊。 雖然在建立 Kafka 傳送端應用程式及事件訂閱之前不需要您建立此密鑰，但此動作可簡化必要的步驟。

    若要建立 kafka-subscription-secret 密鑰，請為 password 及 username 新增文字環境變數。 如需相關資訊，請參閱 使用 CLI 建立密鑰。

    • 使用 Event Streams 服務實例中服務認證的詳細資料中列出的 user 值來指定 username 金鑰。 對於 Event Streams 服務實例，此值為 token。 Code Engine Kafka 事件訂閱與 Kafka 訊息分配管理系統之間的鑑別需要此金鑰。
    • 使用 Event Streams 服務實例中服務認證的詳細資料中列出的 apikey 值來指定 password 金鑰。 寄件者範例以及啟用 Code Engine Kafka 事件訂閱與 Kafka 訊息分配管理系統之間的通訊需要此金鑰。
        ibmcloud ce secret create --name kafka-subscription-secret --from-literal password=<value_of_apikey> --from-literal username=<value_of_user>
    ```
    例如，

    ```txt {: pre}
    ibmcloud ce secret create --name kafka-subscription-secret --from-literal password=abcdeH9tu3qE5Sn8VbJfcDEWtjR_l0iPisB3abcdefgh --from-literal username=token
    ```

  2. 請根據以下資訊建立「kafka-sender-app」。

    • 指定 --image 選項以參照 icr.io/codeengine/kafka-sender 儲存器映像檔。 此映像檔是從 sender.go 建置，可從 Samples for IBM Cloud Code Engine GitHub repo取得。 此範例傳送端應用程式需要儲存在 kafka-subscription-secret 中的 password 認證，並且它需要 BROKERS 環境變數。
    • 指定 --env-from-secret 選項以參照包含 password 認證的完整密鑰 kafka-subscription-secret。
    • 指定 --env 選項以新增文字環境變數 BROKERS，並提供 Event Streams 服務實例中服務認證詳細資料所列出其中一個分配管理系統主機的名稱。 不過，如果您想要指定多個分配管理系統主機名稱，請使用 --env BROKERS-broker1,broker2,broker3 格式。
    • (選用) 指定 --min-scale=1 選項，讓應用程式一律具有執行中且不會調整為零的實例。 當您檢視日誌時，將應用程式配置成一律具有執行中實例非常有用。 如果您在正式作業環境中執行，請考量保留應用程式執行中實例的成本，或者您是否想要 Code Engine 自動調整為零。 依預設，不在使用中時，應用程式會調整為零。
        ibmcloud ce app create --name kafka-sender-app --image icr.io/codeengine/kafka-sender --env-from-secret kafka-subscription-secret --env BROKERS=broker-4-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093 --min-scale 1
    ```

  3. 部署此應用程式之後，請執行 app get 指令，以確認應用程式處於 ready 狀態。

    ibmcloud ce app get -n kafka-sender-app

您已建立 kafka-sender-app 應用程式來產生 Code Engine 事件訂閱的 Kafka 訊息，並已建立包含必要認證的 kafka-subscription-secret 密鑰。

設定 Code Engine Kafka 訂閱

若要讓 Code Engine 與 Kafka 事件配合運作，請設定 Code Engine Kafka 事件訂閱，以連線至 Kafka 事件中介伺服器，並監聽 Kafka 事件。 此外，請設定一個 Code Engine 應用程式，用以接收 Kafka 的事件。 Kafka 事件訂閱定義了 Kafka 事件產生者（發送方）與事件接收者（接收方）之間的關係。

Code Engine Kafka 事件訂閱會連線至您的 Kafka 訊息中介伺服器，並針對每則傳入的 Kafka 訊息，向接收端應用程式發送 HTTP POST 請求。 如需進一步了解 Kafka 事件所包含的資訊，請參閱 HTTP 中關於傳送至應用程式的事件的標頭與正文資訊。

  1. 建立 Code Engine 應用程式，以充當 Kafka 訊息的事件消費者，並接收 Kafka 事件。 例如，建立一個稱為 kafka-receiver-app 的應用程式，以使用 icr.io/codeengine/kafka-receiver 映像檔。 此映像檔是從 receiver.go 建置，可從 Samples for IBM Cloud Code Engine GitHub repo取得。 此範例不需要任何環境變數。 您可以選擇性地指定 --min-scale=1 選項，讓應用程式一律具有執行中且不會調整為零的實例。 當您檢視日誌時，將應用程式配置成一律具有執行中實例非常有用。 如果您在正式作業環境中執行，請考量保留應用程式執行中實例的成本，或者您是否想要 Code Engine 自動調整為零。 依預設，不在使用中時，應用程式會調整為零。

    ibmcloud ce app create -n kafka-receiver-app --image icr.io/codeengine/kafka-receiver --min-scale 1

    預設情況下，事件會被路由至目標應用程式的根網址 URL。 您可以使用 --path 選項，將事件傳送至應用程式內的不同目的地。 例如，如果您的訂閱指定 --path /event，則事件會傳送至 https://<base application URL>/events。

  2. 部署此應用程式之後，請執行 app get 指令，以確認應用程式處於 ready 狀態。

    ibmcloud ce app get -n kafka-receiver-app

  3. 使用 ibmcloud ce sub kafka create 指令，為 Kafka 事件建立 Code Engine Kafka 事件訂閱。 使用您先前建立的 kafka-subscription-secret 密鑰來存取訊息分配管理系統。 根據 Kafka 資源的服務認證資訊來指定分配管理系統資訊。 在此範例中，您可以從 ibmcloud resource service-key myeventstream-key 指令的輸出取得分配管理系統資訊。 請注意，您必須為主題的每一個分配管理系統指定 --broker 選項。 --destination 選項指定接收事件的 Code Engine 資源。

    ibmcloud ce sub kafka create --name mykafkasubscription --destination kafka-receiver-app --secret kafka-subscription-secret --topic kafka-topic1 --broker broker-3-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093 --broker broker-5-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093 --broker  broker-0-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093 --broker broker-1-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093 --broker broker-4-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093 --broker broker-2-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093

  4. 顯示 Kafka 事件訂閱的詳細資料。

    ibmcloud ce sub kafka get -n mykafkasubscription

    輸出範例

    Getting Kafka event subscription 'mykafkasubscription'...
OK
Name:          mykafkasubscription
[...]
Destination Type:                 app
Destination:                      kafka-receiver-app
Brokers:
broker-3-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093
broker-5-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093
broker-0-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093
broker-1-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093
broker-4-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093
broker-2-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093
Consumer Group:                   knative-kafka-source-a4072fe1-1dfa-4470-9d07-bf7a0ff8e340
Topics:
kafka-topic1
Secret key reference (user):      kafka-subscription-secret.username
Secret key reference (password):  kafka-subscription-secret.password
Ready:                            true
Conditions:
Type                     OK    Age  Reason
ConnectionEstablished    true  24s
InitialOffsetsCommitted  true  24s
Ready                    true  24s
Scheduled                true  24s
SinkProvided             true  24s
Events:
Type     Reason           Age  Source                  Messages
Normal   FinalizerUpdate  26s  kafkasource-controller  Updated "mykafkasubscription" finalizers

測試訂閱

既然您已建立引用 kafka-receiver-app 應用程式的 Kafka 事件訂閱，請使用 kafka-sender-app 將訊息事件傳送至接收端應用程式。

  1. URL kafka-sender-app 使用 ibmcloud ce app get 並搭配 --output url 選項，以取得目標應用程式的 URL。

    ibmcloud ce app get -n kafka-sender-app --output url

    輸出範例

    https://kafka-sender-app.abcdabcdabc.us-south.codeengine.appdomain.cloud

  2. 執行 Kafka 事件生產者應用程式 kafka-sender-app，以將事件傳送至目的地 Code Engine 應用程式。 使用 curl 呼叫 kafka-sender-app 應用程式，並指定主題及訊息數的值。 請使用 ibmcloud ce app get 來找出您應用程式的公開 URL。 請務必將值包裝成引號內的 curl，以確保將它視為單一字串。 例如，

    curl "<public_URL_of_Kafka_sender_app>?topic=<your_topic_name>&num=<number_of_messages_to_produce>"

    例如，

    curl "https://kafka-sender-app.abcdabcdabc.us-south.codeengine.appdomain.cloud?topic=kafka-topic1&num=1"

  3. 檢視日誌中的事件。 當您使用有效的分配管理系統、主題及存取密鑰建立 Kafka 事件訂閱時，如果您有一個 Kafka 應用程式會產生該主題的訊息 (例如 kafka-sender-app)，則您可以在日誌中看到目的地 Code Engine 應用程式的事件，該應用程式會接收 Kafka 訊息，例如 kafka-receiver-app。 當您使用 Kafka 接收端應用程式 (icr.io/codeengine/kafka-receiver) 時，請在接收端應用程式的日誌中搜尋 Event data，以查看收到的訊息。

    ibmcloud ce app logs -n kafka-receiver-app

    輸出範例

    Getting logs for all instances of application 'kafka-receiver-app'...
OK
kafka-receiver-app-00001-deployment-66976f7988-9xttm/user-container:
2022/03/31 22:19:45 Listening on port 8080
2022/03/31 22:19:46 ----------
2022/03/31 22:19:46 Path: /
2022/03/31 22:19:46 Header: Accept-Encoding=[gzip]
2022/03/31 22:19:46 Header: Ce-Id=[partition:0/offset:167]
2022/03/31 22:19:46 Header: Ce-Source=[/apis/v1/namespaces/glxo4k7nj7d/kafkasources/mykafkasubscription#kafka-topic1]
2022/03/31 22:19:46 Header: Ce-Specversion=[1.0]
2022/03/31 22:19:46 Header: Ce-Subject=[partition:0#167]
2022/03/31 22:19:46 Header: Ce-Time=[2022-03-31T22:19:36.499Z]
2022/03/31 22:19:46 Header: Ce-Type=[dev.knative.kafka.event]
2022/03/31 22:19:46 Header: Content-Length=[8]
2022/03/31 22:19:46 Header: Forwarded=[for=172.30.208.213;proto=http, for=127.0.0.6]
2022/03/31 22:19:46 Header: K-Proxy-Request=[activator]
2022/03/31 22:19:46 Header: Traceparent=[00-b033708685c715a7c2384cdf05797785-65540b0937e9b0ce-00]
2022/03/31 22:19:46 Header: User-Agent=[Go-http-client/1.1]
2022/03/31 22:19:46 Header: X-B3-Parentspanid=[e1a785d7fdbead6c]
2022/03/31 22:19:46 Header: X-B3-Sampled=[1]
2022/03/31 22:19:46 Header: X-B3-Spanid=[abcde9901e6bf83f]
2022/03/31 22:19:46 Header: X-B3-Traceid=[abcde490a426573772fa0bf60caf5ddb]
2022/03/31 22:19:46 Header: X-Envoy-Attempt-Count=[1]
2022/03/31 22:19:46 Header: X-Forwarded-For=[172.30.208.213, 127.0.0.6, 127.0.0.6]
2022/03/31 22:19:46 Header: X-Forwarded-Proto=[http]
2022/03/31 22:19:46 Header: X-Request-Id=[abcdeb4e-c5ac-abcd-abcd-60e6278abcde]
2022/03/31 22:19:46 Event data: test1: 1

    請注意，應用程式的日誌資訊只會持續一小時。 如需檢視應用程式 (或工作) 日誌的相關資訊，請參閱 檢視日誌。

更新訂閱

若要使用 CLI 更新事件訂閱，請使用 ibmcloud ce subscription kafka update 指令。 下列範例會更新主題名稱。

ibmcloud ce sub kafka update -n mykafkasubscription --topic kafka-topic2

您可以使用 ibmcloud ce subscription kafka update 命令來更新「Kafka」訂閱的相關值。 不過，您無法使用這個指令來修改消費者群組的值。 若要更新訂閱以指向不同的主題，請在更新訂閱之前，確認「Kafka」主題確實存在。

清除 Kafka 訂閱指導教學

準備好刪除 Kafka 訂閱、傳送及接收應用程式，以及密碼嗎? 您可以使用 ibmcloud ce app delete、ibmcloud ce sub kafka delete 及 ibmcloud ce sub kafka delete 指令。 您可以選擇性地使用 -f 選項來強制刪除元件，而不進行確認。

當您刪除「Kafka」訂閱時，此刪除操作並不會刪除該訂閱所引用的應用程式。

若要移除您的訂閱，請執行下列:

ibmcloud ce sub kafka delete --name mykafkasubscription -f

若要移除 Kafka 訊息接收應用程式，

ibmcloud ce app delete --name kafka-receiver-app -f

同樣地，您可以移除 kafka-sender-app。

ibmcloud ce app delete --name kafka-sender-app -f

若要移除 kafka-subscription-secret，請執行下列:

ibmcloud ce secret delete --name kafka-subscription-secret -f

準備好刪除 Event Streams 服務實例的服務實例嗎? --recursive 選項指定移除服務實例的所有資源，其中包括相關聯的服務金鑰。

ibmcloud resource service-instance-delete myeventstream --recursive -f

下一步

如需使用 Kafka 事件訂閱的相關資訊，請參閱 使用 Kafka 事件生產者。

尋找更多程式碼範例? 請參閱 IBM Cloud Code Engine GitHub 儲存庫中的範例。