Azure Function v4本地开发：用Kafka替代Event Hub的配置方法问询

我有一个由Event Hub触发的Azure Function，由于Event Hub无法本地模拟，计划本地开发阶段使用Kafka，部署至云端后切换为Event Hub。但我无法掌握如何在Azure Function v4的TypeScript项目中正确配置Kafka，查阅TypeScript接口未发现相关内容，且线上示例均针对Azure Function v3版本，特此寻求可行的实现方案。

我的docker-compose配置 #

version: '3.4'
services:
  storage:
    container_name: storage
    image: mcr.microsoft.com/azure-storage/azurite
    ports:
       - "10000:10000"
       - "10001:10001"
       - "10002:10002"
    volumes:
       - vol_storage:/data

  zookeeper:
    image: bitnami/zookeeper:3.9.2
    container_name: zookeeper
    environment:
      ALLOW_ANONYMOUS_LOGIN: yes
      ZOO_PORT_NUMBER: 2181
    restart: unless-stopped

  kafka:
    image: bitnami/kafka:3.7.1
    container_name: kafka
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_CLUSTER_NAME: cluster_dev
      KAFKA_CFG_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_CFG_LISTENERS: INTERNAL://:19092,EXTERNAL://:9092
      KAFKA_CFG_ADVERTISED_LISTENERS: INTERNAL://kafka:19092,EXTERNAL://20.64.115.37:9092
      KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP: INTERNAL:PLAINTEXT,EXTERNAL:SASL_PLAINTEXT
      KAFKA_CFG_SASL_ENABLED_MECHANISMS: PLAIN
      KAFKA_CFG_INTER_BROKER_LISTENER_NAME: INTERNAL
      KAFKA_CFG_SASL_MECHANISM_INTER_BROKER_PROTOCOL: PLAIN
      KAFKA_CLIENT_USERS: user
      KAFKA_CLIENT_PASSWORDS: password
    ports:
      - 9092:9092
    depends_on:
      - zookeeper
    restart: unless-stopped

  kafka-ui:
    image: provectuslabs/kafka-ui:latest
    container_name: kafka-ui
    depends_on:
      - kafka
    ports:
      - 8090:8080
    environment:
      KAFKA_CLUSTERS_0_NAME: cluster_dev
      KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka:19092
      KAFKA_CLUSTERS_0_ZOOKEEPER: zookeeper:2181
      AUTH_TYPE: "LOGIN_FORM"
      SPRING_SECURITY_USER_NAME: user
      SPRING_SECURITY_USER_PASSWORD: password

volumes:
  vol_cosmos:
  vol_storage:

我的Azure Function代码 #

import {
  type InvocationContext,
  app
} from '@azure/functions';
import { Config } from '../config/index.js';


export async function messageLogged (
  message: string,
  context: InvocationContext
): Promise<void> {
  context.log(message);
}

/**
 * 开发阶段不连接Event Hub，改用Kafka
 */
if (!Config.isLocalDev) {
  app.eventHub('messageLogged', {
    connection: '',
    eventHubName: Config.eventHub.topic,
    cardinality: Config.eventHub.cardinality,
    consumerGroup: Config.eventHub.consumerGroup,
    handler: messageLogged
  });
} else {
  /**
   * 如何连接Kafka？
   */
}

Azure Functions v4的TypeScript编程模型中，Kafka触发器可通过app.kafka()方法配置，以下是适配你现有环境的实现步骤：

1. 配置Kafka触发器 #

在代码的else块中添加Kafka触发器配置，结合你docker-compose里的SASL PLAIN认证信息，直接复用已有的消息处理函数：

else {
  app.kafka('messageLoggedKafka', {
    // Kafka broker地址，对应docker-compose中配置的EXTERNAL监听地址
    brokerList: '20.64.115.37:9092',
    // 消费的topic名称，与Event Hub的topic保持一致即可
    topicName: Config.eventHub.topic,
    // 消费者组，与Event Hub的consumerGroup保持一致
    consumerGroup: Config.eventHub.consumerGroup,
    // 认证机制，匹配Kafka配置的SASL_PLAINTEXT
    authenticationMechanism: 'plain',
    // SASL认证用户名
    username: 'user',
    // SASL认证密码
    password: 'password',
    // 复用已有的消息处理函数
    handler: messageLogged
  });
}

2. 类型定义补充（可选） #

如果TypeScript提示类型缺失，确保安装的@azure/functions为v4.x最新版本，新版本已包含Kafka触发器的完整类型定义。若仍有问题，可手动引入类型：

import type { KafkaTriggerOptions } from '@azure/functions';

3. 本地开发验证事项 #

• 确保Kafka容器正常运行，通过kafka-ui（http://localhost:8090）确认目标topic已创建
• 检查local.settings.json中存储连接字符串配置正确：

{
  "IsEncrypted": false,
  "Values": {
    "AzureWebJobsStorage": "UseDevelopmentStorage=true",
    "FUNCTIONS_WORKER_RUNTIME": "node"
  }
}

• 向Kafka的目标topic发送测试消息，验证Function是否能正常消费并打印日志

4. 云端部署切换 #

部署至云端时，Config.isLocalDev会自动切换为false，Function将自动使用Event Hub触发器，无需额外修改代码。

内容的提问来源于stack exchange，提问作者thedethfox