import rabbitmq from "rabbitmq"
//创建连接
let connection = rabbitmq.open("amqp://admin:123456@localhost:5672/")
//  创建通道
let ch = connection.channel()
//声明队列
let queue = ch.queueDeclare({name: "queue1"})
//发布消息
ch.publish({key: "queue1", msg: {contextType: "text/plain", body: "hello world"}})
//订阅消息
ch.consume({queue: "queue1"}, (msg) => {
    //收到并处理消息
    console.log(msg)
})
//线程暂停
sleep()

import rabbitmq from "rabbitmq"
//创建连接
let connection = rabbitmq.open("amqp://admin:123456@localhost:5672/")

import rabbitmq from "rabbitmq"
//创建连接
let connection = rabbitmq.open("amqp://admin:123456@localhost:5672/")
//  打开一个频道
let ch = connection.channel()

import rabbitmq from "rabbitmq"
//创建连接
let connection = rabbitmq.open("amqp://admin:123456@localhost:5672/")
//  关闭连接
connection.close()

import rabbitmq from "rabbitmq"
//创建连接
let connection = rabbitmq.open("amqp://admin:123456@localhost:5672/")
// 判断连接是否关闭，下面输出： false
console.log(connection.isClosed())
//  关闭连接
connection.close()
// 下面输出： true
console.log(connection.isClosed())

import rabbitmq from "rabbitmq"

// 创建连接和通道
let conn = rabbitmq.open("amqp://admin:123456@localhost:5672/")
let ch = conn.channel()

// 消费消息
let consumer = ch.consume({queue: "queue1"}, (msg) => {
    // 处理消息...

    // 确认单条消息
    ch.ack(msg.deliveryTag, false)

    // 或者确认此消息及之前的所有消息
    // ch.ack(msg.deliveryTag, true)
})

import rabbitmq from "rabbitmq"

// 创建连接和通道
let conn = rabbitmq.open("amqp://admin:123456@localhost:5672/")
let ch = conn.channel()

// 开始消费
let consumer = ch.consume({queue: "queue1"}, (msg) => {
    console.log("收到消息:", msg)
})

// 一段时间后取消消费者
setTimeout(() => {
    // 不等待服务器响应
    ch.cancel(consumer, true)

    // 或者等待服务器响应
    // ch.cancel(consumer, false)

    console.log("消费者已取消")
}, 10000)

import rabbitmq from "rabbitmq"

// 创建连接和通道
let conn = rabbitmq.open("amqp://admin:123456@localhost:5672/")
let ch = conn.channel()

// 使用通道进行各种操作...

// 关闭通道
ch.close()

// 此时如果再使用该通道会抛出异常

import rabbitmq from "rabbitmq"

// 创建连接和通道
let conn = rabbitmq.open("amqp://admin:123456@localhost:5672/")
let ch = conn.channel()

// 启用发布确认模式（等待服务器响应）
ch.confirm(false)

// 启用发布确认模式（不等待服务器响应）
// ch.confirm(true)

// 现在可以使用发布确认相关功能
ch.publish({
    exchange: "",
    key: "queue1",
    msg: {body: "Hello World"}
})

import rabbitmq from "rabbitmq"

// 创建连接和通道
let conn = rabbitmq.open("amqp://admin:123456@localhost:5672/")
let ch = conn.channel()

// 消费消息，不自动确认
ch.consume({
    queue: "queue1",
    consumer: "myConsumer",  // 可选
    autoAck: false,          // 手动确认
    exclusive: false,
    noLocal: false,
    noWait: false,
    args: {}                 // 可选参数
}, (msg) => {
    console.log("收到消息:", msg.body)
    console.log("来自交换机:", msg.exchange)
    console.log("路由键:", msg.routingKey)

    // 处理消息...

    // 手动确认消息
    msg.ack(false)  // 只确认此条消息

    // 或者拒绝消息并重新入队
    // msg.reject(true)

    // 或者否定确认并重新入队
    // msg.nack(false, true)
})

import rabbitmq from "rabbitmq"

// 创建连接和通道
let conn = rabbitmq.open("amqp://admin:123456@localhost:5672/")
let ch = conn.channel()

// 声明两个交换机
ch.exchangeDeclare({
    name: "logs",
    kind: "fanout"
})

ch.exchangeDeclare({
    name: "backup",
    kind: "fanout"
})

// 将 backup 交换机绑定到 logs 交换机
ch.exchangeBind({
    destination: "backup",
    source: "logs",
    key: "",          // fanout 交换机忽略 routing key
    noWait: false,
    args: {}          // 可选参数
})

import rabbitmq from "rabbitmq"

// 创建连接和通道
let conn = rabbitmq.open("amqp://admin:123456@localhost:5672/")
let ch = conn.channel()

// 声明一个持久化的 direct 交换机
ch.exchangeDeclare({
    name: "direct_logs",
    kind: "direct",
    durable: true,
    autoDelete: false,
    internal: false,
    noWait: false,
    args: {
        "alternate-exchange": "ae"  // 设置备用交换机
    }
})

// 声明一个自动删除的 fanout 交换机
ch.exchangeDeclare({
    name: "fanout_logs",
    kind: "fanout",
    durable: false,
    autoDelete: true
})

import rabbitmq from "rabbitmq"

// 创建连接和通道
let conn = rabbitmq.open("amqp://admin:123456@localhost:5672/")
let ch = conn.channel()

try {
    // 检查交换机是否存在
    ch.exchangeDeclarePassive({
        name: "existing_exchange",
        kind: "direct"
    })
    console.log("交换机存在")
} catch (err) {
    console.log("交换机不存在或发生错误:", err.message)
    // 如果不存在，可以创建它
    ch.exchangeDeclare({
        name: "existing_exchange",
        kind: "direct"
    })
}

import rabbitmq from "rabbitmq"

// 创建连接和通道
let conn = rabbitmq.open("amqp://admin:123456@localhost:5672/")
let ch = conn.channel()

// 删除交换机（无条件）
ch.exchangeDelete({
    name: "unused_exchange"
})

// 只在没有绑定时删除
ch.exchangeDelete({
    name: "maybe_used_exchange",
    ifUnused: true
})

import rabbitmq from "rabbitmq"

// 创建连接和通道
let conn = rabbitmq.open("amqp://admin:123456@localhost:5672/")
let ch = conn.channel()

// 解除交换机绑定
ch.exchangeUnbind({
    destination: "backup",
    source: "logs",
    key: "",
    noWait: false,
    args: {}  // 必须与绑定时使用的参数一致
})

import rabbitmq from "rabbitmq"

// 创建连接和通道
let conn = rabbitmq.open("amqp://admin:123456@localhost:5672/")
let ch = conn.channel()

// 暂停消息流
ch.flow(false)
console.log("消息流已暂停")

// 一段时间后恢复消息流
setTimeout(() => {
    ch.flow(true)
    console.log("消息流已恢复")
}, 10000)

import rabbitmq from "rabbitmq"

// 创建连接和通道
let conn = rabbitmq.open("amqp://admin:123456@localhost:5672/")
let ch = conn.channel()

// 声明队列
ch.queueDeclare({
    name: "queue1",
    durable: false
})

// 发布一条消息
ch.publish({
    exchange: "",
    key: "queue1",
    msg: {body: "Hello World"}
})

// 获取消息（不自动确认）
let msg = ch.get("queue1", false)
if (msg) {
    console.log("获取到消息:", msg.body)
    console.log("队列中剩余消息数:", msg.messageCount)

    // 确认消息
    msg.ack(false)

    // 或者拒绝消息并重新入队
    // msg.reject(true)
} else {
    console.log("队列中没有消息")
}

// 再次获取（应该为空）
let msg2 = ch.get("queue1", false)
if (!msg2) {
    console.log("队列已空")
}

import rabbitmq from "rabbitmq"

// 创建连接和通道
let conn = rabbitmq.open("amqp://admin:123456@localhost:5672/")
let ch = conn.channel()

// 启用发布确认模式
ch.confirm(false)

// 获取下一个发布序列号
let seqNo = ch.getNextPublishSeqNo()
console.log("下一个发布序列号:", seqNo)

// 发布消息
ch.publish({
    exchange: "",
    key: "queue1",
    msg: {body: "Hello World"}
})

// 序列号会增加
let nextSeqNo = ch.getNextPublishSeqNo()
console.log("下一个发布序列号:", nextSeqNo)

import rabbitmq from "rabbitmq"

// 创建连接和通道
let conn = rabbitmq.open("amqp://admin:123456@localhost:5672/")
let ch = conn.channel()

// 检查通道状态
console.log("通道是否关闭:", ch.isClosed()) // false

// 使用通道...

// 关闭通道
ch.close()

// 再次检查
console.log("通道是否关闭:", ch.isClosed()) // true

import rabbitmq from "rabbitmq"

// 创建连接和通道
let conn = rabbitmq.open("amqp://admin:123456@localhost:5672/")
let ch = conn.channel()

// 消费消息（手动确认）
ch.consume({queue: "queue1", autoAck: false}, (msg) => {
    console.log("收到消息:", msg.body)

    try {
        // 处理消息...
        // 假设处理失败，需要拒绝消息

        // 拒绝单条消息并重新入队
        ch.nack(msg.deliveryTag, false, true)

        // 或者拒绝多条消息并丢弃
        // ch.nack(msg.deliveryTag, true, false)

    } catch (err) {
        console.log("处理消息失败，消息已重新入队")
    }
})

import rabbitmq from "rabbitmq"

// 创建连接和通道
let conn = rabbitmq.open("amqp://admin:123456@localhost:5672/")
let ch = conn.channel()

// 注册消费者取消通知
ch.notifyCancel((consumerTag) => {
    console.log("消费者被取消:", consumerTag)
    // 可以进行清理操作...
})

// 开始消费
let consumerTag = ch.consume({queue: "queue1"}, (msg) => {
    console.log("收到消息:", msg.body)
    msg.ack(false)
})

// 一段时间后取消消费者（会触发通知）
setTimeout(() => {
    ch.cancel(consumerTag, false)
}, 10000)

import rabbitmq from "rabbitmq"

// 创建连接和通道
let conn = rabbitmq.open("amqp://admin:123456@localhost:5672/")
let ch = conn.channel()

// 注册通道关闭通知
ch.notifyClose((err) => {
    if (err) {
        console.log("通道因错误关闭:")
        console.log("错误代码:", err.code)
        console.log("错误原因:", err.reason)
        console.log("是否服务器错误:", err.server)
        console.log("是否可恢复:", err.recover)
    } else {
        console.log("通道正常关闭")
    }
})

// 执行一些操作...

// 正常关闭通道（会触发通知，err为null）
ch.close()

// 或者通道可能因错误关闭，例如服务器断开连接

import rabbitmq from "rabbitmq"

// 创建连接和通道
let conn = rabbitmq.open("amqp://admin:123456@localhost:5672/")
let ch = conn.channel()

// 启用发布确认模式
ch.confirm(false)

// 注册确认通知
ch.notifyConfirm(
    (tag) => {
        console.log("消息发布成功，标签:", tag)
    },
    (tag) => {
        console.log("消息发布失败，标签:", tag)
    }
)

// 发布消息（启用确认模式后）
ch.publish({
    exchange: "",
    key: "queue1",
    msg: {body: "Hello World"}
})

ch.publish({
    exchange: "",
    key: "queue1",
    msg: {body: "Another Message"}
})

import rabbitmq from "rabbitmq"

// 创建连接和通道
let conn = rabbitmq.open("amqp://admin:123456@localhost:5672/")
let ch = conn.channel()

// 注册流量控制通知
ch.notifyFlow((active) => {
    if (active) {
        console.log("消息流已恢复，可以继续发送消息")
    } else {
        console.log("消息流已暂停，暂时不能发送消息")
    }
})

// 消费消息
ch.consume({queue: "queue1"}, (msg) => {
    console.log("收到消息:", msg.body)
    msg.ack(false)
})

// 注意：流量控制通常由RabbitMQ服务器触发，
// 当服务器负载过高时可能会暂停消息流

import rabbitmq from "rabbitmq"

// 创建连接和通道
let conn = rabbitmq.open("amqp://admin:123456@localhost:5672/")
let ch = conn.channel()

// 启用发布确认模式
ch.confirm(false)

// 注册发布确认通知
ch.notifyPublish((confirmation) => {
    if (confirmation.ack) {
        console.log("消息已确认，标签:", confirmation.deliveryTag)
    } else {
        console.log("消息未确认（nack），标签:", confirmation.deliveryTag)
        // 可以进行重试等操作
    }
})

// 发布多条消息
for (let i = 0; i < 5; i++) {
    ch.publish({
        exchange: "",
        key: "queue1",
        msg: {body: `Message ${i}`}
    })
}

// 等待所有确认
setTimeout(() => {
    console.log("所有消息确认完成")
}, 1000)

import rabbitmq from "rabbitmq"

// 创建连接和通道
let conn = rabbitmq.open("amqp://admin:123456@localhost:5672/")
let ch = conn.channel()

// 注册返回通知
ch.notifyReturn((returned) => {
    console.log("消息无法路由，已返回:")
    console.log("返回代码:", returned.replyCode)
    console.log("返回原因:", returned.replyText)
    console.log("交换机:", returned.exchange)
    console.log("路由键:", returned.routingKey)
    console.log("消息内容:", returned.body)
})

// 发布消息，设置mandatory标志为true
// 如果消息无法路由到队列，会触发返回通知
ch.publish({
    exchange: "",
    key: "non.existent.queue",
    mandatory: true,  // 重要：必须设置为true才能触发返回
    immediate: false,
    msg: {
        contentType: "text/plain",
        body: "This message cannot be routed"
    }
})

import rabbitmq from "rabbitmq"

// 创建连接和通道
let conn = rabbitmq.open("amqp://admin:123456@localhost:5672/")
let ch = conn.channel()

// 声明队列
ch.queueDeclare({
    name: "queue1",
    durable: false
})

// 发布简单消息
ch.publish({
    exchange: "",
    key: "queue1",
    mandatory: false,
    immediate: false,
    msg: {
        contentType: "text/plain",
        body: "Hello World"
    }
})

// 发布到指定交换机
ch.exchangeDeclare({
    name: "logs",
    kind: "fanout"
})

ch.publish({
    exchange: "logs",
    key: "",  // fanout交换机忽略路由键
    msg: {
        contentType: "text/plain",
        body: "Log message"
    }
})

import rabbitmq from "rabbitmq"

// 创建连接和通道
let conn = rabbitmq.open("amqp://admin:123456@localhost:5672/")
let ch = conn.channel()

// 启用发布确认模式
ch.confirm(false)

// 发布消息并获取延迟确认对象
let deferredConf = ch.publishWithDeferredConfirm({
    exchange: "",
    key: "queue1",
    mandatory: false,
    immediate: false,
    msg: {
        contentType: "text/plain",
        body: "Hello World"
    }
})

console.log("消息已发布，投递标签:", deferredConf.deliveryTag)

// 在需要时等待确认
setTimeout(() => {
    let isConfirmed = deferredConf.confirmed()
    if (isConfirmed) {
        console.log("消息已确认")
    } else {
        console.log("消息未被确认（nack）")
    }
}, 1000)

import rabbitmq from "rabbitmq"

// 创建连接和通道
let conn = rabbitmq.open("amqp://admin:123456@localhost:5672/")
let ch = conn.channel()

// 设置预取：每个消费者最多同时处理1条消息
ch.qos(1, 0, false)

// 设置预取：整个通道最多同时处理10条消息
ch.qos(10, 0, true)

// 消费消息（将受QoS设置影响）
ch.consume({queue: "queue1"}, (msg) => {
    console.log("处理消息:", msg.body)

    // 模拟耗时处理
    setTimeout(() => {
        msg.ack(false)
        console.log("消息处理完成")
    }, 2000)
})

import rabbitmq from "rabbitmq"

// 创建连接和通道
let conn = rabbitmq.open("amqp://admin:123456@localhost:5672/")
let ch = conn.channel()

// 声明队列和交换机
ch.queueDeclare({
    name: "error_logs",
    durable: false
})

ch.exchangeDeclare({
    name: "direct_logs",
    kind: "direct"
})

// 将队列绑定到交换机，只接收错误级别的日志
ch.queueBind({
    name: "error_logs",
    key: "error",      // 路由键为"error"
    exchange: "direct_logs",
    noWait: false,
    args: {}           // 可选参数
})

// 再绑定一个队列接收所有日志
ch.queueDeclare({
    name: "all_logs",
    durable: false
})

ch.queueBind({
    name: "all_logs",
    key: "#",          // "#"匹配所有路由键
    exchange: "direct_logs"
})

import rabbitmq from "rabbitmq"

// 创建连接和通道
let conn = rabbitmq.open("amqp://admin:123456@localhost:5672/")
let ch = conn.channel()

// 声明一个持久化队列
let queueInfo = ch.queueDeclare({
    name: "durable_queue",
    durable: true,
    autoDelete: false,
    exclusive: false,
    args: {
        "x-max-length": 10000  // 限制队列最大长度
    }
})

console.log("队列声明成功:", queueInfo.name)
console.log("当前消费者数:", queueInfo.consumers)
console.log("当前消息数:", queueInfo.messages)

// 声明一个自动生成的临时队列
let tempQueue = ch.queueDeclare({
    name: "",  // 空字符串表示自动生成
    durable: false,
    autoDelete: true,
    exclusive: true
})

console.log("自动生成的队列名:", tempQueue.name)

import rabbitmq from "rabbitmq"

// 创建连接和通道
let conn = rabbitmq.open("amqp://admin:123456@localhost:5672/")
let ch = conn.channel()

try {
    // 检查队列是否存在
    let queueInfo = ch.queueDeclarePassive({
        name: "existing_queue"
    })

    console.log("队列存在:")
    console.log("名称:", queueInfo.name)
    console.log("消费者数:", queueInfo.consumers)
    console.log("消息数:", queueInfo.messages)
} catch (err) {
    console.log("队列不存在或发生错误:", err.message)

    // 如果不存在，可以创建它
    ch.queueDeclare({
        name: "existing_queue",
        durable: false
    })
}

import rabbitmq from "rabbitmq"

// 创建连接和通道
let conn = rabbitmq.open("amqp://admin:123456@localhost:5672/")
let ch = conn.channel()

// 创建测试队列并发布一些消息
let queueInfo = ch.queueDeclare({
    name: "test_queue",
    durable: false
})

// 发布5条消息
for (let i = 0; i < 5; i++) {
    ch.publish({
        exchange: "",
        key: "test_queue",
        msg: {body: `Message ${i}`}
    })
}

// 无条件删除队列（返回被删除的消息数）
let deletedCount = ch.queueDelete({
    name: "test_queue"
})
console.log(`队列已删除，删除了${deletedCount}条消息`)

// 创建新队列
ch.queueDeclare({
    name: "another_queue",
    durable: false
})

// 只在队列为空时删除
try {
    let count = ch.queueDelete({
        name: "another_queue",
        ifEmpty: true
    })
    console.log(`队列已删除，删除了${count}条消息`)
} catch (err) {
    console.log("删除失败（可能队列不为空）:", err.message)
}

import rabbitmq from "rabbitmq"

// 创建连接和通道
let conn = rabbitmq.open("amqp://admin:123456@localhost:5672/")
let ch = conn.channel()

// 检查队列状态
try {
    let queueInfo = ch.queueInspect("my_queue")

    console.log("队列状态:")
    console.log("名称:", queueInfo.name)
    console.log("消费者数:", queueInfo.consumers)
    console.log("消息数:", queueInfo.messages)

    if (queueInfo.messages > 1000) {
        console.log("警告: 队列积压了大量消息!")
    }
} catch (err) {
    console.log("队列不存在:", err.message)
}

import rabbitmq from "rabbitmq"

// 创建连接和通道
let conn = rabbitmq.open("amqp://admin:123456@localhost:5672/")
let ch = conn.channel()

// 先向队列发布一些消息
for (let i = 0; i < 10; i++) {
    ch.publish({
        exchange: "",
        key: "queue_to_purge",
        msg: {body: `Message ${i}`}
    })
}

// 清空队列
let purgedCount = ch.queuePurge("queue_to_purge", false)
console.log(`清空了${purgedCount}条消息`)

// 验证队列已空
let queueInfo = ch.queueInspect("queue_to_purge")
console.log("队列中剩余消息:", queueInfo.messages) // 应为0

import rabbitmq from "rabbitmq"

// 创建连接和通道
let conn = rabbitmq.open("amqp://admin:123456@localhost:5672/")
let ch = conn.channel()

// 声明队列和交换机
ch.queueDeclare({
    name: "logs_queue",
    durable: false
})

ch.exchangeDeclare({
    name: "logs_exchange",
    kind: "fanout"
})

// 绑定队列到交换机
ch.queueBind({
    name: "logs_queue",
    key: "",  // fanout交换机忽略路由键
    exchange: "logs_exchange"
})

// ... 一段时间后，解除绑定
ch.queueUnbind({
    name: "logs_queue",
    key: "",
    exchange: "logs_exchange",
    args: {}  // 必须与绑定时使用的参数一致
})

console.log("队列已从交换机解绑")

import rabbitmq from "rabbitmq"

// 创建连接和通道
let conn = rabbitmq.open("amqp://admin:123456@localhost:5672/")
let ch = conn.channel()

// 消费消息（不自动确认）
ch.consume({queue: "queue1", autoAck: false}, (msg) => {
    console.log("收到消息:", msg.body)

    // 不确认消息，模拟消费者崩溃
    // 注意：这里故意不调用 msg.ack() 或 msg.nack()
})

// 一段时间后，由于消费者崩溃，我们可以重新投递未确认的消息
setTimeout(() => {
    try {
        // 重新投递未确认的消息（重新入队）
        ch.recover(true)
        console.log("已重新投递未确认的消息")
    } catch (err) {
        console.log("重新投递失败:", err.message)
    }
}, 5000)

import rabbitmq from "rabbitmq"

// 创建连接和通道
let conn = rabbitmq.open("amqp://admin:123456@localhost:5672/")
let ch = conn.channel()

// 消费消息（手动确认）
ch.consume({queue: "queue1", autoAck: false}, (msg) => {
    console.log("收到消息:", msg.body)

    // 检查消息是否有效
    if (msg.body.includes("error")) {
        console.log("发现错误消息，拒绝并丢弃")

        // 拒绝消息并丢弃（不重新入队）
        ch.reject(msg.deliveryTag, false)
    } else if (msg.body.includes("retry")) {
        console.log("需要重试的消息，拒绝并重新入队")

        // 拒绝消息并重新入队
        ch.reject(msg.deliveryTag, true)
    } else {
        // 正常处理消息
        console.log("正常处理消息")
        msg.ack(false)
    }
})

import rabbitmq from "rabbitmq"

// 创建连接和通道
let conn = rabbitmq.open("amqp://admin:123456@localhost:5672/")
let ch = conn.channel()

// 启动事务模式
ch.tx()

console.log("事务模式已启动")

// 现在所有发布和确认操作都在事务中
// 可以执行多个操作...

// 提交或回滚事务

import rabbitmq from "rabbitmq"

// 创建连接和通道
let conn = rabbitmq.open("amqp://admin:123456@localhost:5672/")
let ch = conn.channel()

// 启动事务模式
ch.tx()

try {
    // 执行一系列操作
    ch.publish({
        exchange: "",
        key: "queue1",
        msg: {body: "Message 1"}
    })

    ch.publish({
        exchange: "",
        key: "queue1",
        msg: {body: "Message 2"}
    })

    // 确认一些消息（如果在同一个事务中）
    // ch.ack(someTag, false)

    // 提交事务（所有操作要么全部成功，要么全部失败）
    ch.txCommit()

    console.log("事务提交成功")
} catch (err) {
    console.log("操作失败，事务将回滚:", err.message)
    ch.txRollback()
}

import rabbitmq from "rabbitmq"

// 创建连接和通道
let conn = rabbitmq.open("amqp://admin:123456@localhost:5672/")
let ch = conn.channel()

// 启动事务模式
ch.tx()

try {
    // 执行第一个操作
    ch.publish({
        exchange: "",
        key: "queue1",
        msg: {body: "Message 1"}
    })

    // 执行第二个操作（可能失败）
    // 假设这个操作失败了
    throw new Error("模拟操作失败")

    // 如果到达这里，提交事务
    ch.txCommit()
} catch (err) {
    console.log("操作失败:", err.message)

    // 回滚事务（之前的所有操作都会被撤销）
    ch.txRollback()

    console.log("事务已回滚")
}

// 验证队列是否真的收到了消息
// 由于事务回滚，队列应该没有收到任何消息
let queueInfo = ch.queueInspect("queue1")
console.log("队列中的消息数:", queueInfo.messages) // 应为0