Redis - bull message queue

이번에는 express.js에서 bull 라이브러리를 사용한 메세지 큐에 대해 포스팅을 하겠습니다.

먼저, 기존에는 클라이언트쪽에서 for문을 돌면서 여러개의 엑셀파일을 직접 만들고 있었는데, 이를 api화 하면서 엑셀 저장을 해야되는 상황이 있었습니다.

그런데 보통 api는 30초 후에는 timeout이 걸리기 때문에 엑셀 1개당 저장 속도는 약 1분에서 1분30초이기 때문에 이를 메세지 큐를 활용하여 비동기로 통신하며 엑셀을 저장하면 좀 더 효율적이라고 판단하여 redis의 bull 라이브러리를 이용하여 메세지 큐를 구현하기로 하였습니다.

그러면 먼저 bull 에 대해서 살펴보겠습니다.

1. Bull 라이브러리란?

Bull은 Node.js용 빠르고 신뢰할 수 있는 작업 및 메시지 큐 라이브러리로, Redis를 백엔드로 사용합니다. 주로 다음과 같은 이유로 사용됩니다:

• 대량의 작업 처리
• 작업 재시도 및 지연 처리 지원
• 작업 우선순위 지정
• 작업 이벤트 처리

2. 프로젝트 설정

먼저 Express.js 프로젝트를 초기화하고 필요한 패키지를 설치합니다.

2.1. 프로젝트 초기화

mkdir express-bull-queue
cd express-bull-queue
npm init -y

2.2. 필요한 패키지 설치

npm install express bull ioredis
npm install dotenv --save

• express: Node.js 웹 프레임워크
• bull: 메시지 큐 라이브러리
• ioredis: Redis 클라이언트
• dotenv: 환경 변수 관리

 

3. Redis 설정

Bull은 Redis를 백엔드로 사용하므로, Redis가 필요합니다. 로컬에 Redis를 설치하거나 Docker를 사용하여 Redis를 실행할 수 있습니다.

3.1. Docker로 Redis 실행

docker run -d --name redis-server -p 6379:6379 redis

Redis가 제대로 실행 중인지 확인하려면 다음 명령어를 사용할 수 있습니다:

docker logs redis-server

 

4. Express.js와 Bull 연동

4.1. 디렉토리 구조

express-bull-queue
├── .env
├── package.json
├── server.js
└── queue
    ├── emailQueue.js
    └── emailProcessor.js

4.2. .env 파일 설정

.env 파일에 Redis 연결 정보를 추가합니다:

REDIS_HOST=127.0.0.1
REDIS_PORT=6379

4.3. 큐 설정 (queue/emailQueue.js)

const Queue = require('bull');
require('dotenv').config();

// Redis 연결 설정
const emailQueue = new Queue('email', {
  redis: {
    host: process.env.REDIS_HOST,
    port: process.env.REDIS_PORT,
  },
});

module.exports = emailQueue;

4.4. 작업 처리 로직 (queue/emailProcessor.js)

module.exports = async (job) => {
  console.log(`Processing job with id: ${job.id}`);
  console.log(`Sending email to: ${job.data.email}`);

  // 이메일 처리 로직 (여기서는 모의 작업)
  await new Promise((resolve) => setTimeout(resolve, 2000));

  console.log(`Email sent to: ${job.data.email}`);
};

4.5. Express.js 서버 설정 (server.js)

const express = require('express');
const emailQueue = require('./queue/emailQueue');
const emailProcessor = require('./queue/emailProcessor');

const app = express();
app.use(express.json());

// 작업 프로세서 등록
emailQueue.process(emailProcessor);

// 작업 추가 엔드포인트
app.post('/send-email', async (req, res) => {
  const { email } = req.body;

  if (!email) {
    return res.status(400).json({ error: 'Email is required' });
  }

  // 작업 큐에 추가
  const job = await emailQueue.add({ email });
  console.log(`Job added with id: ${job.id}`);

  res.status(200).json({ message: 'Email job added', jobId: job.id });
});

const PORT = 3000;
app.listen(PORT, () => {
  console.log(`Server running on http://localhost:${PORT}`);
});

---

5. 실행 및 테스트

5.1. 서버 실행

node server.js

5.2. 테스트

Postman 또는 cURL을 사용하여 /send-email 엔드포인트를 호출합니다.

cURL 예시

curl -X POST http://localhost:3000/send-email \
-H "Content-Type: application/json" \
-d '{"email": "test@example.com"}'

결과 로그

Job added with id: 1
Processing job with id: 1
Sending email to: test@example.com
Email sent to: test@example.com

---

6. 추가 기능

6.1. 작업 지연

const job = await emailQueue.add({ email }, { delay: 5000 });

위 코드는 작업을 5초 후에 실행하도록 설정합니다.

6.2. 작업 우선순위

const job = await emailQueue.add({ email }, { priority: 1 });

작업 우선순위를 설정하여 더 중요한 작업이 먼저 처리되도록 할 수 있습니다.

6.3. 작업 재시도

Bull은 작업 실패 시 재시도를 지원합니다:

emailQueue.process(async (job) => {
  try {
    // 작업 로직
  } catch (error) {
    throw new Error('Retry this job');
  }
});

 

Bull 에 대한 기본 개념 및 예시 문구를 소개했고, 다음에는 어떻게 도입했는지에 대해 포스팅 하도록 하겠습니다.