【Azure 应用服务】本地创建Azure Function Kafka Trigger 函数和Kafka output的HTTP Trigger函数实验

【Azure 应用服务】本地创建Azure Function Kafka Trigger 函数和Kafka output的HTTP Trigger函数实验

问题描述

在上一篇博文(​​Studio 2022创建Azure Function 作为生产者和消费者在本地进行验证

生产者：使用HTTP Trigger函数，以 kafka output 作为输出消费者：使用Kafka Trigger函数

解题步骤

1：打开VS 2022，开始创建Azure Funciton工程

2：选择 Azure Function模板，并使用.NET 6.0作为运行时，然后选择 Kafka Trigger。其他值保持默认即可。保存。

3:   把BorkerList添加到本地配置文件中（ local.settings.json ），然后修改正确的topic名称。因为Kafka服务器没有启用SSL和Password，所以这里 Protocol 和 AuthenticationMode 都需要修改为 NotSet。

local.setting.json 配置文件：

{ "IsEncrypted": false, "Values": { "AzureWebJobsStorage": "UseDevelopmentStorage=true", "FUNCTIONS_WORKER_RUNTIME": "dotnet", "BrokerList": "xxx.xxx.xxx.xxx:9092", "KafkaPassword": "", "ConnectionString": "" }}

KafkaTrigger Function代码：

using Microsoft.Azure.WebJobs;using Microsoft.Azure.WebJobs.Extensions.Kafka;using Microsoft.Extensions.Logging;namespace FunctionApp2{ public class Function1 { // KafkaTrigger sample // Consume the message from "topic" on the LocalBroker. // Add `BrokerList` and `KafkaPassword` to the local.settings.json // For EventHubs // "BrokerList": "{EVENT_HUBS_NAMESPACE}.servicebus.windows.net:9093" // "KafkaPassword":"{EVENT_HUBS_CONNECTION_STRING} [FunctionName("Function1")] public void Run( [KafkaTrigger("BrokerList", "test_topic", Username = "$ConnectionString", Password = "%KafkaPassword%", Protocol = BrokerProtocol.NotSet, AuthenticationMode = BrokerAuthenticationMode.NotSet, ConsumerGroup = "$Default")] KafkaEventData[] events, ILogger log) { foreach (KafkaEventData eventData in events) { log.LogInformation($"C# Kafka trigger function processed a message: {eventData.Value}"); } } }}

4：同样，继续添加一个 Kafka output 的Function, （与第二步相同）。其他值保持默认即可。保存。

5：与第三步相同，修改正确的topic名称。因为Kafka服务器没有启用SSL和Password，所以这里 Protocol 和 AuthenticationMode 都需要修改为 NotSet。

using Microsoft.AspNetCore.Http;using Microsoft.AspNetCore.Mvc;using Microsoft.Azure.WebJobs;using Microsoft.Azure.WebJobs.Extensions.Http;using Microsoft.Azure.WebJobs.Extensions.Kafka;using Microsoft.Extensions.Logging;namespace FunctionApp2{ public class Function2 { // KafkaOutputBinding sample // This KafkaOutput binding will create a my_topic "my_topic" on the LocalBroker if it doesn't exists. // Call this function then the KafkaTrigger will be trigged. [FunctionName("Function2")] public IActionResult Output( [HttpTrigger(AuthorizationLevel.Anonymous, "get", Route = null)] HttpRequest req, [Kafka("BrokerList", "test_topic", Username = "$ConnectionString", Password = "%KafkaPassword%", Protocol = BrokerProtocol.NotSet, AuthenticationMode = BrokerAuthenticationMode.NotSet )] out string eventData, ILogger log) { log.LogInformation("C# HTTP trigger function processed a request."); string message = req.Query["message"]; string responseMessage = string.IsNullOrEmpty(message) ? "This HTTP triggered function executed successfully. Pass a message in the query string" : $"Message {message} sent to the broker. This HTTP triggered function executed successfully."; eventData = $"Received message: {message}"; return new OkObjectResult(responseMessage); } }}

​​​​

6：F5运行Function Project，使用HTTP Trigger的URL发送消息，然后用Kafka Trigger的函数接受消息。

整个步骤的示例动画：

参考文档

适用于 Azure Functions 的 Apache Kafka 绑定概述：​​云中，恰是如此!

分类: ​​【Azure 应用服务】​​

标签: ​​App Service​​, ​​Azure Developer​​, ​​Azure Function​​, ​​Kafka Trigger​​, ​​Kafka output​​

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。