알림서비스를 위해 웹소켓 통신을 rabbitMQ로 대체하기

전제 조건

- 웹소켓 서버에 알림을 발송하는 REST API가 존재함

- 여러 개의 클라이언트와 웹소켓 서버가 연결되어 있음

- 백엔드 서버도 클라이언트와 같은 방식으로 웹소켓 서버와 연결되어 있음

 

웹소켓은 양방향 통신이 가능한 프로토콜.

실시간성이 엄청 중요한 알림이 아니었기 때문에 굳이 웹소켓을 사용할 필요는 없다.

 

알림을 발송할 때 필요한 것

1. (타겟 유저의 클라이언트가 켜져 있다면) 유저 클라이언트에게 알림을 보낸다.

2. 데이터베이스에 알림 이력을 쌓는다.

 

이러나 저러나 알림 이력을 쌓을거라면 무조건 알림도 보내고 이력도 쌓으면 되지 않나?

어차피 API는 이미 만들어져 있으니.. 클라이언트가 꺼져 있다면 알림은 도착하지 않고 재로그인시 n건의 알림이 있다고 알려줄 거니까.

 

어떤 이유로 잠시 웹소켓이 꺼져버렸고 클라이언트는 계속 켜져 있다고 해 보자.

이 때 REST API를 통해 웹소켓 서버에서 클라이언트에 알림을 보낸다면 Timeout이 날 것이다.

나중에 다시 웹소켓이 복구됐다고 해도 이전에 날렸던 API의 응답을 받기까지 렉이 걸릴 수도 있고..

불가능한 건 절대 아니지만 HttpRequest를 비동기식으로 처리한다면 크게 문제되지는 않는다.

그리고 REST API의 특성상 요청을 보내고 나서 응답을 받는데, 단순 알림 발송이므로 굳이 응답을 받을 필요도 없다.

 

물론 있는 걸 이용한다는 건 좋지만, 효용에 비해 너무 큰 일이 될 것임!

 

대신, 메세지큐를 이용하면 서버마다의 의존성을 줄일 수 있다.

또한 중간에 연결이 끊겼을 때도 메세지큐는 consume되지만 않으면 계속 쌓아 둘 수 있어 데이터 유실도 적다.

그러니까 rabbitMQ 서버라는 완충재를 백엔드 서버와 웹소켓 서버 사이에 두는 게 낫다.

 

설치한 라이브러리는 amqplib (typescript를 사용하므로 이것도 설치)

 

대충 이런 구조

 

웹소켓 서버에서 rabbitMQ 연결

import { connect, Channel } from "amqplib";
import { throttle } from "lodash-es";
import { RABBITMQ_RECONNECT_INTERVAL } from "./constants.js";

const connectRabbitMQ = throttle(async () => {
  try {
    const connection = await connect({
      username: userName,
      password: password,
      hostname: host,
    });

    channel = await connection.createChannel();
    await readPreviousLogs();
  } catch (e) {
    console.log("rabbitMQ 연결 실패");
  }
}, RABBITMQ_RECONNECT_INTERVAL);

 

웹소켓 내부에서 메시지를 보내야할 경우를 대비한 함수

const sendToQueue = async ({ queueName, message }: ISendToQueue) => {
  try {
    await channel.assertQueue(queueName, {
      durable: true,
    });
    channel.sendToQueue(queueName, Buffer.from(message));
  } catch (e) {
    console.log("rabbitMQ 메시지 전송 실패");
    insertLog(`${queueName}${sep}${message}`);
    //파일로 작성해야함.
    await connectRabbitMQ();
  }
};

 

자바에서 convertAndSend로 보내 준 메시지를 consume하는 함수

const consumeMessage = async () => {
  await channel.assertQueue(USER_NOTIFICATION_QUEUE, {
    durable: true,
  });

  channel.consume(
    USER_NOTIFICATION_QUEUE,
    (msg) => {
      if (!msg) {
        return logger.error(`Invalid incoming message`);
      }
      handleIncomingNotification(msg?.content?.toString());
      channel.ack(msg);
    },
    {
      noAck: false,
    }
  );
};

 

메시지에 따라 알림을 보내주는 함수

const handleIncomingNotification = (msg: string) => {
  try {
  
    let parsedMessage: DwQueueMessage = JSON.parse(msg);

    if (typeof parsedMessage === "string") {
      parsedMessage = JSON.parse(parsedMessage);
    }

    if (!parsedMessage.params) {
      logger.error("발송할 알림 정보가 없습니다.");
      return;
    }

    const noti: CustomNotiClass = parsedMessage.params;

    const { title = "알림", notiTo } = noti;
    const user = ...; // websocket 유저 정보
    user && user.send(msgType.langNotice, { title: title, body: { content: "notiContent" } });
  } catch (error) {
    logger.error(error);
    logger.error(`Error While Parsing and Sending the message`);
  }
};

 

 

https://hassanfouad.medium.com/using-rabbitmq-with-nodejs-and-typescript-8b33d56a62cc