Nest.js 如何将 Kafka 消费者与 WebSocket 一起使用答案

【问题标题】：Nest.js How to use Kafka consumer with WebSocketNest.js 如何将 Kafka 消费者与 WebSocket 一起使用
【发布时间】：2021-09-09 00:53:22
【问题描述】：

我在 nest.js 中有一个微服务，我有一个 kafka 消费者。我需要在我的消费者在@MessagePatter（在控制器中）中侦听新消息之后，它会发出一个套接字事件。我有一个@WebSocketServer ()，但是如果我将我的套接字服务器注入控制器，这不会生成消息。有什么想法，在nest.js中收到带有kafka的消息后如何使用socket发出消息？

我的网络套接字

    import {
      SubscribeMessage,
      WebSocketGateway,
      OnGatewayInit,
      WebSocketServer,
      OnGatewayConnection,
      OnGatewayDisconnect,
      MessageBody,
      ConnectedSocket,
    } from '@nestjs/websockets';
    import { Logger } from '@nestjs/common';
    import { Socket, Server } from 'socket.io';


    @WebSocketGateway(89, { transports: ['websocket'] })
    export class EventsGatewayCon
      implements OnGatewayInit, OnGatewayConnection, OnGatewayDisconnect
    {
      @WebSocketServer() server: Server;

      private logger: Logger = new Logger('EventsGateway');

      @SubscribeMessage('testSocket')
      async testSocket(
        @ConnectedSocket() client: Socket,
        @MessageBody() payload: any,
      ): Promise<void> { 
        console.log(payload);
      }

      afterInit(server: Server) {
        this.logger.log('Init');
      }

      handleDisconnect(client: Socket) {
        this.logger.log(`Client disconnected: ${client.id}`);
      }

      handleConnection(client: Socket, ...args: any[]) {
        let theUid = null;
        if (client.handshake.query['isClient']) {
          theUid = client.handshake.query['uidClient'];
       this.server.sockets.emit('testSocket', 'new message');
        client.join(theUid);
        this.logger.log(`Client connected: ${client.id}`);
      }
    }

控制器。收听kafka消息

 import { Controller } from '@nestjs/common';
import { MessagePattern, Payload } from '@nestjs/microservices';
import { AppService } from './app.service'; 
import { EventsGatewayCon } from './events/events.gateway';

@Controller()
export class AppController {
  constructor(
    private readonly appService: AppService,
    private readonly event: EventsGatewayCon,
  ) {}

  @MessagePattern('myListener') // Our topic name
  async getTest(@Payload() message) {
    console.log(message.value);
    this.event.server.sockets.emit('testSocket', 'send new messa');
    return 'I am ApiKafka'; //return the message but dont emit the message with socket
  }
}

应用模块

import { Module } from '@nestjs/common'; 
import { EventsController } from './events.controller';
import { EventsGatewayCon } from './events.gateway';

@Module({
  imports: [],
  controllers: [EventsController],
  providers: [EventsGatewayCon],
  exports: [EventsGatewayCon],
})
export class EventsModule {}

【问题讨论】：

我在这里没有看到任何 Kafka 消费者。换句话说，你在哪里配置引导地址？

标签： javascript node.js apache-kafka socket.io microservices

【解决方案1】：

检查您的 Kafka 微服务实施。

应该是这样的：

main.ts

app.connectMicroservice({
    transport: Transport.KAFKA,
    options: {
      client: {
        clientId: 'KAFKA_CLIENT_ID',
        brokers: ['KAFKA_URIS'],
      },
    },
  });

  await app.startAllMicroservices();
  await app
    .listen(port))
    .then(() => {
      Logger.log(`API Listen on ${port}`);
    })
    .catch((error: any) => Logger.error(error));

【讨论】：