在 Go 中检测和处理失效的 RabbitMQ 连接
在提供的 RabbitMQ 消费者脚本中,RabbitMQ 服务器操作的中断导致脚本运行但对消息没有反应。为了解决这个问题,必须检测死连接并重新连接或终止脚本。
底层库 streadway/amqp 提供了一个解决方案。 amqp.Connection 类型有一个 NotifyClose() 方法,该方法返回通道信令传输或协议错误。通过使用此通道,可以检测连接失败。
这是包含连接错误处理的脚本的更新版本:
import ( "log" "github.com/streadway/amqp" ) func main() { // Connection loop for { conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/") if err != nil { log.Printf("Failed to connect to RabbitMQ: %v", err) continue } notify := conn.NotifyClose(make(chan *amqp.Error)) ch, err := conn.Channel() if err != nil { log.Printf("Failed to open a channel: %v", err) continue } q, err := ch.QueueDeclare( "test_task_queue", true, false, false, false, nil, ) if err != nil { log.Printf("Failed to declare a queue: %v", err) continue } err = ch.Qos( 1, 0, false, ) if err != nil { log.Printf("Failed to set QoS: %v", err) continue } msgs, err := ch.Consume( q.Name, "", false, false, false, false, nil, ) if err != nil { log.Printf("Failed to register a consumer: %v", err) continue } // Message receive loop for { select { case err := <-notify: // Connection error handling log.Printf("Connection error: %v", err) break // Reconnect case d := <-msgs: // Message handling log.Printf("Received a message: %s", d.Body) d.Ack(false) dot_count := bytes.Count(d.Body, []byte(".")) t := time.Duration(dot_count) time.Sleep(t * time.Second) log.Printf("Done") } } } }
在此修订后的脚本中,连接循环包装了其他操作。在消息接收循环中,select 语句检查错误通道和传入消息。如果检测到连接错误,它会记录该错误并触发重新连接。这种机制确保脚本可以从瞬态连接问题中恢复并恢复正常操作。
以上是如何在 Go 中检测和处理死掉的 RabbitMQ 连接?的详细内容。更多信息请关注PHP中文网其他相关文章!