Nodejs Nestjs Kafka消息队列插件kafkajs的使用

Nodejs Nestjs Kafka消息队列插件kafkajs的使用
NestJS 是一个用于构建高效、可扩展的服务器端应用程序的框架，而 Kafka 是一个分布式流处理平台。在 NestJS 中使用 Kafka 可以通过 kafkajs 插件来实现。以下是一个简单的示例，展示如何在 NestJS 项目中设置和使用 Kafka。

1. 安装必要的依赖

首先，你需要安装 @nestjs-modules/microservices 和 kafkajs：

npm install @nestjs/microservices kafkajs

2. 配置 Kafka 模块

在你的 NestJS 应用程序中创建一个 Kafka 模块，并配置它：

import { Module } from '@nestjs/common';
import { ClientsModule, Transport } from '@nestjs/microservices';

@Module({
  imports: [
    ClientsModule.register([
      {
        name: 'KAFKA_SERVICE',
        transport: Transport.KAFKA,
        options: {
          client: {
            brokers: ['localhost:9092'], // Kafka 服务器地址
          },
          consumer: {
            groupId: 'nest-group', // 消费者组 ID
          },
        },
      },
    ]),
  ],
})
export class KafkaModule {}

3. 创建 Kafka 服务

接下来，创建一个服务来处理 Kafka 的消息：

import { Injectable } from '@nestjs/common';
import { ClientKafka } from '@nestjs/microservices';

@Injectable()
export class KafkaService {
  constructor(private readonly client: ClientKafka) {}

  async onModuleInit() {
    this.client.subscribeToResponseOf('hello')
      .subscribe(data => console.log('Received:', data));
  }

  sendMessage(message: string) {
    this.client.emit('hello', message);
  }
}

4. 在控制器中使用 Kafka 服务

最后，在控制器中注入并使用 Kafka 服务：

import { Controller, Get } from '@nestjs/common';
import { KafkaService } from './kafka.service';

@Controller()
export class AppController {
  constructor(private readonly kafkaService: KafkaService) {}

  @Get('send-message')
  async sendMessage() {
    this.kafkaService.sendMessage('Hello, Kafka!');
    return { message: 'Message sent' };
  }
}

5. 启动应用

现在你可以启动你的 NestJS 应用了。如果你发送一个 GET 请求到 /send-message，你应该能在控制台看到接收到的消息。

npm run start

注意事项

  • 确保 Kafka 服务器正在运行并且可以在指定的地址上访问。
  • 你可能需要调整 Kafka 的配置选项，如消费者组 ID、主题等，以适应你的实际需求。
  • 对于生产环境，建议使用更复杂的错误处理和重试机制。

以上就是一个基本的 NestJS + Kafka 的集成示例。希望这对你有所帮助！


3 回复

当然，NestJS 和 Kafka 结合起来可以构建非常强大的实时应用。KafkaJS 是一个很受欢迎的 Kafka 客户端库。首先，你需要安装 @nestjs/microservices 和 kafkajs：

npm install @nestjs/microservices kafkajs

然后，在你的 NestJS 模块中配置 Kafka：

import { Module } from '@nestjs/common';
import { ClientsModule, Transport } from '@nestjs/microservices';

@Module({
  imports: [
    ClientsModule.register([
      {
        name: 'KAFKA_SERVICE',
        transport: Transport.KAFKA,
        options: {
          client: {
            brokers: ['localhost:9092'],
            connectionTimeout: 3000,
          },
          consumer: {
            groupId: 'nest-consumer-group',
          },
        },
      },
    ]),
  ],
})
export class AppModule {}

接着，你可以创建一个服务来处理 Kafka 消息：

import { MessagePattern } from '@nestjs/microservices';
import { Controller } from '@nestjs/common';

@Controller()
export class KafkaController {
  @MessagePattern({ cmd: 'greeting' })
  getGreeting(data: any) {
    console.log('Received:', data);
    return `Hello ${data.name}`;
  }
}

最后，别忘了在你的模块中引入这个控制器。

祝你在 Kafka 的世界里愉快地跳跃！如果需要更多的蹦跶技巧，随时回来找我！


NestJS 是一个用于构建高效、可扩展的服务器端应用程序的框架，它基于 TypeScript 和 JavaScript。Kafka 是一种高吞吐量、分布式发布-订阅消息系统。kafkajs 是一个用于 Node.js 的 Kafka 客户端库。结合 NestJS 和 kafkajs 可以帮助您实现一个高效的消息驱动系统。

下面我将为您展示如何在 NestJS 项目中配置和使用 kafkajs 插件来创建一个简单的生产者和消费者。

步骤 1: 初始化 NestJS 项目

首先，确保您的机器上安装了 Node.js 和 npm。然后运行以下命令初始化一个新的 NestJS 项目：

npm i -g @nestjs/cli
nest new nest-kafka-app
cd nest-kafka-app

步骤 2: 安装必要的依赖

接下来，我们需要安装 NestJS Kafka 模块以及 kafkajs 库：

npm install @nestjs/microservices kafkajs

步骤 3: 配置 Kafka 客户端

在 src/main.ts 中配置 Kafka 模块。这里我们为 Kafka 创建了一个模块，并将其注入到应用模块中。

import { NestFactory } from '@nestjs/core';
import { Transport } from '@nestjs/microservices';
import { AppModule } from './app.module';

async function bootstrap() {
  const app = await NestFactory.createMicroservice(AppModule, {
    transport: Transport.KAFKA,
    options: {
      client: {
        brokers: ['localhost:9092'], // 根据实际情况修改
      },
      consumer: {
        groupId: 'test-consumer-group',
      },
    },
  });
  await app.listen();
}
bootstrap();

步骤 4: 创建消费者

在 src/app.controller.ts 中添加消费者逻辑：

import { Controller, MessagePattern } from '@nestjs/common';
import { ClientKafka } from '@nestjs/microservices';

@Controller()
export class AppController {
  constructor(private readonly client: ClientKafka) {
    this.client.connect();
  }

  @MessagePattern({ cmd: 'sayHello' })
  async sayHello(data: any) {
    console.log(`Received message: ${data.name}`);
    return `Hello, ${data.name}`;
  }
}

步骤 5: 创建生产者

在 src/app.service.ts 中创建一个生产者服务：

import { Injectable } from '@nestjs/common';
import { ClientKafka } from '@nestjs/microservices';

@Injectable()
export class AppService {
  constructor(private readonly client: ClientKafka) {}

  async sendMessage(name: string) {
    this.client.emit('sayHello', { name });
  }
}

测试

您可以使用 NestJS 提供的 CLI 命令启动应用程序：

npm run start:dev

然后，在另一个终端窗口中调用生产者的 sendMessage 方法进行测试：

(async () => {
  const app = await NestFactory.createApplicationContext(AppModule);
  const service = app.get(AppService);
  await service.sendMessage('World');
})();

以上就是使用 NestJS 和 kafkajs 创建一个简单的 Kafka 生产者和消费者的示例。希望这可以帮助您开始构建自己的消息驱动系统。

NestJS结合Kafka使用时，首先安装@nestjs-modules/microservices和kafkajs。然后，在模块中配置Kafka客户端：

import { Module } from '@nestjs/common';
import { ClientsModule, Transport } from '@nestjs/microservices';

@Module({
  imports: [
    ClientsModule.register([
      {
        name: 'KAFKA_CLIENT',
        transport: Transport.KAFKA,
        options: {
          client: {
            brokers: ['localhost:9092'],
          },
          consumer: {
            groupId: 'nest-group',
          },
        },
      },
    ]),
  ],
})
export class AppModule {}

处理消息时，可以使用@MessagePattern或@EventPattern装饰器：

import { MessagePattern } from '@nestjs/microservices';

@Controller()
export class AppController {
  @MessagePattern({ cmd: 'hello' })
  getHello(data: any): string {
    return `Hello ${data.name}`;
  }
}

这样就完成了基本的NestJS与Kafka的集成。

回到顶部