[Kafka] Nest.js 로 kafka 통신 테스트 해보기

오늘은 Nest.js 로 kafka 통신 테스트를 진행해 보겠다. 진행해보기에 앞서 docker-compose로 yml 파일을 사용할 것이기 때문에 "[Kafka] Docker Compose - Single Broker 사용으로 통신해보기" 를 참고하면 좋다.

 

 

* Producer: 메시지 생성에 대한 책임

* Consumer: 메시지 소비에 대한 책임

* Connector: Producer/Cosumer API 및 link topics 2가지 모두를 "재사용" 가능 

* Streams: 입력을 출력 결과로 전환하는 책임

* Admin: Kafka topics를 관리하는데 사용

 

 

 

---

 

 

 

Test Start!!

 

1) nest js CLI install

npm i -g @nestjs/cli

 

 

2) Producer API를 위한 nest.js 프로젝트 생성 

- 개인적으로 npm을 선호하기 때문에 npm 사용 yarn을 사용해도 무관하다.

nest new producer

 

 

3) 생성된 프로젝트에 kafkajs 와 microservies 설치해준다.

npm i --save kafkajs

npm i --save @nestjs/microservices

 

 

 

Producer 프로젝트

4) app.module.ts 파일에서 모듈을 설정해준다.

import { Module } from '@nestjs/common';
import { AppController } from './app.controller';
import { AppService } from './app.service';
import { ClientsModule, Transport } from '@nestjs/microservices';
 
@Module({
  imports: [
    ClientsModule.register([
      {
        name: 'name_jung',
        transport: Transport.KAFKA,
        options: {
          client: {
            clientId: 'client_id_jung',
            brokers: ['localhost:29092'],
          },
        },
      },
    ]),
  ],
 
  controllers: [AppController],
  providers: [AppService],
})
export class AppModule {}
 

 

위의 brokers: ['localhost:29092']는 docker-compose.yml 에 설정한 포트이다. 실서비스에서 사용할때는 .env로 관리하는걸 추천한다.

 

 

5) Controller에서 @Get('kafka-test')를 추가해준다.

import { Controller, Get, Inject } from '@nestjs/common';
import { AppService } from './app.service';
import { ClientKafka } from '@nestjs/microservices';
 
@Controller()
export class AppController {
  constructor(
    private readonly appService: AppService,
    @Inject('name_jung') private readonly client: ClientKafka,
  ) {}
 
  @Get()
  getHello(): string {
    return this.appService.getHello();
  }
 
  @Get('kafka-test')
  testKafka() {
    return this.client.emit('medium.rock', {
      foo: 'bar',
      data: new Date().toString(),
      str: 'kafka-test',
    });
  }
 
}

 

 

6) localhost:300/kafka-test 를 Get 쿼리를 날리면 Controller의 'medium.rock' 항목에 JSON 메시지를 내보내려고 한다. 

curl --location --request GET 'http://localhost:3000/kafka-test'

 

curl을 사용해서 query를 날리면 응답이 오는것을 확인 할 수 있다.

 

 

 

Consumer 프로젝트

7) Producer 프로젝트를 만들었을 때와 마찬가지로 cli를 사용해준다.

nest new consumer

 

 

8) 생성된 프로젝트에 kafkajs 와 microservies 설치해준다.

npm i --save kafkajs

npm i --save @nestjs/microservices

 

 

9) app.module.ts 에 아래와 같이 코드를 입력해준다.

import { Module } from '@nestjs/common';
import { AppController } from './app.controller';
import { AppService } from './app.service';
import { ClientsModule, Transport } from '@nestjs/microservices';
 
@Module({
  imports: [
    ClientsModule.register([
      {
        name: 'name_jung',
        transport: Transport.KAFKA,
        options: {
          client: {
            clientId: 'client_id_jung',
            brokers: ['localhost:29092'],
          },
          consumer: {
            groupId: 'unique_id_jung',
          },
        },
      },
    ]),
  ],
 
  controllers: [AppController],
  providers: [AppService],
})
export class AppModule {}
 

 

 

10) main.ts 에 아래와 같이 코드를 입력해준다.

import { NestFactory } from '@nestjs/core';
import { AppModule } from './app.module';
import {
  KafkaOptions,
  MicroserviceOptions,
  Transport,
} from '@nestjs/microservices';
 
async function bootstrap() {
  const microserviceOptions: KafkaOptions = {
    transport: Transport.KAFKA,
    options: {
      client: {
        brokers: ['localhost:29092'],
      },
    },
  };
 
  const app = await NestFactory.createMicroservice<MicroserviceOptions>(
    AppModule,
    microserviceOptions,
  );
  await app.listen();
}
bootstrap();

 

 

11) app.controller.ts 에 아래와 같이 코드를 입력해준다.

import { Controller } from '@nestjs/common';
import {
  Ctx,
  KafkaContext,
  MessagePattern,
  Payload,
} from '@nestjs/microservices';
 
@Controller()
export class AppController {
  constructor() {}
 
  @MessagePattern('medium.rock')
  readMessage(@Payload() message: any, @Ctx() context: KafkaContext) {
    const originalMessage = context.getMessage();
    const response =
      'Receiving a new message from topic : ' +
      JSON.stringify(originalMessage.value);
    console.log(response);
 
    return response;
  }
}

 

 

12) Producer/Consumer 프로젝트를 npm run start:dev 를 해준다. 그리고 나서 curl 을 사용해서 query를 날려준다.

그러면 cmd > producer에게 날린 쿼리가 consumer에서 출력되는것을 확인할  수 있다.

 

 

13) Producer 의 app.controller를 다음과 같이 수정해준다.

import { Controller, Get, Inject } from '@nestjs/common';
import { AppService } from './app.service';
import { ClientKafka } from '@nestjs/microservices';
 
@Controller()
export class AppController {
  constructor(
    private readonly appService: AppService,
    @Inject('name_jung') private readonly client: ClientKafka,
  ) {}
 
  async onModuleInit() {
    ['medium.rock'].forEach((key: string) =>
      this.client.subscribeToResponseOf(`${key}`),
    );
 
    await this.client.connect();
  }
 
  async onModuleDestroy() {
    await this.client.close();
  }
 
  @Get()
  getHello(): string {
    return this.appService.getHello();
  }
 
  @Get('kafka-test')
  testKafka() {
    return this.client.emit('medium.rock', {
      foo: 'bar',
      data: new Date().toString(),
      str: 'kafka-test',
    });
  }
 
  @Get('kafka-test-with-response')
  testKafkaWithResponse() {
    return this.client.send('medium.rock', {
      foo: 'bar',
      data: new Date().toString(),
      str: 'kafka-test-with-response',
    });
  }
}

 

 

14)  curl을 이용해서 새로 추가된 'kafka-test-with-response'를 사용해본다.

curl --location --request GET 'http://localhost:3000/kafka-test-with-response'

 

그러면 Conumer 프로젝트에서는 다음과 같이 출력된다.

 

 

 

 

 

 

 

 

reference

* Marcos Henrique da Silva - An Introduction to Kafka with TypeScript using NestJS