golang中的rabbitmq以及vhost示例详解

go get github.com/streadway/amqp

conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
if err != nil {
    log.Fatalf("Failed to connect to RabbitMQ: %v", err)
}
defer conn.Close()

ch, err := conn.Channel()
if err != nil {
    log.Fatalf("Failed to open a channel: %v", err)
}
defer ch.Close()

q, err := ch.QueueDeclare(
    "task_queue", // 队列名称
    true,         // 是否持久化
    false,        // 是否自动删除
    false,        // 是否独占
    false,        // 是否阻塞
    nil,          // 额外参数
)
if err != nil {
    log.Fatalf("Failed to declare a queue: %v", err)
}

body := "Hello, RabbitMQ!"
err = ch.Publish(
    "",     // 交换机名称（空字符串表示默认交换机）
    q.Name, // 路由键（队列名称）
    false,  // 是否强制路由
    false,  // 是否立即发送
    amqp.Publishing{
        DeliveryMode: amqp.Persistent, // 消息持久化
        ContentType:  "text/plain",
        Body:         []byte(body),
    },
)
if err != nil {
    log.Fatalf("Failed to publish a message: %v", err)
}

msgs, err := ch.Consume(
    q.Name, // 队列名称
    "",     // 消费者标签
    false,  // 是否自动确认
    false,  // 是否独占
    false,  // 是否阻塞
    false,  // 额外参数
    nil,
)
if err != nil {
    log.Fatalf("Failed to register a consumer: %v", err)
}
forever := make(chan bool)
go func() {
    for d := range msgs {
        log.Printf("Received a message: %s", d.Body)
        d.Ack(false) // 手动确认消息
    }
}()
log.Printf("Waiting for messages...")
<-forever

err = ch.ExchangeDeclare(
    "logs_direct", // 交换机名称
    "direct",      // 交换机类型
    true,          // 是否持久化
    false,         // 是否自动删除
    false,         // 是否内部使用
    false,         // 是否阻塞
    nil,
)
if err != nil {
    log.Fatalf("Failed to declare an exchange: %v", err)
}
err = ch.QueueBind(
    q.Name,        // 队列名称
    "error",       // 路由键
    "logs_direct", // 交换机名称
    false,         // 是否阻塞
    nil,
)
if err != nil {
    log.Fatalf("Failed to bind a queue: %v", err)
}

for d := range msgs {
    if err := processMessage(d.Body); err != nil {
        d.Nack(false, true) // 拒绝消息并重新入队
    } else {
        d.Ack(false) // 确认消息
    }
}

notifyClose := make(chan *amqp.Error)
ch.NotifyClose(notifyClose)
go func() {
    for err := range notifyClose {
        log.Printf("Channel closed: %v", err)
        // 实现重连逻辑
    }
}()

err = ch.Qos(
    1,     // 预取计数
    0,     // 预取大小
    false, // 是否全局
)
if err != nil {
    log.Fatalf("Failed to set QoS: %v", err)
}

# 创建VHost
rabbitmqctl add_vhost /my_vhost
# 列出所有VHost
rabbitmqctl list_vhosts
# 删除VHost
rabbitmqctl delete_vhost /my_vhost

import pika
connection = pika.BlockingConnection(pika.ConnectionParameters(
    host='localhost',
    virtual_host='/my_vhost'
))

rabbitmqctl set_permissions -p /my_vhost username ".*" ".*" ".*"