메세지와 이벤트를 통한 확장성과 내결함성.
이 글은 Asynchronous Microservices with RabbitMQ and Node.js를 번역한 글입니다.
해당 글은 2019년 2월 19일에 작성되었습니다.
번역 글은 여기에서도 확인할 수 있습니다.
웹의 거인 (Netflix, Amazon…)은 10 년 이상 마이크로 서비스를 사용해 왔습니다. 그러나 최근에 이러한 기술은 소규모 소프트웨어 프로젝트에서 더 쉽게 접근 할 수 있게 되었습니다. 이제 신생 기업은 이를 활용하여 강력하고 확장 가능한 클라우드 소프트웨어를 구축할 수 있습니다.
이 문서에서는 RabbitMQ 및 Node.js를 사용하여 마이크로 서비스를 시작하는 간단한 방법을 살펴봅니다.
저는 Martin Fowler의 간단한 정의를 좋아합니다.:
A particular way of designing software applications as suites of independently deployable services.
(소프트웨어 응용 프로그램을 독립적으로 배포 할 수있는 서비스 모음으로 설계하는 특별한 방법입니다.)– MARTIN FOWLER
전통적인 Monolith 앱은 API 요청, 인증, 여러 작업에 대한 백그라운드 데이터 처리 등을 처리합니다. 아이디어는 각각 더 작은 작업 집합을 처리하는 특수 마이크로 서비스로 분할하는 것입니다.
이제 모놀리스를 여러 서비스로 분할 했으므로 이들이 협업 할 수있는 방법을 설정해야합니다. 서비스 간 통신에 사용되는 가장 일반적인 2 가지 패턴은 다음과 같습니다.
Remote Procedures는 기본적으로 서비스 A에서 서비스 B로 직접 요청을 보내고 동기적으로 응답을 기다립니다. 이를 달성하기 위한 기술의 예로는 REST 및 gRPC가 있습니다.
Asynchronous Messaging에는 별도의 계층 또는 메시지 브로커를 사용하여 서비스 A에서 서비스 B로 메시지를 보냅니다. 서비스 A는 응답을 기다리지 않습니다. 서비스 B는 일반적으로 동일한 메시징 시스템을 통해 결과를 사용할 수 있을 때 (결과가 예상되는 경우) 결과를 보냅니다.
Don’t call us, we’ll call you
(우리한테 전화 하지마. 우리가 너에게 전화할게.)— ASYNCHRONOUS MESSAGING
이를 달성하기 위해 활용할 수있는 기술의 예로 RabbitMQ와 Apache Kafka가 있습니다.
마이크로 서비스의 간단한 사용 사례를 설명하기 위해 REST API를 통해 외부 요청을 수신하고 데이터에 대해 비용이 많이 드는 처리를 수행하고 결과를 저장하는 웹 애플리케이션을 구축 할 것입니다.
이 앱의 요구 사항은 다음과 같습니다.
이 문제를 해결하는 한 가지 방법은 이 프로세스의 모든 단계를 처리하는 단일 앱을 빌드하는 것입니다. 그러나 이 접근 방식의 몇 가지 문제는 다음과 같습니다.
우리는 앱을 WebService와 ProcessorService의 두 가지 서비스로 분할할 것입니다.
WebService와 ProcessorService가 통신하는 방법으로 REST API를 사용하려면 다음과 같은 잠재적인 문제를 처리해야합니다.
이러한 문제에 대한 가능한 해결책은 WebService에서 ProcessorService로 메시지를 전달하고 ProcessorService에서 WebService로 메시지를 전달할 때 중개자 역할을 하는 "메시지 브로커"를 포함하는 것입니다. 이를 구현하는 데 사용할 기술은 다음과 같습니다.
코드를 참조하고 따라 할 수 있도록 github 저장소를 설정했습니다.
여기에서 다운로드 할 수있는 최신 Node.js LTS 릴리스 (이 기사 작성 시점의 10.15.0)(역주 번역 당시 LTS는 14.16.0입니다.)*를 사용할 것입니다.
RabbitMQ를 시작하는 가장 쉬운 방법은 CloudAMQP를 사용하는 것입니다. 그들은 RabbitMQ를 서비스로 제공하므로 RabbitMQ를 구성하고 유지하는 대신 앱 구축에 집중할 수 있습니다. 이 방법으로 RabbitMQ를 설정하려면 몇 가지 간단한 단계 만 필요합니다.
무료 계정을 처음 등록하십시오. 그런 다음 프로젝트를 만듭니다.
프로젝트에 CloudAMQP 리소스를 추가합니다.
CloudAMQP 무료 플랜 선택 :
다음 화면에서“Show Credentials”를 클릭 한 다음“Download .env”를 클릭하십시오. 프로젝트 경로에서 파일 이름 ".env"로 파일을 저장합니다.
이제 RabbitMQ 인스턴스를 사용할 준비가되었습니다. 다음 Node.js 스크립트를 실행하여 큐를 설정할 수 있습니다.
require('dotenv').config()
const amqp = require('amqplib')
// RabbitMQ connection string
const messageQueueConnectionString = process.env.CLOUDAMQP_URL
async function setup() {
console.log('Setting up RabbitMQ Exchanges/Queues')
// connect to RabbitMQ Instance
let connection = await amqp.connect(messageQueueConnectionString)
// create a channel
let channel = await connection.createChannel()
// create exchange
await channel.assertExchange('processing', 'direct', { durable: true })
// create queues
await channel.assertQueue('processing.requests', { durable: true })
await channel.assertQueue('processing.results', { durable: true })
// bind queues
await channel.bindQueue('processing.requests', 'processing', 'request')
await channel.bindQueue('processing.results', 'processing', 'result')
console.log('Setup DONE')
process.exit()
}
setup()
위의 스크립트에서 수행한 작업은 다음과 같습니다.
여기서 RabbitMQ 개념에 대한 자세한 내용을 읽을 수 있지만 일반적인 아이디어는 WebService
가 processing.requests
대기열로 요청을 보내고 ProcessorService
는 거기에서 요청을 읽고 WebService
가 액세스 할 수있는 processing.results
에 결과를 게시한다는 것입니다.
WebService에 대한 코드는 여기에서 찾을 수 있습니다. Express.js를 사용하여 엔드 포인트 /api/v1/processData
에 대한 POST
요청을 처리하는 웹 API입니다. 각 요청에 requestId를 할당하고이를 RabbitMQ에 보내고 requestId를 응답으로 표시합니다. 서비스는 또한 ProcessorService의 결과를 수신하고 기록합니다.
const path = require('path')
require('dotenv').config({ path: path.resolve(process.cwd(), '../.env') })
const express = require('express')
const app = express()
const http = require('http')
const bodyParser = require('body-parser')
const amqp = require('amqplib')
// Middleware
app.use(bodyParser.json())
// simulate request ids
let lastRequestId = 1
// RabbitMQ connection string
const messageQueueConnectionString = process.env.CLOUDAMQP_URL
// handle the request
app.post('/api/v1/processData', async function(req, res) {
// save request id and increment
let requestId = lastRequestId
lastRequestId++
// connect to Rabbit MQ and create a channel
let connection = await amqp.connect(messageQueueConnectionString)
let channel = await connection.createConfirmChannel()
// publish the data to Rabbit MQ
let requestData = req.body.data
console.log('Published a request message, requestId:', requestId)
await publishToChannel(channel, {
routingKey: 'request',
exchangeName: 'processing',
data: { requestId, requestData },
})
// send the request id in the response
res.send({ requestId })
})
// utility function to publish messages to a channel
function publishToChannel(channel, { routingKey, exchangeName, data }) {
return new Promise((resolve, reject) => {
channel.publish(
exchangeName,
routingKey,
Buffer.from(JSON.stringify(data), 'utf-8'),
{ persistent: true },
function(err, ok) {
if (err) {
return reject(err)
}
resolve()
}
)
})
}
async function listenForResults() {
// connect to Rabbit MQ
let connection = await amqp.connect(messageQueueConnectionString)
// create a channel and prefetch 1 message at a time
let channel = await connection.createChannel()
await channel.prefetch(1)
// start consuming messages
await consume({ connection, channel })
}
// consume messages from RabbitMQ
function consume({ connection, channel, resultsChannel }) {
return new Promise((resolve, reject) => {
channel.consume('processing.results', async function(msg) {
// parse message
let msgBody = msg.content.toString()
let data = JSON.parse(msgBody)
let requestId = data.requestId
let processingResults = data.processingResults
console.log(
'Received a result message, requestId:',
requestId,
'processingResults:',
processingResults
)
// acknowledge message as received
await channel.ack(msg)
})
// handle connection closed
connection.on('close', err => {
return reject(err)
})
// handle errors
connection.on('error', err => {
return reject(err)
})
})
}
// Start the server
const PORT = 3000
server = http.createServer(app)
server.listen(PORT, 'localhost', function(err) {
if (err) {
console.error(err)
} else {
console.info('Listening on port %s.', PORT)
}
})
// listen for results on RabbitMQ
listenForResults()
ProcessorService의 코드는 여기에서 찾을 수 있습니다. RabbitMQ 채널 처리에서 요청 메시지를 수신하고 이전에 정의한 요청을 처리 한 후 결과를 채널 처리로 다시 보냅니다. 이 데모의 목적을 위해 5 초 동안 기다렸다가 입력 문자열 끝에 '-processed'를 연결하여 heavy-processing 부분을 시뮬레이션합니다.
const path = require('path')
require('dotenv').config({ path: path.resolve(process.cwd(), '../.env') })
const amqp = require('amqplib')
// RabbitMQ connection string
const messageQueueConnectionString = process.env.CLOUDAMQP_URL
async function listenForMessages() {
// connect to Rabbit MQ
let connection = await amqp.connect(messageQueueConnectionString)
// create a channel and prefetch 1 message at a time
let channel = await connection.createChannel()
await channel.prefetch(1)
// create a second channel to send back the results
let resultsChannel = await connection.createConfirmChannel()
// start consuming messages
await consume({ connection, channel, resultsChannel })
}
// utility function to publish messages to a channel
function publishToChannel(channel, { routingKey, exchangeName, data }) {
return new Promise((resolve, reject) => {
channel.publish(
exchangeName,
routingKey,
Buffer.from(JSON.stringify(data), 'utf-8'),
{ persistent: true },
function(err, ok) {
if (err) {
return reject(err)
}
resolve()
}
)
})
}
// consume messages from RabbitMQ
function consume({ connection, channel, resultsChannel }) {
return new Promise((resolve, reject) => {
channel.consume('processing.requests', async function(msg) {
// parse message
let msgBody = msg.content.toString()
let data = JSON.parse(msgBody)
let requestId = data.requestId
let requestData = data.requestData
console.log('Received a request message, requestId:', requestId)
// process data
let processingResults = await processMessage(requestData)
// publish results to channel
await publishToChannel(resultsChannel, {
exchangeName: 'processing',
routingKey: 'result',
data: { requestId, processingResults },
})
console.log('Published results for requestId:', requestId)
// acknowledge message as processed successfully
await channel.ack(msg)
})
// handle connection closed
connection.on('close', err => {
return reject(err)
})
// handle errors
connection.on('error', err => {
return reject(err)
})
})
}
// simulate data processing that takes 5 seconds
function processMessage(requestData) {
return new Promise((resolve, reject) => {
setTimeout(() => {
resolve(requestData + '-processed')
}, 5000)
})
}
listenForMessages()
WebService만 실행하고 터미널에서 몇 가지 요청을 보내는 것으로 시작합니다.
$ *cd web-service
$ node web-service.js
*Listening on port 3000.
*// From a different terminal
*$ *curl --header "Content-Type: application/json" \
--request POST \
--data '{"data":"my-data"}' \
[http://localhost:3000/api/v1/processData](http://localhost:3000/api/v1/processData)*
{"requestId":1}
$ *curl --header "Content-Type: application/json" \
--request POST \
--data '{"data":"more-data"}' \
[http://localhost:3000/api/v1/processData](http://localhost:3000/api/v1/processData)*
{"requestId":2}
각 인스턴스는 메시지를 처리했습니다. RabbitMQ 대시 보드로 돌아가서 결과가 대기열에 있음을 확인합니다.
마지막으로 웹 서비스를 다시 시작합니다.
$ cd web-service
$ node web-service.js
Received a result message, requestId: 1 processingResults: my-data-processed
Received a result message, requestId: 2processingResults: more-data-processed
대시 보드를 다시 확인하면 모든 메시지가 처리되었으므로 두 대기열이 모두 비어 있어야합니다.
분리 된 아키텍처 덕분에 더 많은 WebService 또는 ProcessorService 인스턴스를 독립적으로 추가하여이 앱을 확장 할 수 있습니다. 실패한 메시지는 RabbitMQ에 자동으로 다시 추가되어 앱이 좀 더 강력 해집니다.
이 글에서는 CloudAMQP를 통해 Node.js + RabbitMQ를 살펴 보았지만 RabbitMQ를 다음과 같은 다른 솔루션으로 대체 할 수 있습니다.
마지막으로 솔루션을 프로덕션에 배포하는 데 몇 가지 추가 단계가 필요합니다.