ic-ctbu-backend/plugin/mq/index.js

const amqp = require('amqplib')
const path = require('path')
const config = require('../../config.json')
const Logger = require('../../lib/Logger')
const { mq } = require('./mqPrefix')

class MQManager {
    constructor() {
        this.url = config.rabbitmq.url
        this.connection = null
        this.channels = new Map()
        this.logger = new Logger(path.join(__dirname, '../../logs/RabbitMQ.log'), 'INFO')
        this.reconnecting = false
    }

    async init() {
        if (this.connection) return

        try {
            this.logger.info('RabbitMQ 初始化连接...')
            this.connection = await amqp.connect(this.url)

            this.connection.on('close', () => {
                this.logger.warn('RabbitMQ 连接断开，准备重连')
                this.connection = null
                this.channels.clear()
                this.reconnect()
            })

            this.connection.on('error', (err) => {
                this.logger.error('RabbitMQ 连接错误:', err.message)
            })

            this.logger.info('RabbitMQ 连接成功')
        } catch (e) {
            this.logger.error('RabbitMQ 初始化失败:', e.message)
            this.reconnect()
            throw e
        }
    }

    async reconnect() {
        if (this.reconnecting) return
        this.reconnecting = true

        setTimeout(async () => {
            try {
                await this.init()
                this.reconnecting = false
            } catch {
                this.reconnecting = false
            }
        }, 5000)
    }

    async getChannel(name = 'default') {
        if (!this.connection) {
            await this.init()
        }

        const key = mq(name)
        if (this.channels.has(key)) {
            const cached = this.channels.get(key)
            if (!cached.__runforgeClosed) return cached
            this.channels.delete(key)
        }

        const channel = await this.connection.createChannel()
        channel.__runforgeClosed = false
        this.channels.set(key, channel)

        channel.on('close', () => {
            channel.__runforgeClosed = true
            this.logger.warn(`Channel [${key}] 已关闭`)
            this.channels.delete(key)
        })

        channel.on('error', (err) => {
            channel.__runforgeClosed = true
            this.logger.error(`Channel [${key}] 错误: ${err.message || err}`)
            this.channels.delete(key)
        })

        return channel
    }

    invalidateChannel(name = 'default') {
        const key = mq(name)
        const channel = this.channels.get(key)
        if (channel) {
            channel.__runforgeClosed = true
            this.channels.delete(key)
        }
    }

    async close() {
        for (const ch of this.channels.values()) {
            await ch.close()
        }
        this.channels.clear()

        if (this.connection) {
            await this.connection.close()
            this.connection = null
        }
    }
}

module.exports = new MQManager()