feat: add reconnect for mq producer #27
Open
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Walkthrough本次变更为 RabbitMQ Producer 增加了自动重连和重试机制。通过引入连接状态检测、新的重连函数以及并发去重机制，提升了在连接关闭时的健壮性，并在消息发布失败时自动尝试重连和重发。 Changes
Sequence Diagram(s)sequenceDiagram
participant Producer
participant RabbitMQ
Producer->>RabbitMQ: Publish 消息
alt 连接正常
RabbitMQ-->>Producer: 返回成功/失败
else 连接关闭
Producer->>Producer: 调用 connectFn 重连
alt 重连成功
Producer->>RabbitMQ: 再次 Publish
RabbitMQ-->>Producer: 返回结果
else 重连失败
Producer-->>Producer: 返回错误
end
end
Poem
✨ Finishing Touches
🧪 Generate Unit Tests
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. 🪧 TipsChatThere are 3 ways to chat with CodeRabbit:
SupportNeed help? Create a ticket on our support page for assistance with any issues or questions. Note: Be mindful of the bot's finite context window. It's strongly recommended to break down tasks such as reading entire modules into smaller chunks. For a focused discussion, use review comments to chat about specific files and their changes, instead of using the PR comments. CodeRabbit Commands (Invoked using PR comments)
Other keywords and placeholders
CodeRabbit Configuration File (
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
🧹 Nitpick comments (1)
mq/producer.go (1)
193-202: 建议改进 Shutdown 方法考虑到新增的重连机制，Shutdown 方法可以进行以下改进：
func (p *Producer) Shutdown() error { + // 等待所有重连操作完成 + p.sf.Forget("reconnect") + + // 先关闭 channel + if p.Channel != nil && !p.Channel.IsClosed() { + if err := p.Channel.Close(); err != nil { + return fmt.Errorf("AMQP channel close error: %v", err) + } + } if err := p.Conn.Close(); err != nil { - return fmt.Errorf("AMQP connection close error: %s", err) + return fmt.Errorf("AMQP connection close error: %v", err) } - defer fmt.Printf("AMQP shutdown OK") + // 考虑使用正式的日志记录器而不是 fmt.Printf return nil }
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (1)
mq/producer.go(4 hunks)
🧰 Additional context used
🧬 Code Graph Analysis (1)
mq/producer.go (1)
mq/init.go (1)
UserIdKey(14-14)
🔇 Additional comments (4)
mq/producer.go (4)
6-13: 导入包合理且必要！新增的导入包都是实现重连功能所必需的，特别是使用
singleflight来防止并发重连是一个很好的设计选择。
20-22: 结构体字段设计合理！新增的字段用于支持重连功能，命名清晰，注释明确。
32-34: 构造函数更新正确！新字段的初始化正确，保持了与原有代码的一致性。
47-49: 连接状态检查方法实现正确！
isConnected()方法简洁有效地检查了连接和通道的状态。
Comment on lines
+167
to
+187
| if err != nil && errors.Is(err, amqp.ErrClosed) { | ||
| if err = p.connectFn(); err != nil { | ||
| return err | ||
| } | ||
|
|
||
| err = p.Channel.PublishWithContext( | ||
| ctx, | ||
| exchangeName, | ||
| key, | ||
| false, | ||
| false, | ||
| amqp.Publishing{ | ||
| ContentType: "application/json", | ||
| DeliveryMode: amqp.Persistent, | ||
| Body: msg, | ||
| AppId: p.appId, | ||
| UserId: opts[UserIdKey], | ||
| MessageId: fmt.Sprintf("%x", md5.Sum(msg)), | ||
| Headers: headers, | ||
| }) | ||
| } |
There was a problem hiding this comment.
🛠️ Refactor suggestion
重构建议：消除代码重复并改进重试逻辑
当前实现存在代码重复问题，且重试逻辑可以更加健壮。
建议将发布逻辑提取到一个内部方法中，并实现更完善的重试机制：
- err := p.Channel.PublishWithContext(
- ctx,
- exchangeName,
- key,
- false,
- false,
- amqp.Publishing{
- ContentType: "application/json",
- DeliveryMode: amqp.Persistent,
- Body: msg,
- AppId: p.appId,
- UserId: opts[UserIdKey],
- MessageId: fmt.Sprintf("%x", md5.Sum(msg)),
- Headers: headers,
- })
-
- if err != nil && errors.Is(err, amqp.ErrClosed) {
- if err = p.connectFn(); err != nil {
- return err
- }
-
- err = p.Channel.PublishWithContext(
- ctx,
- exchangeName,
- key,
- false,
- false,
- amqp.Publishing{
- ContentType: "application/json",
- DeliveryMode: amqp.Persistent,
- Body: msg,
- AppId: p.appId,
- UserId: opts[UserIdKey],
- MessageId: fmt.Sprintf("%x", md5.Sum(msg)),
- Headers: headers,
- })
- }
-
- return err
+ publishing := amqp.Publishing{
+ ContentType: "application/json",
+ DeliveryMode: amqp.Persistent,
+ Body: msg,
+ AppId: p.appId,
+ UserId: opts[UserIdKey],
+ MessageId: fmt.Sprintf("%x", md5.Sum(msg)),
+ Headers: headers,
+ }
+
+ // 最多重试 2 次
+ for i := 0; i < 2; i++ {
+ err := p.Channel.PublishWithContext(
+ ctx,
+ exchangeName,
+ key,
+ false,
+ false,
+ publishing,
+ )
+
+ if err == nil {
+ return nil
+ }
+
+ if !errors.Is(err, amqp.ErrClosed) {
+ return err
+ }
+
+ // 尝试重连
+ if reconnectErr := p.connectFn(); reconnectErr != nil {
+ return fmt.Errorf("publish failed: %v, reconnect failed: %v", err, reconnectErr)
+ }
+ }
+
+ return fmt.Errorf("publish failed after %d attempts", 2)这样可以：
- 消除代码重复
- 支持多次重试
- 提供更清晰的错误信息
Committable suggestion skipped: line range outside the PR's diff.
🤖 Prompt for AI Agents
In mq/producer.go around lines 167 to 187, the code duplicates the message
publishing logic and has a simplistic retry mechanism. Refactor by extracting
the publishing code into a separate internal function and implement a loop or
retry mechanism that attempts to publish multiple times upon failure, especially
when the connection is closed. This will remove duplication, allow multiple
retries, and improve error handling clarity.
Comment on lines
+51
to
+92
| func (p *Producer) connectFn() error { | ||
| if p.isConnected() { | ||
| return nil | ||
| } | ||
|
|
||
| _, err, _ := p.sf.Do("reconnect", func() (interface{}, error) { | ||
| var lastErr error | ||
| for i := 0; i < 3; i++ { | ||
| if p.isConnected() { | ||
| return nil, nil | ||
| } | ||
|
|
||
| if i > 0 { | ||
| time.Sleep(time.Second * time.Duration(math.Pow(2, float64(i-1)))) | ||
| } | ||
|
|
||
| conn, channel, err := initConnection(p.amqpURI) | ||
| if err != nil { | ||
| lastErr = fmt.Errorf("reconnect attempt %d failed: %s", i+1, err) | ||
| continue | ||
| } | ||
|
|
||
| oldConn := p.Conn | ||
| oldChannel := p.Channel | ||
| p.Conn = conn | ||
| p.Channel = channel | ||
|
|
||
| if oldChannel != nil { | ||
| _ = oldChannel.Close() | ||
| } | ||
| if oldConn != nil { | ||
| _ = oldConn.Close() | ||
| } | ||
|
|
||
| return nil, nil | ||
| } | ||
|
|
||
| return nil, lastErr | ||
| }) | ||
|
|
||
| return err | ||
| } |
There was a problem hiding this comment.
存在线程安全问题，需要加锁保护连接替换操作
当前的连接替换逻辑（第73-83行）存在竞态条件。在读取旧连接和关闭它们之间，其他 goroutine 可能正在使用这些连接，导致潜在的并发问题。
建议使用互斥锁保护连接替换操作：
type Producer struct {
Conn *amqp.Connection
Channel *amqp.Channel
+ mu sync.RWMutex // 保护连接的读写
amqpURI string
sf *singleflight.Group
appId string
}然后在 connectFn 中：
- oldConn := p.Conn
- oldChannel := p.Channel
- p.Conn = conn
- p.Channel = channel
+ p.mu.Lock()
+ oldConn := p.Conn
+ oldChannel := p.Channel
+ p.Conn = conn
+ p.Channel = channel
+ p.mu.Unlock()同时，需要在 isConnected() 和 publish() 方法中使用读锁来访问连接。
另外，建议将错误格式化从 %s 改为 %v 以更好地处理各种错误类型：
- lastErr = fmt.Errorf("reconnect attempt %d failed: %s", i+1, err)
+ lastErr = fmt.Errorf("reconnect attempt %d failed: %v", i+1, err)🤖 Prompt for AI Agents
In mq/producer.go lines 51 to 92, the connection replacement code from lines 73
to 83 is not thread-safe and can cause race conditions when multiple goroutines
access or modify the connection concurrently. To fix this, introduce a mutex
lock to protect the connection and channel replacement operations inside the
reconnect logic. Also, modify the isConnected() and publish() methods to use a
read lock when accessing the connection to ensure safe concurrent reads.
Additionally, update the error formatting in the reconnect attempts from using
%s to %v to properly handle different error types.
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Summary by CodeRabbit