golang框架在分布式系统中实现消息队列的实战

package main

import (
    "fmt"
    "sync"

    "github.com/streadway/amqp"
)

// 定义消息队列
type Queue struct {
    sync.Mutex
    messages []amqp.Delivery
}

func main() {
    conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
    if err != nil {
        panic(err)
    }
    defer conn.Close()

    ch, err := conn.Channel()
    if err != nil {
        panic(err)
    }
    defer ch.Close()

    q, err := ch.QueueDeclare(
        "my-queue", // 队列名称
        false,       // 是否持久化
        false,       // 是否独占
        false,       // 是否自动删除
        false,       // 是否等待服务器确认
        nil,          // 其他参数
    )
    if err != nil {
        panic(err)
    }

    // 发送消息
    body := "Hello, world!"
    err = ch.Publish(
        "",        // 交换机名称（空字符串表示使用默认交换机）
        q.Name,    // 队列名称
        false,     // 是否强制路由（即即使队列不存在也不失败）
        false,     // 是否使用持久化发布
        amqp.Publishing{
            ContentType: "text/plain",
            Body:        []byte(body),
        },
    )
    if err != nil {
        panic(err)
    }
    fmt.Println("Sent message:", body)

    // 接收消息
    msgs, err := ch.Consume(
        q.Name, // 队列名称
        "",     // 消费者名称（空字符串表示使用默认名称）
        true,   // 是否自动确认
        false,  // 是否独占
        false,  // 是否无等待本地
        false,  // 是否全局
        nil,    // 其他参数
    )
    if err != nil {
        panic(err)
    }

    forever := make(chan bool)
    go func() {
        for d := range msgs {
            fmt.Println("Received message:", string(d.Body))
        }
    }()
    <-forever
}

package main

import (
    "bytes"
    "context"
    "encoding/json"
    "log"
    "time"

    "github.com/streadway/amqp"
)

type Message struct {
    ID        string
    Payload   string
    Timestamp time.Time
}

func main() {
    ctx := context.Background()

    conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
    if err != nil {
        log.Fatal(err)
    }
    defer conn.Close()

    ch, err := conn.Channel()
    if err != nil {
        log.Fatal(err)
    }
    defer ch.Close()

    q, err := ch.QueueDeclare(
        "my-queue", // 队列名称
        true,       // 持久化
        false,      // 不独占
        false,      // 不自动删除
        false,      // 不等待服务器确认
        nil,        // 其他参数
    )
    if err != nil {
        log.Fatal(err)
    }

    // 模拟发送消息（使用 JSON 序列化）
    msg := Message{
        ID:        "123",
        Payload:   "Hello, world!",
        Timestamp: time.Now(),
    }
    body, err := json.Marshal(msg)
    if err != nil {
        log.Fatal(err)
    }
    err = ch.Publish(
        "",        // 交换机名称（空字符串表示使用默认交换机）
        q.Name,    // 队列名称
        false,     // 是否强制路由
        false,     // 是否持久化发布
        amqp.Publishing{
            ContentType: "application/json",
            Body:        body,
        },
    )
    if err != nil {
        log.Fatal(err)
    }

    // 模拟接收消息并解序列化
    msgs, err := ch.Consume(
        q.Name, // 队列名称
        "",     // 消费者名称（空字符串表示使用默认名称）
        true,   // 自动确认
        false,  // 不独占
        false,  // 无等待本地
        false,  // 全局
        nil,    // 其他参数
    )
    if err != nil {
        log.Fatal(err)
    }

    for d := range msgs {
        var msg Message
        err := json.NewDecoder(bytes.NewReader(d.Body)).Decode(&msg)
        if err != nil {
            log.Fatal(err)
        }
        log.Printf("Received message: %s", msg.Payload)
    }
}