消息收发模式

中间件

RabbitMQ

消息收发模式

对应 RabbitMQ 官方六大教程的完整 Go 实现：Hello World、Work Queue（工作队列）、Pub/Sub（Fanout 广播）、Routing（Direct 路由）、Topics（主题通配符）与 RPC（请求/响应）。

Hello World — 最简单的生产者和消费者

send.go（生产者）：

package main

import (
    "context"
    "log"
    rmq "github.com/rabbitmq/rabbitmq-amqp-go-client/pkg/rabbitmqamqp"
)

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

    // 1. 创建 Environment
    env := rmq.NewEnvironment("amqp://guest:guest@localhost:5672/", nil)
    defer env.CloseConnections(context.Background())

    // 2. 建立连接
    conn, err := env.NewConnection(ctx)
    if err != nil {
        log.Panicf("连接失败: %v", err)
    }

    // 3. 声明队列
    _, err = conn.Management().DeclareQueue(ctx,
        &rmq.QuorumQueueSpecification{Name: "hello"})
    if err != nil {
        log.Panicf("声明队列失败: %v", err)
    }

    // 4. 创建发布者
    publisher, err := conn.NewPublisher(ctx,
        &rmq.QueueAddress{Queue: "hello"}, nil)
    if err != nil {
        log.Panicf("创建发布者失败: %v", err)
    }
    defer publisher.Close(context.Background())

    // 5. 发布消息
    result, err := publisher.Publish(ctx,
        rmq.NewMessage([]byte("Hello, RabbitMQ AMQP 1.0!")))
    if err != nil {
        log.Panicf("发布失败: %v", err)
    }

    // 6. 验证发送结果
    switch result.Outcome.(type) {
    case *rmq.StateAccepted:
        log.Println("✓ 消息发送成功")
    default:
        log.Printf("✗ 意外结果: %v", result.Outcome)
    }
}

receive.go（消费者）：

package main

import (
    "context"
    "errors"
    "log"
    rmq "github.com/rabbitmq/rabbitmq-amqp-go-client/pkg/rabbitmqamqp"
)

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

    // 1. 创建 Environment
    env := rmq.NewEnvironment("amqp://guest:guest@localhost:5672/", nil)
    defer env.CloseConnections(context.Background())

    // 2. 建立连接
    conn, err := env.NewConnection(ctx)
    if err != nil {
        log.Panicf("连接失败: %v", err)
    }

    // 3. 声明队列（幂等操作，确保队列存在）
    _, err = conn.Management().DeclareQueue(ctx,
        &rmq.QuorumQueueSpecification{Name: "hello"})
    if err != nil {
        log.Panicf("声明队列失败: %v", err)
    }

    // 4. 创建消费者
    consumer, err := conn.NewConsumer(ctx, "hello", nil)
    if err != nil {
        log.Panicf("创建消费者失败: %v", err)
    }
    defer consumer.Close(context.Background())

    log.Println("[*] 等待消息...按 Ctrl+C 退出")

    // 5. 循环接收消息
    for {
        delivery, err := consumer.Receive(ctx)
        if err != nil {
            if errors.Is(err, context.Canceled) {
                log.Println("消费者正常退出")
                return
            }
            log.Panicf("接收消息失败: %v", err)
        }

        // 6. 处理消息
        msg := delivery.Message()
        body := string(msg.Data()[0])
        log.Printf("[x] 收到: %s", body)

        // 7. 确认消息
        err = delivery.Accept(ctx)
        if err != nil {
            log.Printf("确认消息失败: %v", err)
        }
    }
}

Work Queue（工作队列）— 任务分发

new_task.go（任务生产者）：

package main

import (
    "context"
    "fmt"
    "log"
    "os"
    "strings"
    rmq "github.com/rabbitmq/rabbitmq-amqp-go-client/pkg/rabbitmqamqp"
)

func main() {
    ctx := context.Background()
    env := rmq.NewEnvironment("amqp://guest:guest@localhost:5672/", nil)
    defer env.CloseConnections(context.Background())

    conn, _ := env.NewConnection(ctx)

    // 声明持久化队列
    _, err := conn.Management().DeclareQueue(ctx,
        &rmq.QuorumQueueSpecification{Name: "task_queue"})
    if err != nil {
        log.Panicf("声明队列失败: %v", err)
    }

    publisher, _ := conn.NewPublisher(ctx,
        &rmq.QueueAddress{Queue: "task_queue"}, nil)
    defer publisher.Close(context.Background())

    // 从命令行参数获取消息
    body := strings.Join(os.Args[1:], " ")
    if body == "" {
        body = "Hello World!"
    }

    msg := rmq.NewMessage([]byte(body))
    msg.SetContentType("text/plain")
    // 设置消息持久化
    msg.SetMessageAnnotation("x-opt-durable", true)

    result, err := publisher.Publish(ctx, msg)
    if err != nil {
        log.Panicf("发布失败: %v", err)
    }

    switch result.Outcome.(type) {
    case *rmq.StateAccepted:
        fmt.Printf("[x] 已发送: %s\n", body)
    }
}

worker.go（工作者）：

package main

import (
    "context"
    "errors"
    "fmt"
    "log"
    "strings"
    "time"
    rmq "github.com/rabbitmq/rabbitmq-amqp-go-client/pkg/rabbitmqamqp"
)

func main() {
    ctx := context.Background()
    env := rmq.NewEnvironment("amqp://guest:guest@localhost:5672/", nil)
    defer env.CloseConnections(context.Background())

    conn, _ := env.NewConnection(ctx)

    _, err := conn.Management().DeclareQueue(ctx,
        &rmq.QuorumQueueSpecification{Name: "task_queue"})
    if err != nil {
        log.Panicf("声明队列失败: %v", err)
    }

    // InitialCredits: 1 实现公平分发（一次只取一个任务）
    consumer, _ := conn.NewConsumer(ctx, "task_queue",
        &rmq.ConsumerOptions{InitialCredits: 1})
    defer consumer.Close(context.Background())

    fmt.Println("[*] 等待任务...")

    for {
        delivery, err := consumer.Receive(ctx)
        if err != nil {
            if errors.Is(err, context.Canceled) {
                return
            }
            log.Panicf("接收失败: %v", err)
        }

        msg := delivery.Message()
        body := string(msg.Data()[0])

        // 模拟耗时任务
        fmt.Printf("[x] 收到: %s\n", body)
        dots := strings.Count(body, ".")
        time.Sleep(time.Duration(dots) * time.Second)
        fmt.Printf("[x] 完成\n")

        delivery.Accept(ctx)
    }
}

发布/订阅模式（Pub/Sub）

使用 Fanout 交换机

// ─── 发布者 ───
func publishLogs(ctx context.Context, conn *rmq.Connection) error {
    // 声明 fanout 交换机
    err := conn.Management().DeclareExchange(ctx,
        &rmq.FanoutExchangeSpecification{
            Name: "logs",
        })
    if err != nil {
        return err
    }

    publisher, err := conn.NewPublisher(ctx,
        &rmq.ExchangeAddress{Exchange: "logs", Key: ""}, nil)
    if err != nil {
        return err
    }
    defer publisher.Close(context.Background())

    // 发布消息
    msg := rmq.NewMessage([]byte("系统启动完成"))
    _, err = publisher.Publish(ctx, msg)
    return err
}

// ─── 订阅者 ───
func subscribeLogs(ctx context.Context, conn *rmq.Connection) error {
    // 声明 fanout 交换机
    err := conn.Management().DeclareExchange(ctx,
        &rmq.FanoutExchangeSpecification{
            Name: "logs",
        })
    if err != nil {
        return err
    }

    // 创建临时独占队列
    queueInfo, err := conn.Management().DeclareQueue(ctx,
        &rmq.AutoGeneratedQueueSpecification{
            IsExclusive:  true,
            IsAutoDelete: true,
        })
    if err != nil {
        return err
    }

    // 绑定队列到交换机
    err = conn.Management().BindQueue(ctx, &rmq.BindingSpecification{
        SourceExchange:   "logs",
        DestinationQueue: queueInfo.Name(),
        BindingKey:       "", // fanout 忽略 routing key
    })
    if err != nil {
        return err
    }

    // 消费消息
    consumer, err := conn.NewConsumer(ctx, queueInfo.Name(), nil)
    if err != nil {
        return err
    }
    defer consumer.Close(context.Background())

    fmt.Printf("[*] 等待日志消息（队列: %s）\n", queueInfo.Name())

    for {
        delivery, err := consumer.Receive(ctx)
        if err != nil {
            return err
        }
        body := string(delivery.Message().Data()[0])
        fmt.Printf("[x] 收到日志: %s\n", body)
        delivery.Accept(ctx)
    }
}

Routing 路由模式

使用 Direct 交换机

// ─── 发布者（按严重级别发布）───
func publishWithSeverity(ctx context.Context, conn *rmq.Connection,
    severity, message string) error {

    err := conn.Management().DeclareExchange(ctx,
        &rmq.DirectExchangeSpecification{
            Name:      "direct_logs",
            IsDurable: true,
        })
    if err != nil {
        return err
    }

    publisher, err := conn.NewPublisher(ctx,
        &rmq.ExchangeAddress{
            Exchange: "direct_logs",
            Key:      severity, // "info" / "warning" / "error"
        }, nil)
    if err != nil {
        return err
    }
    defer publisher.Close(context.Background())

    msg := rmq.NewMessage([]byte(message))
    msg.SetContentType("text/plain")

    _, err = publisher.Publish(ctx, msg)
    return err
}

// ─── 消费者（按严重级别订阅）───
func consumeSeverity(ctx context.Context, conn *rmq.Connection,
    severities []string) error {

    err := conn.Management().DeclareExchange(ctx,
        &rmq.DirectExchangeSpecification{
            Name: "direct_logs",
        })
    if err != nil {
        return err
    }

    // 创建临时队列
    queueInfo, err := conn.Management().DeclareQueue(ctx,
        &rmq.AutoGeneratedQueueSpecification{
            IsExclusive:  true,
            IsAutoDelete: true,
        })
    if err != nil {
        return err
    }

    // 为每个严重级别绑定
    for _, severity := range severities {
        err = conn.Management().BindQueue(ctx, &rmq.BindingSpecification{
            SourceExchange:   "direct_logs",
            DestinationQueue: queueInfo.Name(),
            BindingKey:       severity,
        })
        if err != nil {
            return err
        }
    }

    consumer, err := conn.NewConsumer(ctx, queueInfo.Name(), nil)
    if err != nil {
        return err
    }
    defer consumer.Close(context.Background())

    fmt.Printf("[*] 等待 %v 级别的日志...\n", severities)

    for {
        delivery, err := consumer.Receive(ctx)
        if err != nil {
            return err
        }
        body := string(delivery.Message().Data()[0])
        subject := delivery.Message().Subject()
        fmt.Printf("[x] [%s] %s\n", subject, body)
        delivery.Accept(ctx)
    }
}

Topics 主题模式

使用 Topic 交换机

// ─── 发布者 ───
func publishTopic(ctx context.Context, conn *rmq.Connection,
    routingKey, message string) error {

    err := conn.Management().DeclareExchange(ctx,
        &rmq.TopicExchangeSpecification{
            Name:      "topic_logs",
            IsDurable: true,
        })
    if err != nil {
        return err
    }

    publisher, err := conn.NewPublisher(ctx,
        &rmq.ExchangeAddress{
            Exchange: "topic_logs",
            Key:      routingKey, // 如 "kern.critical", "cron.warning"
        }, nil)
    if err != nil {
        return err
    }
    defer publisher.Close(context.Background())

    msg := rmq.NewMessage([]byte(message))
    _, err = publisher.Publish(ctx, msg)
    return err
}

// ─── 订阅者（使用通配符）───
// 通配符规则：
//   * (星号) 匹配恰好一个单词
//   # (井号) 匹配零个或多个单词
//
// 示例：
//   "*.critical" → 匹配 "kern.critical", "app.critical"
//   "kern.#"     → 匹配所有以 "kern" 开头的
//   "#"          → 匹配所有消息

func subscribeTopic(ctx context.Context, conn *rmq.Connection,
    bindingPattern string) error {

    err := conn.Management().DeclareExchange(ctx,
        &rmq.TopicExchangeSpecification{
            Name: "topic_logs",
        })
    if err != nil {
        return err
    }

    queueInfo, err := conn.Management().DeclareQueue(ctx,
        &rmq.AutoGeneratedQueueSpecification{
            IsExclusive:  true,
            IsAutoDelete: true,
        })
    if err != nil {
        return err
    }

    err = conn.Management().BindQueue(ctx, &rmq.BindingSpecification{
        SourceExchange:   "topic_logs",
        DestinationQueue: queueInfo.Name(),
        BindingKey:       bindingPattern, // 如 "kern.*", "*.critical", "#"
    })
    if err != nil {
        return err
    }

    consumer, err := conn.NewConsumer(ctx, queueInfo.Name(), nil)
    if err != nil {
        return err
    }
    defer consumer.Close(context.Background())

    fmt.Printf("[*] 等待匹配 '%s' 的消息...\n", bindingPattern)

    for {
        delivery, err := consumer.Receive(ctx)
        if err != nil {
            return err
        }
        body := string(delivery.Message().Data()[0])
        fmt.Printf("[x] 收到: %s\n", body)
        delivery.Accept(ctx)
    }
}

RPC 请求/响应模式

RPC 服务端

func runRPCServer(ctx context.Context, conn *rmq.Connection) error {
    // 声明 RPC 队列
    _, err := conn.Management().DeclareQueue(ctx,
        &rmq.QuorumQueueSpecification{Name: "rpc_queue"})
    if err != nil {
        return err
    }

    // 创建消费者，每次只取一条消息
    consumer, err := conn.NewConsumer(ctx, "rpc_queue",
        &rmq.ConsumerOptions{InitialCredits: 1})
    if err != nil {
        return err
    }
    defer consumer.Close(context.Background())

    fmt.Println("RPC 服务端就绪，等待请求...")

    for {
        delivery, err := consumer.Receive(ctx)
        if err != nil {
            return err
        }

        msg := delivery.Message()
        request := string(msg.Data()[0])

        // 获取回复地址
        replyTo := msg.ReplyTo()
        correlationID := msg.CorrelationID()

        // 处理请求
        fmt.Printf("收到请求: %s (correlationID: %s)\n", request, correlationID)
        response := processRPCRequest(request)

        // 发送响应
        publisher, err := conn.NewPublisher(ctx,
            &rmq.QueueAddress{Queue: replyTo}, nil)
        if err != nil {
            delivery.Requeue(ctx)
            continue
        }

        respMsg := rmq.NewMessage([]byte(response))
        respMsg.SetCorrelationID(correlationID)

        _, err = publisher.Publish(ctx, respMsg)
        publisher.Close(context.Background())

        if err != nil {
            delivery.Requeue(ctx)
            continue
        }

        delivery.Accept(ctx)
    }
}

func processRPCRequest(req string) string {
    // 实际业务逻辑
    return "处理结果: " + req
}

RPC 客户端

func callRPC(ctx context.Context, conn *rmq.Connection,
    request string) (string, error) {

    // 创建临时回复队列
    replyQueueInfo, err := conn.Management().DeclareQueue(ctx,
        &rmq.AutoGeneratedQueueSpecification{
            IsExclusive:  true,
            IsAutoDelete: true,
        })
    if err != nil {
        return "", err
    }
    replyQueueName := replyQueueInfo.Name()

    correlationID := fmt.Sprintf("rpc-%d", time.Now().UnixNano())

    // 发布请求
    publisher, err := conn.NewPublisher(ctx,
        &rmq.QueueAddress{Queue: "rpc_queue"}, nil)
    if err != nil {
        return "", err
    }
    defer publisher.Close(context.Background())

    msg := rmq.NewMessage([]byte(request))
    msg.SetReplyTo(replyQueueName)
    msg.SetCorrelationID(correlationID)

    _, err = publisher.Publish(ctx, msg)
    if err != nil {
        return "", err
    }

    // 等待响应
    consumer, err := conn.NewConsumer(ctx, replyQueueName,
        &rmq.ConsumerOptions{InitialCredits: 1})
    if err != nil {
        return "", err
    }
    defer consumer.Close(context.Background())

    // 带超时的等待
    timeoutCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
    defer cancel()

    for {
        delivery, err := consumer.Receive(timeoutCtx)
        if err != nil {
            return "", fmt.Errorf("等待响应超时: %w", err)
        }

        respMsg := delivery.Message()
        if respMsg.CorrelationID() == correlationID {
            body := string(respMsg.Data()[0])
            delivery.Accept(ctx)
            return body, nil
        }
        // 不是我们的响应，继续等待
        delivery.Accept(ctx)
    }
}

消费与消息确认 Stream 流队列