RabbitMQ を使用してメッセージ確認を実装し、Golang の信頼性を確保するためのヒント

package main

import (
    "log"

    "github.com/streadway/amqp"
)

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

    // ...
}

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

    // ...

    queueName := "my_queue"
    _, err = channel.QueueDeclare(
        queueName, // 队列名
        true,      // 是否持久化
        false,     // 是否自动删除
        false,     // 是否排他
        false,     // 是否等待服务器完成的通知
        nil,       // 额外的属性
    )
    if err != nil {
        log.Fatalf("Failed to declare a queue: %v", err)
    }

    // ...

    err = channel.Publish(
        "",        // 交换机名称
        queueName, // 队列名称
        false,     // 是否等待服务端确认
        false,     // 是否等待生产者确认
        amqp.Publishing{
            ContentType: "text/plain",
            Body:        []byte("Hello, RabbitMQ!"),
        },
    )
    if err != nil {
        log.Fatalf("Failed to publish a message: %v", err)
    }

    // ...

    messages, err := channel.Consume(
        queueName, // 队列名称
        "",        // 消费者标签
        true,      // 是否自动确认
        false,     // 是否排他
        false,     // 是否阻塞等待
        false,     // 额外的属性
        nil,       // 可选的回调函数
    )
    if err != nil {
        log.Fatalf("Failed to register a consumer: %v", err)
    }

    go func() {
        for msg := range messages {
            log.Printf("Received a message: %s", msg.Body)
        }
    }()

    // ...

    go func() {
        for msg := range messages {
            log.Printf("Received a message: %s", msg.Body)

            // 模拟处理逻辑
            time.Sleep(time.Second * 2)

            // 确认消息
            msg.Ack(false)
        }
    }()

    go func() {
        for msg := range messages {
            log.Printf("Received a message: %s", msg.Body)

            // 模拟处理逻辑
            time.Sleep(time.Second * 2)

            // 如果发生错误，则重新放入队列
            if err := handleMsg(msg.Body); err != nil {
                msg.Nack(false, true)
            } else {
                msg.Ack(false)
            }
        }
    }()

    // ...