Azure独立进程模式下KafkaTrigger如何获取Kafka消息头部

解决方案 #

1. 升级NuGet包 #

你当前使用的Microsoft.Azure.Functions.Worker.Extensions.Kafka 3.2.1版本尚未原生支持带Headers的消息类型绑定，先将该包升级到3.3.0及以上稳定版本，这是官方支持Headers读取的最低版本要求。

2. 修改触发器参数绑定类型 #

不要直接将消息绑定为string类型，改为绑定为Microsoft.Azure.Functions.Worker.Extensions.Kafka命名空间下的KafkaEventData<T>类型，该类型内置Headers属性可直接读取消息头部。

修改后的代码示例：

using Microsoft.Azure.Functions.Worker;
using Microsoft.Azure.Functions.Worker.Extensions.Kafka;
using System.Text;

namespace Func
{
    public class Functions
    {
        [Function("KafkaFunction")]
        public static void Run([KafkaTrigger("broker", "topic",
            ConsumerGroup = "consumer", AuthenticationMode = BrokerAuthenticationMode.Plain)]
            KafkaEventData<string> kafkaEvent, FunctionContext ctx)
        {
            // 直接读取Headers集合
            var headers = kafkaEvent.Headers;
            
            // 遍历获取具体Header值示例（Value为byte数组，按需转换为对应编码的字符串）
            foreach (var header in headers)
            {
                string headerValue = Encoding.UTF8.GetString(header.Value.Span);
                // 自定义头部处理逻辑
            }

            // 原有消息属性依然可以正常访问
            var offset = kafkaEvent.Offset;
            var partition = kafkaEvent.Partition;
            var msgContent = kafkaEvent.Value;
        }
    }
}

如果需要批量消费，也可以将参数改为KafkaEventData<string>[] events，和进程内模式的用法基本一致，仅命名空间不同。

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