Skip to content
连接管理

连接管理

Environment / Connection / Management 三层对象模型、连接 URI 完整格式、TLS/SSL 安全连接(含 mTLS),以及 EnvironmentOptions / ConsumerOptions / URI 查询参数等高级配置。


环境初始化与连接管理

核心概念

概念 说明
Environment 顶层管理对象,持有共享的连接配置(SASL、TLS、超时等)。一个应用通常只需要一个 Environment 实例。
Connection 到 RabbitMQ Broker 的 TCP 连接。从 Environment 创建,使用完毕后需要关闭。
Management 通过 Connection 获取的管理接口,用于声明/删除队列和交换机。

创建 Environment

// 基础方式:使用 URI
env := rmq.NewEnvironment("amqp://guest:guest@localhost:5672/", nil)

// URI 格式说明:
// amqp://用户名:密码@主机:端口/虚拟主机
// 虚拟主机默认为 "/"

// 程序退出前关闭所有连接
defer func() {
    _ = env.CloseConnections(context.Background())
}()

建立连接

ctx := context.Background()

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

// 连接使用完毕后关闭
defer func() {
    _ = conn.Close(context.Background())
}()

连接 URI 的完整格式

amqp://[用户名]:[密码]@[主机]:[端口]/[虚拟主机]?[参数]
amqps://[用户名]:[密码]@[主机]:[端口]/[虚拟主机]?[参数]   # TLS 加密

示例:

URI 说明
amqp://guest:guest@localhost:5672/ 本地开发环境
amqp://admin:pass@rabbitmq.example.com:5672/vhost1 指定虚拟主机
amqps://user:pass@secure-rabbit:5671/ TLS 安全连接

TLS/SSL 安全连接

基础 TLS 连接

// 使用 amqps:// 协议前缀
env := rmq.NewEnvironment("amqps://guest:guest@localhost:5671/", nil)
conn, err := env.NewConnection(context.Background())

自定义 TLS 配置

import (
    "crypto/tls"
    "crypto/x509"
    "os"
)

// ─── 场景一:自签名证书 / 自定义 CA ───
caCert, err := os.ReadFile("./ca_certificate.pem")
if err != nil {
    log.Fatal(err)
}

caCertPool := x509.NewCertPool()
caCertPool.AppendCertsFromPEM(caCert)

tlsConfig := &tls.Config{
    RootCAs:    caCertPool,
    MinVersion: tls.VersionTLS12,
}

env := rmq.NewEnvironment("amqps://guest:guest@rabbitmq.example.com:5671/",
    &rmq.EnvironmentOptions{
        TLSConfig: tlsConfig,
    })

// ─── 场景二:双向 TLS(mTLS / 客户端证书认证)───
clientCert, err := tls.LoadX509KeyPair("client-cert.pem", "client-key.pem")
if err != nil {
    log.Fatal(err)
}

caCertPool := x509.NewCertPool()
caCert, _ := os.ReadFile("ca_certificate.pem")
caCertPool.AppendCertsFromPEM(caCert)

tlsConfig := &tls.Config{
    Certificates: []tls.Certificate{clientCert},
    RootCAs:      caCertPool,
    MinVersion:   tls.VersionTLS12,
}

env := rmq.NewEnvironment("amqps://rabbitmq.example.com:5671/",
    &rmq.EnvironmentOptions{
        TLSConfig: tlsConfig,
    })

// ─── 场景三:跳过证书验证(仅限开发/测试环境)───
tlsConfig := &tls.Config{
    InsecureSkipVerify: true, // ⚠️ 生产环境绝对不要使用
}

env := rmq.NewEnvironment("amqps://localhost:5671/",
    &rmq.EnvironmentOptions{
        TLSConfig: tlsConfig,
    })

EnvironmentOptions 完整 TLS 配置

type EnvironmentOptions struct {
    TLSConfig *tls.Config  // TLS 配置
    // SASLType  string       // SASL 认证类型 (PLAIN, EXTERNAL 等)
    // 其他配置...
}

高级配置选项

EnvironmentOptions

env := rmq.NewEnvironment("amqp://guest:guest@localhost:5672/",
    &rmq.EnvironmentOptions{
        // TLS 配置
        TLSConfig: &tls.Config{
            MinVersion: tls.VersionTLS12,
        },

        // 连接超时相关(具体选项取决于库版本)
        // ConnectionTimeout: 30 * time.Second,

        // 自定义 SASL 认证
        // 默认使用 PLAIN SASL
    })

ConsumerOptions

consumer, err := conn.NewConsumer(ctx, "queue-name", &rmq.ConsumerOptions{
    // 初始信用额度(流量控制)
    InitialCredits: 10,

    // 消费者标签(用于标识)
    // LinkName: "my-consumer-1",

    // 是否独占消费者(取决于库版本支持)
    // Exclusive: true,

    // Stream 相关选项
    StreamOptions: &rmq.StreamOptions{
        Offset:       "first",           // "first", "last", "next", 数字
        FilterValues: []string{"tag1"},  // 消息过滤值
    },
})

连接参数(URI 查询参数)

amqp://guest:guest@localhost:5672/?heartbeat=30&connection_timeout=10000
参数 说明 默认值
heartbeat 心跳间隔(秒) 60
connection_timeout 连接超时(毫秒) 30000
idle_time_out 空闲超时(毫秒)