BullMQ로 중복된 이벤트 제거하기

문제 상황

 

최근에 현업에서 업무를 진행하면서 특정 시간마다 DB에 부하가 심하게 발생하여 해당 클러스터를 사용하는 전체 서비스에 장애가 발생하고 있었습니다. 중복된 이벤트가 발생하여 수많은 콜백 로직이 동시다발적으로 수행되어 DB에 수많은 쿼리가 한 번에 요청되는 것이 문제였습니다. 과도한 Write 요청으로 인해 락을 획득하지 못하는 작업이 대기열에 쌓이고 Timout에 도달하여 오류가 발생했습니다. Unable to acquire ticket with mode '2' within a max lock request timeout of '5ms' milliseconds. 와 같은 에러 메시지에서도 볼 수 있듯이, 락 모드 '2'인 배타적 락을 획득하기 위한 티켓을 획득하지 못하여 문제가 발생하는 것입니다.

 

문제 원인

1️⃣ 레디스 Pub/Sub

이때, DB에 문제를 가하는 중복된 이벤트가 발생한 이유는 Redis Pub/Sub 구조가 모든 구독자에게 메시지를 푸시하기 때문입니다.

Messages sent by other clients to these channels will be pushed by Redis to all the subscribed clients. Subscribers receive the messages in the order that the messages are published.

https://redis.io/docs/latest/develop/interact/pubsub/

Redis Pub/Sub

 

broadcast 방식으로 구독자에게 메시지를 발행하므로 redis에서 발행한 이벤트를 모든 k8s 파드에서 전달받아 파드 개수만큼 메시지가 중복되고, 1번만 수행해야 할 후속 로직이 N배로 증가하여 부하가 심하게 발생합니다.

이뿐만 아니라, 각 메시지를 한 번의 쿼리로 수행하지 않고 각각 쿼리로 요청하기 때문에 그만큼 DB 통신 횟수가 증가하는 것도 문제 중 하나였습니다.

2️⃣ MongoDB 과부하

MongoDB는 동시성 제어를 위해 읽기/쓰기 티켓 시스템을 제공합니다.

 

기본적으로 MongoDB는 128개의 쓰기 티켓을 제공하며, 쓰기 작업 하나 당 1개의 티켓을 점유하고, 티켓이 모두 소진되면 후속 작업은 대기열에 들어갑니다. 대기열에 들어간 시간이 maxLockRequestTime을 초과하면 오류가 발생합니다. 너무 과한 쓰기 작업을 요청하면 티켓이 부족하여 실패할 가능성이 있습니다.

 

커넥션 개수가 과하게 증가하면 DB 리소스에 부하가 발생합니다.

 

커넥션은 각각 스레드를 점유하여 컨텍스트 스위칭 비용이 증가하고 자연스레 CPU에 부담을 주게 됩니다. CPU에 부담을 주므로 대기열에 쌓이는 작업을 빠르게 해소하지 못해 메모리에도 부담을 줍니다. 또한, 커넥션 자체가 많아지므로 병목이 발생합니다.

 

정리

결론적으로 단기간에 과도한 요청이 몰려 DB에 부하를 주었고, 트랜잭션이 락을 위한 티켓을 획득하지 못하여 오류가 발생하는 것으로 판단하였습니다.

 

그래서 이를 해결하고자 파드별로 수신한 메시지들의 중복을 없애고, 메시지 수신 후 후속 로직 작업을 Batch로 묶어서 쿼리하는 것을 중점적으로 개발하였습니다.

 

문제 해결

1️⃣ 중복된 이벤트 제거

1. BullMQ에 job을 추가하며 중복된 job을 제거

BullMQ는 JobId를 명시하여 큐에 메시지를 적재할 수 있습니다. 이때, job id는 메시지 별로 유니크하기 때문에 동일한 job id를 가진 메시지가 들어온다면 중복된 메시지로 판단하여 작업을 수행하지 않습니다.

 

예를 들어 아래 빨간 박스처럼 여러 개의 메시지를 적재할 때, 두 파드에서 같은 키의 이벤트를 발행하였지만 내부적으로 duplicated 처리되어 작업을 수행할 수 있게 됩니다.

 

BullMQ는 job을 처리할 때, added -> waiting -> active 순으로 작업을 완료하기 때문에 active 상태가 되기 전에 동일한 jobId의 요청이 발생한다면, duplicated 처리되어 작업을 더 이상 수행하지 않게 됩니다.

 

자세한 코드는 다음과 같습니다.

@Injectable()
export class EventQueueService implements OnModuleInit {
	private readonly logger = new Logger(EventQueueService.name);

	constructor(
		@Inject(REDIS_READ_CLIENT) private redisClient: Redis, // 레디스 클라이언트를 주입받습니다.
		@InjectQueue('event-queue') private readonly eventQueue: Queue,
	) {}
	
	async onModuleInit() {
		this.redisClient.subscribe("__keyevent@0__:expired", (err, count) => {
			if (err) {
				this.logger.error("구독 실패:", err);
			} else {
				this.logger.log(`구독 성공! 현재 구독된 채널 수: ${count}`);
			}
		});

		this.redisClient.on("message", (channel, message) => {
			await this.addToQueue(message);
		});

		this.eventQueue.trimEvents(1000); // stream 자료형에 쌓이는 bullmq 이벤트 개수를 1000개까지 보존
	}
	
	private async addToQueue(key: string) {
		await this.eventQueue.add(
			'event',
			{ key },
			{
				jobId: key, // 키를 jobId로 사용하여 중복 방지
				attempts: 3, // 실패 시 3번까지 재시도
				backoff: {
					type: 'exponential',
					delay: 1000, // 초기 지연 시간 1초
				},
				removeOnComplete: true, // 성공적으로 완료된 작업 제거
				removeOnFail: false, // 실패한 작업은 유지하여 조사 가능
			}
		);
	}
}

 

먼저 레디스를 구독하여 큐에 메시지를 적재할 객체를 선언합니다. 레디스 채널을 구독하여 메시지 이벤트가 발생하면 큐에 작업을 적재합니다. 이 과정에서 jobId가 중복으로 사용된 경우 중복 요청은 에러 없이 단순히 무시됩니다.

 

큐에 쌓은 작업을 이후 처리하게 됩니다.

@Processor("event-queue")
export class EventProcessor extends WorkerHost {
	private readonly logger = new Logger(EventProcessor.name);

	async process(job: Job<any, any, string>): Promise<any> {
		switch (job.name) {
			case "event": // job 이름
				try {
					await this.processEvent(job.data.key);
				} catch (error) {
					this.logger.error(`Error processing key ${job.data.key}:`, error);
					throw error; // 재시도를 위해 에러를 던지기
				}
				return;
		}
	}

	/** 후속 작업 수행 */
	private async processEvent(key: string) { /* 수행할 작업 */ }
}

 

@Processor 데코레이터를 붙인 후, 특정 큐 이름을 넣으면 이 객체는 큐에 쌓인 작업을 처리하는 객체가 됩니다. WorkerHost 추상 클래스를 상속하여 process 메서드를 구현하여 원하는 작업 이름에 따라 후처리를 진행하면 됩니다.

2️⃣ 여러 개의 쿼리를 하나의 쿼리로 묶어서 요청

해당 해결방안은 쿼리를 묶어 Bulk 요청을 수행하도록 하는 것이고 구현 방법이 다양하므로, 별도로 구현한 내용을 서술하지 않겠습니다.

 

결과

• 처리한 이벤트 개수
  • 1분 40초간 총 14000건의 이벤트를 처리하였습니다.
  • → 기존 방식은 72개의 파드에 레디스가 만료된 메시지 이벤트를 모두 브로드캐스팅하므로 11300 * 72 = 813,600 건의 DB Write 요청이 발생했을 것으로 추정됩니다(동시성 이슈로 2700 건의 중복 이벤트가 적재되긴 했습니다 -> 14000 - 11300 = 2700).
  • 즉, DB 쿼리 수를 약 98.6%로 줄일 수 있었고, 자정에 발생하던 DB 부하 문제도 해결할 수 있었습니다.
• DB 메트릭
  • CPU 사용률, 쿼리 수행 소요 시간, 커넥션 개수 측면에서 진폭이 줄어들었습니다.
  • 개선 이전

• 개선 이후

한계

BullMQ는 레디스를 기반으로 하는 메시지 큐입니다. 즉, 레디스에 부하가 발생할 수 있는 것입니다. DB의 부하는 줄일 수 있었으나, 레디스의 부하는 줄일 수 없으므로 어디에든 큰 부하가 발생한다는 것이 한계점이었습니다.

 

하지만, 당장 급선무였던 DB 부하를 개선할 수 있었으며, 레디스의 성능에도 무리가 없는 작업이었으므로 향후 레디스에 문제가 발생할 때까지 유지해도 괜찮다는 판단을 내렸습니다.

 

 

앞으로 중복 작업을 제거하거나, 카프카보다 간단하게 메시지 큐를 사용해야 한다면 BullMQ를 자주 사용하게 될 것 같습니다.