如何实现NestJS仅在特定请求头值下消费Kafka EventPattern

NestJS官方的@EventPattern装饰器并没有直接支持头部过滤的配置，但可以通过两种方式实现你想要的效果：

方案一：手动在消费方法内校验头部 #

这是最直接的方式，通过@Headers()装饰器获取Kafka消息的头部，然后在方法内判断是否符合条件：

import { Controller, EventPattern, Payload, Headers } from '@nestjs/common';

@Controller()
export class KafkaConsumerController {
  @EventPattern('my-topic')
  async handleMyTopicEvent(
    @Payload() data: any,
    @Headers() headers: Record<string, string>,
  ) {
    // 校验指定头部的值
    if (headers['my-header-variable'] !== 'secret') {
      // 不符合条件则直接返回，不执行后续消费逻辑
      return;
    }

    // 这里编写你的业务逻辑
    console.log('处理符合条件的消息:', data);
  }
}

方案二：自定义装饰器+守卫实现复用性过滤 #

如果多个消费方法都需要类似的头部校验，可以抽离逻辑到自定义装饰器和守卫中，实现类似你期望的@MyEventPattern语法：

1. 创建自定义装饰器 #

用于绑定主题和头部过滤规则：

import { SetMetadata } from '@nestjs/common';
import { EventPattern } from '@nestjs/microservices';

export const HEADER_FILTER_KEY = 'header_filter';

export const MyEventPattern = (topic: string, headerRules: Record<string, string>) => {
  return (target: any, propertyKey: string, descriptor: PropertyDescriptor) => {
    // 存储头部校验规则到元数据
    Reflect.defineMetadata(HEADER_FILTER_KEY, headerRules, descriptor.value);
    // 绑定原有的EventPattern装饰器
    EventPattern(topic)(target, propertyKey, descriptor);
  };
};

2. 创建头部过滤守卫 #

在消费前自动校验头部规则：

import { CanActivate, ExecutionContext, Injectable } from '@nestjs/common';
import { Reflector } from '@nestjs/core';
import { KafkaContext } from '@nestjs/microservices';
import { HEADER_FILTER_KEY } from './my-event-pattern.decorator';

@Injectable()
export class HeaderFilterGuard implements CanActivate {
  constructor(private readonly reflector: Reflector) {}

  canActivate(context: ExecutionContext): boolean {
    // 获取当前方法的头部校验规则
    const headerRules = this.reflector.get<Record<string, string>>(
      HEADER_FILTER_KEY,
      context.getHandler(),
    );

    if (!headerRules) {
      // 没有配置规则则直接允许执行
      return true;
    }

    // 从Kafka上下文获取消息头部
    const kafkaContext = context.switchToRpc().getContext<KafkaContext>();
    const messageHeaders = kafkaContext.getMessage().headers as Record<string, string>;

    // 校验所有头部规则是否匹配
    return Object.entries(headerRules).every(
      ([key, expectedValue]) => messageHeaders[key] === expectedValue,
    );
  }
}

3. 使用自定义装饰器和守卫 #

可以全局注册守卫（对所有消费方法生效），或者局部在控制器/方法上使用：

全局注册守卫（在AppModule中）

import { Module } from '@nestjs/common';
import { APP_GUARD } from '@nestjs/core';
import { HeaderFilterGuard } from './header-filter.guard';

@Module({
  providers: [
    {
      provide: APP_GUARD,
      useClass: HeaderFilterGuard,
    },
  ],
})
export class AppModule {}

局部使用示例

import { Controller, UseGuards } from '@nestjs/common';
import { MyEventPattern } from './my-event-pattern.decorator';
import { HeaderFilterGuard } from './header-filter.guard';

@Controller()
@UseGuards(HeaderFilterGuard)
export class KafkaConsumerController {
  @MyEventPattern('my-topic', { 'my-header-variable': 'secret' })
  async handleMyTopicEvent(@Payload() data: any) {
    // 仅当头部符合规则时才会执行这里的逻辑
    console.log('处理符合条件的消息:', data);
  }
}

方案对比 #

• 方案一适合单个消费方法的简单场景，代码量少，无需额外配置。
• 方案二更适合多个方法需要相同/类似头部校验的场景，逻辑复用性强，代码更整洁。

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