doucongqian6644 2016-03-23 13:09
浏览 352
已采纳

在go lang中是否在amqp.Dial具有线程安全性时每次创建连接

As it is mentioned in the RabbitMQ docs that tcp connections are expensive to make. So, for that concept of channel was introduced. Now i came across this example. In the main() it creates the connection everytime a message is publised. conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/"). Shouldn't it be declared globally once and there should be failover mechanism in case connection get closed like singleton object. If amqp.Dial is thread-safe, which i suppose it should be

Edited question :

I am handling the connection error in the following manner. In which i listen on a channel and create a new connection on error. But when i kill the existing connection and try to publish message. I get the following error.

error :

2016/03/30 19:20:08 Failed to open a channel: write tcp 172.16.5.48:51085->172.16.0.20:5672: use of closed network connection
exit status 1
7:25 PM

Code :

 func main() {

        Conn, err := amqp.Dial("amqp://guest:guest@172.16.0.20:5672/")
        failOnError(err, "Failed to connect to RabbitMQ")
         context := &appContext{queueName: "QUEUENAME",exchangeName: "ExchangeName",exchangeType: "direct",routingKey: "RoutingKey",conn: Conn}
        c := make(chan *amqp.Error)

        go func() {
            error := <-c
            if(error != nil){                
                Conn, err = amqp.Dial("amqp://guest:guest@172.16.0.20:5672/")            
                failOnError(err, "Failed to connect to RabbitMQ")            
                Conn.NotifyClose(c)                                           
            }            
        }()

        Conn.NotifyClose(c)
        r := web.New()
        // We pass an instance to our context pointer, and our handler.
        r.Get("/", appHandler{context, IndexHandler})
        graceful.ListenAndServe(":8086", r)  

    }
  • 写回答

1条回答 默认 最新

  • duanguzhong5776 2016-03-23 13:28
    关注

    Of course, you shouldn't create a connection for each request. Make it a global variable or better part of an application context which you initialize once at startup.

    You can handle connection errors by registering a channel using Connection.NotifyClose:

    func initialize() {
      c := make(chan *amqp.Error)
      go func() {
        err := <-c
        log.Println("reconnect: " + err.Error())
        initialize()
      }()
    
      conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
      if err != nil {
        panic("cannot connect")
      }
      conn.NotifyClose(c)
    
      // create topology
    }
    
    本回答被题主选为最佳回答 , 对您是否有帮助呢?
    评论

报告相同问题?

悬赏问题

  • ¥100 X轴为分离变量(因子变量),如何控制X轴每个分类变量的长度。
  • ¥30 求给定范围的全体素数p的(p-2)的连乘积
  • ¥15 VFP如何使用阿里TTS实现文字转语音?
  • ¥100 需要跳转番茄畅听app的adb命令
  • ¥50 寻找一位有逆向游戏盾sdk 应用程序经验的技术
  • ¥15 请问有用MZmine处理 “Waters SYNAPT G2-Si QTOF质谱仪在MSE模式下采集的非靶向数据” 的分析教程吗
  • ¥50 opencv4nodejs 如何安装
  • ¥15 adb push异常 adb: error: 1409-byte write failed: Invalid argument
  • ¥15 nginx反向代理获取ip,java获取真实ip
  • ¥15 eda:门禁系统设计