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", // 隊(duì)列名稱
    true,         // 是否持久化
    false,        // 是否自動(dòng)刪除
    false,        // 是否獨(dú)占
    false,        // 是否阻塞
    nil,          // 額外參數(shù)
)
if err != nil {
    log.Fatalf("Failed to declare a queue: %v", err)
}

body := "Hello, RabbitMQ!"
err = ch.Publish(
    "",     // 交換機(jī)名稱（空字符串表示默認(rèn)交換機(jī)）
    q.Name, // 路由鍵（隊(duì)列名稱）
    false,  // 是否強(qiáng)制路由
    false,  // 是否立即發(fā)送
    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, // 隊(duì)列名稱
    "",     // 消費(fèi)者標(biāo)簽
    false,  // 是否自動(dòng)確認(rèn)
    false,  // 是否獨(dú)占
    false,  // 是否阻塞
    false,  // 額外參數(shù)
    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) // 手動(dòng)確認(rèn)消息
    }
}()
log.Printf("Waiting for messages...")
<-forever

err = ch.ExchangeDeclare(
    "logs_direct", // 交換機(jī)名稱
    "direct",      // 交換機(jī)類型
    true,          // 是否持久化
    false,         // 是否自動(dòng)刪除
    false,         // 是否內(nèi)部使用
    false,         // 是否阻塞
    nil,
)
if err != nil {
    log.Fatalf("Failed to declare an exchange: %v", err)
}
err = ch.QueueBind(
    q.Name,        // 隊(duì)列名稱
    "error",       // 路由鍵
    "logs_direct", // 交換機(jī)名稱
    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) // 拒絕消息并重新入隊(duì)
    } else {
        d.Ack(false) // 確認(rèn)消息
    }
}

notifyClose := make(chan *amqp.Error)
ch.NotifyClose(notifyClose)
go func() {
    for err := range notifyClose {
        log.Printf("Channel closed: %v", err)
        // 實(shí)現(xiàn)重連邏輯
    }
}()

err = ch.Qos(
    1,     // 預(yù)取計(jì)數(shù)
    0,     // 預(yù)取大小
    false, // 是否全局
)
if err != nil {
    log.Fatalf("Failed to set QoS: %v", err)
}

# 創(chuàng)建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 ".*" ".*" ".*"