198 lines
6.8 KiB
Markdown
198 lines
6.8 KiB
Markdown
---
|
||
tags:
|
||
- MQ
|
||
create time: 2026-05-24 19:52
|
||
---
|
||
|
||
# MQ 客户端连接管理
|
||
|
||
## 概述
|
||
|
||
连接是 MQ 客户端与 Broker 之间的通信通道，连接管理的好坏直接影响消息收发的稳定性和性能。本文深入探讨长连接与心跳机制、重连策略、Consumer Rebalance、客户端容错以及多数据中心路由等话题。
|
||
|
||
## 正文
|
||
|
||
### 长连接 vs 短连接
|
||
|
||
MQ 客户端几乎都使用**长连接**，原因有三：
|
||
|
||
1. **建立连接成本高**：TCP 三次握手 + MQ 协议握手 + 认证鉴权，一次连接建立可能耗时几十到几百毫秒。如果每发一条消息都建一次连接，性能无法接受。
|
||
2. **连接状态维护**：消费者需要维持 Offset、订阅关系等状态，这些状态绑定在连接上。
|
||
3. **推送模式依赖长连接**：Broker 需要通过已有连接向 Consumer 推送消息（或通知拉取），短连接无法实现。
|
||
|
||
短连接的唯一优势是不需要保活，适合极低频的调用场景。但 MQ 显然不是这种场景。
|
||
|
||
### 心跳机制
|
||
|
||
长连接需要心跳来维持。心跳有两个作用：
|
||
|
||
- **保活检测**：防止中间设备（NAT、防火墙、LB）因连接空闲超时而断开。
|
||
- **故障发现**：Broker 通过心跳超时检测客户端是否存活，及时清理资源。
|
||
|
||
```go
|
||
// 心跳管理器
|
||
type HeartbeatManager struct {
|
||
interval time.Duration
|
||
timeout time.Duration
|
||
conn net.Conn
|
||
lastAck time.Time
|
||
mu sync.RWMutex
|
||
}
|
||
|
||
func (h *HeartbeatManager) Start(ctx context.Context) {
|
||
ticker := time.NewTicker(h.interval)
|
||
defer ticker.Stop()
|
||
|
||
for {
|
||
select {
|
||
case <-ctx.Done():
|
||
return
|
||
case <-ticker.C:
|
||
// 发送心跳
|
||
if err := h.sendPing(); err != nil {
|
||
log.Warn("heartbeat failed", zap.Error(err))
|
||
h.onHeartbeatFailed()
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
func (h *HeartbeatManager) sendPing() error {
|
||
h.conn.SetWriteDeadline(time.Now().Add(h.timeout))
|
||
_, err := h.conn.Write([]byte{0x01}) // 心跳包
|
||
if err == nil {
|
||
h.mu.Lock()
|
||
h.lastAck = time.Now()
|
||
h.mu.Unlock()
|
||
}
|
||
return err
|
||
}
|
||
```
|
||
|
||
典型的配置：
|
||
- 心跳间隔：3-30 秒
|
||
- Broker 超时：通常为心跳间隔的 3 倍（比如心跳 10 秒，超时 30 秒）
|
||
- Kafka 默认心跳间隔 3 秒，session timeout 45 秒
|
||
|
||
### 重连策略
|
||
|
||
网络不可避免会抖动，客户端必须有健壮的重连机制。核心是**指数退避 + 抖动**：
|
||
|
||
```go
|
||
type ReconnectPolicy struct {
|
||
BaseDelay time.Duration // 初始延迟，如 1s
|
||
MaxDelay time.Duration // 最大延迟，如 60s
|
||
MaxRetries int // 最大重试次数，0 表示无限
|
||
Multiplier float64 // 退避倍数，通常 2.0
|
||
JitterFactor float64 // 抖动因子，通常 0.1-0.3
|
||
}
|
||
|
||
func (p *ReconnectPolicy) NextDelay(attempt int) time.Duration {
|
||
// 指数退避：delay = base * multiplier^attempt
|
||
delay := float64(p.BaseDelay) * math.Pow(p.Multiplier, float64(attempt))
|
||
if delay > float64(p.MaxDelay) {
|
||
delay = float64(p.MaxDelay)
|
||
}
|
||
// 加入抖动：避免多客户端同时重连造成"惊群效应"
|
||
jitter := delay * p.JitterFactor * (rand.Float64()*2 - 1)
|
||
return time.Duration(delay + jitter)
|
||
}
|
||
```
|
||
|
||
**为什么需要抖动？** 如果 1000 个客户端同时断线，没有抖动的话会在同一时刻全部重连，形成"惊群效应"，直接把 Broker 打崩。抖动让重连时间分散开，平滑了压力。
|
||
|
||
### Consumer Rebalance 深入
|
||
|
||
Rebalance 是 Consumer Group 中最重要的机制之一。当 Group 成员变化时（新加入、离开、心跳超时），需要重新分配 Partition。
|
||
|
||
**触发条件：**
|
||
1. 新 Consumer 加入 Group
|
||
2. Consumer 主动离开（关闭）
|
||
3. Consumer 心跳超时（被认为宕机）
|
||
4. Topic 的 Partition 数量变化
|
||
|
||
**分配策略：**
|
||
- **Range**：按 Topic 分区连续分配，简单但容易不均。
|
||
- **RoundRobin**：轮询分配，更均匀但实现复杂。
|
||
- **Sticky**：尽量保持原有分配不变，只调整必要的部分，减少 Rebalance 开销。
|
||
- **CooperativeSticky**（Kafka 2.4+）：增量式 Rebalance，不需要 Stop-The-World。
|
||
|
||
**静态成员（Static Membership）**：Kafka 2.3 引入。给每个 Consumer 分配一个固定的 `group.instance.id`，短暂离线后重新加入时保持原来的 Partition 分配，避免不必要的 Rebalance。
|
||
|
||
```go
|
||
// 静态成员配置
|
||
config := sarama.NewConfig()
|
||
config.Consumer.Group.InstanceId = "consumer-pod-1" // 固定 ID
|
||
config.Consumer.Group.Rebalance.Strategy = sarama.BalanceStrategySticky
|
||
config.Consumer.Group.Rebalance.Timeout = 60 * time.Second
|
||
```
|
||
|
||
> [!question]
|
||
> Consumer Rebalance 期间消费会暂停，如何实现"零停顿"的 Rebalance？提示：想想 CooperativeSticky 策略的工作方式。
|
||
|
||
### 客户端容错
|
||
|
||
客户端需要感知 Broker 层的变化：
|
||
|
||
- **Broker 故障转移**：Producer 连接的 Broker 宕机时，需要自动切换到其他 Broker。
|
||
- **分区 Leader 迁移**：某个 Partition 的 Leader 变了，客户端需要感知并更新路由。
|
||
- **元数据刷新**：定期从 Broker 拉取最新的元数据（Topic 列表、分区信息、Leader 位置）。
|
||
|
||
```go
|
||
// 元数据自动刷新
|
||
type MetadataRefresher struct {
|
||
client KafkaClient
|
||
interval time.Duration
|
||
topics []string
|
||
}
|
||
|
||
func (m *MetadataRefresher) Start(ctx context.Context) {
|
||
ticker := time.NewTicker(m.interval)
|
||
defer ticker.Stop()
|
||
for {
|
||
select {
|
||
case <-ctx.Done():
|
||
return
|
||
case <-ticker.C:
|
||
metadata, err := m.client.RefreshMetadata(m.topics)
|
||
if err != nil {
|
||
log.Error("metadata refresh failed", zap.Error(err))
|
||
continue
|
||
}
|
||
m.updateRoutingTable(metadata)
|
||
}
|
||
}
|
||
}
|
||
```
|
||
|
||
### 多数据中心客户端路由
|
||
|
||
全球化部署时，客户端需要智能路由：
|
||
|
||
- **就近连接**：优先连接同区域的 Broker，降低延迟。
|
||
- **故障转移**：本地 Broker 不可用时，自动切换到远端 Broker。
|
||
- **读写分离**：本地集群负责读写，远端集群只作为灾备。
|
||
|
||
```mermaid
|
||
stateDiagram-v2
|
||
[*] --> Idle
|
||
Idle --> Connecting: "发起连接"
|
||
Connecting --> Connected: "握手成功"
|
||
Connecting --> Reconnecting: "连接失败"
|
||
Connected --> Heartbeating: "启动心跳"
|
||
Heartbeating --> Connected: "心跳正常"
|
||
Heartbeating --> Reconnecting: "心跳超时"
|
||
Connected --> Reconnecting: "连接断开"
|
||
Reconnecting --> Connecting: "退避结束"
|
||
Reconnecting --> Closed: "超过最大重试"
|
||
Connected --> Closing: "优雅关闭"
|
||
Closing --> Closed: "资源释放完成"
|
||
Closed --> [*]
|
||
```
|
||
|
||
## 关联笔记
|
||
|
||
- [[44-MQ-高可用架构]]
|
||
- [[48-MQ-客户端-SDK-最佳实践]]
|
||
- [[50-MQ-设计与实现]]
|