O que é um serviço de mensageria
O serviço de mensageria é uma forma de trocar informações entre diferentes partes de uma aplicação ou sistema de forma organizada e eficiente. Ele ajuda a separar as partes da aplicação e facilita a sua manutenção e desenvolvimento.
Imaginando uma aplicação de e-commerce em que o serviço de compra de produtos fica separado do serviço que processa o pagamento e efetivamente realiza a transação entre as contas bancárias, a forma mais segura e efetiva de realizar a comunicação entre eles é via mensageria.
Com mensageria, o serviço de compra de produtos vai enviar uma mensagem para uma fila com as informações de transação, enquanto o serviço de pagamento vai buscar as mensagens nesta fila e realizar as transações. Dessa forma é possível garantir que, mesmo que o serviço de compra de produtos caia, ou esteja passando por uma demanda gigantesca que possa comprometer o processamento de novas requisições, os pagamentos ainda assim serão processados de forma correta. Isso também garante que, mesmo havendo milhares de pagamentos na fila para serem processados, o usuário não precisará aguardar a transação ser realizada, já que uma vez enviado o pagamento para fila, é garantido que ele será consumido pelo serviço de processamento do pagamento, independemente do tempo que demore.
Objetivo neste tutorial
Neste tutorial iremos criar um exemplo básico de um serviço de e-commerce (producer) que se comunica com um serviço de pagamento (consumer), e toda a comunicação será feita via mensageria.
Setup
Como o objetivo é entender o funcionamento de um serviço de mensageria utilizando Amazon SQS, não iremos nos aprofundar tanto na criação dos serviços de e-commerce e pagamento. Em vez disso, vamos nos concentrar na comunicação entre os dois usando como base um projeto hospedado no repositório: https://github.com/anti-duhring/amazon-sqs-tutorial
Abra o terminal e digite:
git clone https://github.com/anti-duhring/amazon-sqs-tutorial.git
Após isso, instale as dependências da aplicação com o comando:
npm install
Executando a aplicação
Iremos executar ambos os serviços: producer e consumer.
Vá até o root da pasta local do repositório que você acabou de clonar e abra dois terminais.
Em um deles, digite o comando:
npm run start:producer
No outro, digite o comando:
npm run start:consumer
Entendendo o funcionamento
A aplicação consiste em dois serviços: producer e consumer.
-
Producer: Uma simulação de um e-commerce, que possui uma rota/request-transfer, cujo objetivo é enviar ao serviço de pagamento (consumer) uma requisição com o número da conta que fará o pagamento, o número da conta que receberá o pagamento e a quantia de dinheiro a ser transferido. -
Consumer: Um exemplo de um serviço de pagamento que recebe as informações por meio da rota/process-transfere realiza a transação.
Por ora a comunicação entre os dois serviços está sendo feita de forma direta, ou seja, dentro da própria rota /request-transfer o producer recebe as informações do pagamento e já dispara uma requisição para a rota /process-transfer do consumer, que realiza a transação.
Agora, vamos adaptar o projeto para que o producer não precise mais se comunicar diretamente com o consumer. Em vez disso, o producer será responsável apenas por enviar os pedidos de transação para uma fila, e o consumer será responsável apenas por buscar esses pedidos constantemente e processá-los.
Criando nossa fila principal
Vá até o serviço Amazon SQS no seu console da AWS.
Após acessar a página do Amazon SQS, clique no botão "Create queue" (criar fila).
Dentro da página de criação da fila você verá uma série de opções para configuração. Entre elas, destaco aqui o type (tipo) da fila:
O tipo da fila, pode ser standard(comum) ou FIFO (First In First Out). Filas comuns não possuem ordem nas mensagens, enquanto as FIFO ordena a fila em forma de pilha, onde a primeira mensagem que foi criada será a primeira a ser encontrada ao buscar mensagens dessa fila.
Para os propósitos deste tutorial, vamos usar as seguintes configurações que já são atribuídas por padrão pelo próprio SQS:
Clique no link Info ao lado da configuração para entender melhor o que cada propriedade representa.
Após ter criado a fila copie o URL gerado no final.
Criando o service de comunicação com o Amazon SQS
Para facilitar a comunicação da nossa aplicação com o Amazon SQS, vamos criar um service, dentro da pasta src/service, que terá três funções principais: enviar uma mensagem para a fila, recuperar mensagens da fila e deletar mensagens da fila.
Esse service será útil para desacoplar o SQS da nossa aplicação, permitindo que ele seja facilmente adaptado para funcionar com outros serviços de fila também.
Criando a classe SqsService
Vamos começar com uma classe com apenas dois atributos, ambos privados, client e queueUrl.
export class SqsService {
#client
#queueUrl
constructor() {
}
}
A ideia é que client seja uma instância do SQSClient do pacote @aws-sdk/client-sqs, que serve pra estabelecer uma conexão com a nossa fila. Enquanto que queueUrl será o URL da fila copiado anteriormente, que deverá ser passado via constructor, assim vamos poder usar o service para outras filas, se for preciso.
Adicionando o client do SQS no constructor
Agora precisamos adicionar o SQSClient no constructor, e atribuí-lo como valor da propriedade client.
import { SQSClient } from "@aws-sdk/client-sqs";
import * as dotenv from 'dotenv';
dotenv.config()
export class SqsService {
#client
#queueUrl
constructor(queue) {
this.#client = new SQSClient({
region: process.env.AWS_REGION,
credentials: {
accessKeyId: process.env.AWS_ACCESS_KEY_ID,
secretAccessKey: process.env.AWS_ACESS_KEY
}
})
this.#queueUrl = queue
}
}
Perceba que, para estabelecer uma conexão, o client requere algumas informações, que são parte da interface SQSClientConfig: region, accessKeyId e secretAccessKey. Essas informações são referentes a sua conta da AWS e não devem ser compartilhadas, por isso iremos usar um arquivo .env pra criar variáveis de ambiente.
Basta criar um arquivo .env no root do projeto com as seguintes variáveis:
AWS_REGION=REGIÃO (ex: us-east-1)
AWS_ACCESS_KEY_ID=ID DO USUÁRIO
AWS_ACESS_KEY=CHAVE DE ACESSO
AWS_MAIN_QUEUE_URL=URL DA SUA FILA
Adicionando método para enviar mensagens para fila
Agora precisamos criar um método que irá enviar uma mensagem para a nossa fila usando o client.
O método terá como parâmetro um objeto com duas propriedades: body e attributes. O body será o texto que ficará no corpo da mensagem, enquanto attributes é um valor opcional que fornece mais informações sobre a mensagem.
Aqui está um exemplo do objeto, mas você pode entender melhor suas propriedades na documentação oficial da AWS:
{
body: 'MENSAGEM QUE IRÁ NO CORPO',
attributes: {
'VALOR_OPCIONAL1': {
DataType: 'String',
StringValue: 'VALOR DO ATRIBUTO'
}
}
}
Nosso método de enviar mensagem será assim:
import { SQSClient } from "@aws-sdk/client-sqs";
import * as dotenv from 'dotenv';
dotenv.config()
export class SqsService {
#client
#queueUrl
constructor(queue) {
this.#client = new SQSClient({
region: process.env.AWS_REGION,
credentials: {
accessKeyId: process.env.AWS_ACCESS_KEY_ID,
secretAccessKey: process.env.AWS_ACESS_KEY
}
})
this.#queueUrl = queue
}
async sendMessage({ body, attributes }) {
try {
const sendMessageCommand = new SendMessageCommand({
QueueUrl: this.#queueUrl,
MessageBody: body,
MessageAttributes: attributes,
DelaySeconds: 0
})
const response = await this.#client.send(sendMessageCommand)
return response
} catch(err) {
throw err
}
}
}
Agora podemos enviar mensagens para a fila com o nosso service!
Adicionando método para buscar mensagens na fila
Além de enviar mensagens, precisamos também de um método para buscar mensagens na fila e retornar uma array delas:
async receiveMessages() {
try {
const receiveMessageCommand = new ReceiveMessageCommand({
QueueUrl: this.#queueUrl,
MaxNumberOfMessages: 10,
WaitTimeSeconds: 20,
MessageAttributeNames: ['All']
})
const response = await this.#client.send(receiveMessageCommand)
const messages = response?.Messages?.length? response.Messages : []
return messages
} catch(err) {
throw err
}
}
Entendendo as propriedades:
-
QueueUrl: A URL da nossa fila. -
MaxNumberOfMessages: O número máximo de mensagens a serem retornadas. O Amazon SQS nunca retorna mais mensagens do que esse valor (no entanto, menos mensagens podem ser retornadas). -
WaitTimeSeconds: A duração (em segundos) pela qual a chamada aguarda a chegada de uma mensagem na fila antes de retornar. -
MessageAttributeNames: Atributos que serão recebidos.
Para saber mais sobre as propriedades consulte a documentação oficial sobre a interface ReceiveMessageCommandInput da AWS.
Adicionando método de deletar mensagem da fila
Após a mensagem ser recebida e a transação ser concluída pelo consumer, é importante excluir a mensagem para evitar processamento duplicado.
Para isso, vamos criar um método em nosso service que receberá um parâmetro chamado ReceiptHandle, que é um identificador associado ao recebimento de uma mensagem. Um novo identificador de recebimento é gerado cada vez que uma mensagem é recebida. Ao excluir uma mensagem, é necessário fornecer o ReceiptHandle mais recente associado a ela para garantir que a mensagem correta seja excluída.
async deleteMessage(receiptHandle) {
try {
const deleteMessageCommand = new DeleteMessageCommand({
QueueUrl: this.#queueUrl,
ReceiptHandle: receiptHandle
})
await this.#client.send(deleteMessageCommand)
} catch(err) {
throw err
}
}
Entendendo as propriedades:
-
QueueUrl: link da nossa fila. -
ReceiptHandle: Identificador da nossa mensagem.
Para saber mais sobre as propriedades consulte a documentação oficial sobre a interface DeleteMessageCommandInput da AWS.
Usando o service de comunicação com o Amazon SQS no producer
Agora que criamos o service de comunicação com o Amazon SQS, podemos finalmente modificar a nossa rota /request-transfer do producer. Em vez de disparar uma requisição diretamente para a rota /process-transfer do consumer, o producer enviará uma mensagem com as informações do pagamento para a nossa fila, para que o consumer possa recebê-la e consumi-la no momento adequado.
Nossa rota /request-transfer terá, agora, o seguinte conteúdo:
Primeiro criamos uma nova instância do SqsService e, após isso, chamamos o método sendMessage(), passando como body as informações da nossa transação. Como o body precisa ser uma string, o método JSON.stringify() é chamado para converter nosso objeto com as informações da transação em texto.
app.post('/transfer', async(req, res) => {
const {
fromAccount,
toAccount,
amount
} = req.body;
try {
const sqsService = new SqsService(process.env.AWS_MAIN_QUEUE_URL)
await sqsService.sendMessage({
body: JSON.stringify({
fromAccount,
toAccount,
amount
})
})
res.status(200).json({ message: 'Sua transação será processada' })
} catch(err) {
console.log(err)
res.status(500).json({ message: 'Erro interno' })
}
})
Consumindo as mensagens da fila no consumer
Agora que o producer envia as mensagens para a fila e não se conecta mais diretamente ao consumer, precisamos criar um worker que busque constantemente por transações em nossa fila e as processe. Para isso, usaremos uma biblioteca bem conhecida do Node.js para agendamento de tarefas, chamada node-cron. Com ela, poderemos atribuir uma tarefa a ser executada repetidamente pelo consumer. Essa tarefa será buscar novos pedidos de transações na fila e, no nosso caso, para facilitar a visualização do que está acontecendo, vamos definir um intervalo de 1 minuto. Ou seja, o consumer buscará novas transações na fila e as processará a cada 1 minuto.
Criando a task agendada com node-cron
Para criar uma task com node-cron que seja executada a cada 1min, basta substituirmos nossa rota /process-transfer por essa função:
import cron from 'node-cron'
cron.schedule('* * * * *', () => {
console.log('Running every minute')
})
Agora sempre que o nosso serviço consumer estiver sendo executado, uma task será realizada automaticamente a cada 1min.
Consumindo mensagens do Amazon SQS dentro da task
Agora que nossa task está sendo executada a cada 1min, precisamos fazer com que ela busque mensagens que estão na nossa fila, para isso, iremos criar uma nova instância do SqsService e chamar o método receiveMessages() que criamos anteriormente.
cron.schedule('* * * * *', async() => {
const sqsService = new SqsService(process.env.AWS_MAIN_QUEUE_URL)
const messages = await sqsService.receiveMessages()
console.log(messages)
})
O output será uma array de objetos Message, que são as nossas mensagens:
Removendo mensagens da fila
Agora que o nosso consumer já recebe as mensagens da fila de forma autônoma, isto é, sem precisar ser "chamado" pelo producer, vamos implementar um protótipo que simula o processamento da transação de cada mensagem. Ao final do processamento, a mensagem será removida da fila.
É importante destacar que toda mensagem precisa ser removida da fila após ser consumida, para evitar que ela seja processada duas vezes.
A função que iremos adicionar faz um loop por cada mensagem recebida e simula seu processamento com um setTimeout dentro de uma Promise, demorando 5 segundos pra cada mensagem:
cron.schedule('* * * * *', async() => {
const sqsService = new SqsService(process.env.AWS_MAIN_QUEUE_URL)
const messages = await sqsService.receiveMessages()
for(let i = 0; i < messages.length; i++) {
await new Promise(resolve =>
setTimeout(() => {
console.log(`Simulando processamento da mensagem ${messages[i].MessageId}...`)
resolve()
}, 5000)
)
}
})
Após o processamento precisamos remover a mensagem da fila. Pra isso iremos chamar o método deleteMessage() do nosso SqsService. Ele recebe como parâmetro o ReceiptHandle da mensagem.
cron.schedule('* * * * *', async() => {
const sqsService = new SqsService(process.env.AWS_MAIN_QUEUE_URL)
const messages = await sqsService.receiveMessages()
for(let i = 0; i < messages.length; i++) {
await new Promise(resolve =>
setTimeout(() => {
console.log(`Simulando processamento da mensagem ${messages[i].MessageId}...`)
resolve()
}, 5000)
)
await sqsService.deleteMessage(messages[i].ReceiptHandle)
console.log(`Mensagem ${messages[i].MessageId} processada e removida da fila...`)
}
})
Pronto! Criamos um serviço de pagamentos que se comunica via mensageria com o Amazon SQS. Perceba que, agora, o serviço que pede que pagamentos sejam feitos e o serviço que de fato processa o pagamento são totalmente independentes um do outro, o que facilita a escalabilidade e manutenção.
Top comments (0)