请求/回复模式

import {nanoid} from 'nanoid'

export function createRequestChannel(channel) { // (1)
    const correlationMap = new Map()

    function sendRequest(data) { // (2)
        console.log('Sending request', data)
        return new Promise((resolve, reject) => {
            const correlationId = nanoid()
            const replyTimeout = setTimeout(() => {
                correlationMap.delete(correlationId)
                reject(new Error('Request timeout'))
            }, 10000)
            correlationMap.set(correlationId, (replyData) => {
                correlationMap.delete(correlationId)
                clearTimeout(replyTimeout)
                resolve(replyData)
            })
            channel.send({
                type: 'request',
                data,
                id: correlationId
            })
        })
    }

    channel.on('message', message => { // (3)
        const callback = correlationMap.get(message.inReplyTo)
        if (callback) {
            callback(message.data)
        }
    })
    return sendRequest
}

export function createReplyChannel(channel) {
    return function registerHandler(handler) {
        channel.on('message', async message => {
            if (message.type !== 'request') {
                return
            }
            const replyData = await handler(message.data) // (1)
            channel.send({ // (2)
                type: 'response',
                data: replyData,
                inReplyTo: message.id
            })
        })
    }
}

import {createReplyChannel} from './createReplyChannel.js'

const registerReplyHandler = createReplyChannel(process)
registerReplyHandler(req => {
    return new Promise(resolve => {
        setTimeout(() => {
            resolve({sum: req.a + req.b})
        }, req.delay)
    })
})
process.send('ready')

import {fork} from 'child_process'
import {dirname, join} from 'path'
import {fileURLToPath} from 'url'
import {once} from 'events'
import {createRequestChannel} from './createRequestChannel.js'

const __dirname = dirname(fileURLToPath(import.meta.url))

async function main() {
    const channel = fork(join(__dirname, 'replier.js')) // (1)
    const request = createRequestChannel(channel)
    try {
        const [message] = await once(channel, 'message') // (2)
        console.log(`Child process initialized: ${message}`)
        const p1 = request({a: 1, b: 2, delay: 500}) // (3)
            .then(res => {
                console.log(`Reply: 1 + 2 = ${res.sum}`)
            })
        const p2 = request({a: 6, b: 1, delay: 100}) // (4)
            .then(res => {
                console.log(`Reply: 6 + 1 = ${res.sum}`)
            })
        await Promise.all([p1, p2]) // (5)
    } finally {
        channel.disconnect() // (6)
    }
}

main().catch(err => console.error(err))

Child process initialized: ready
Sending request { a: 1, b: 2, delay: 500 }
Sending request { a: 6, b: 1, delay: 100 }
Reply: 6 + 1 = 7
Reply: 1 + 2 = 3

export class AMQPRequest {
    constructor () {
        this.correlationMap = new Map()
    }
//...

async initialize () {
    this.connection = await amqp.connect('amqp://localhost')
    this.channel = await this.connection.createChannel()
    const { queue } = await this.channel.assertQueue('', // (1)
        { exclusive: true })
    this.replyQueue = queue
    this.channel.consume(this.replyQueue, msg => { // (2)
        const correlationId = msg.properties.correlationId
        const handler = this.correlationMap.get(correlationId)
        if (handler) {
            handler(JSON.parse(msg.content.toString()))
        }
    }, { noAck: true })
}

send (queue, message) {
    return new Promise((resolve, reject) => {
        const id = nanoid() // (1)
        const replyTimeout = setTimeout(() => {
            this.correlationMap.delete(id)
            reject(new Error('Request timeout'))
        }, 10000)
        this.correlationMap.set(id, (replyData) => { // (2)
            this.correlationMap.delete(id)
            clearTimeout(replyTimeout)
            resolve(replyData)
        })
        this.channel.sendToQueue(queue, // (3)
            Buffer.from(JSON.stringify(message)),
            { correlationId: id, replyTo: this.replyQueue }
        )
    })
}

    destroy () {
        this.channel.close()
        this.connection.close()
    }
}

import amqp from 'amqplib'

export class AMQPReply {
    constructor(requestsQueueName) {
        this.requestsQueueName = requestsQueueName
    }

    async initialize() {
        const connection = await amqp.connect('amqp://localhost')
        this.channel = await connection.createChannel()
        const {queue} = await this.channel.assertQueue( // (1)
            this.requestsQueueName)
        this.queue = queue
    }

    handleRequests(handler) { // (2)
        this.channel.consume(this.queue, async msg => {
            const content = JSON.parse(msg.content.toString())
            const replyData = await handler(content)
            this.channel.sendToQueue( // (3)
                msg.properties.replyTo,
                Buffer.from(JSON.stringify(replyData)),
                {correlationId: msg.properties.correlationId}
            )
            this.channel.ack(msg)
        })
    }
}

import {AMQPReply} from './amqpReply.js'

async function main() {
    const reply = new AMQPReply('requests_queue')
    await reply.initialize()
    reply.handleRequests(req => {
        console.log('Request received', req)
        return {sum: req.a + req.b}
    })
}

main().catch(err => console.error(err))

import {AMQPRequest} from './amqpRequest.js'
import delay from 'delay'

async function main() {
    const request = new AMQPRequest()
    await request.initialize()

    async function sendRandomRequest() {
        const a = Math.round(Math.random() * 100)
        const b = Math.round(Math.random() * 100)
        const reply = await request.send('requests_queue', {a, b})
        console.log(`${a} + ${b} = ${reply.sum}`)
    }

    for (let i = 0; i < 20; i++) {
        await sendRandomRequest()
        await delay(1000)
    }
    request.destroy()
}

main().catch(err => console.error(err))

node replier.js
node requestor.js
node requestor.js