Golang使用RabbitMQ消息中間件amqp協議

"github.com/streadway/amqp"

Publish發布

// amqp://<user>:<password>@<ip>:<port>
var addr = "amqp://test:test123@127.0.0.1:5672" //test

func main() {
      
    // 建立連接
    conn, err := amqp.Dial(addr)
    if nil != err {
	logs.Error(err)
	return
    }

    defer conn.Close()
    // 申請通道
    ch, err := conn.Channel()
    if nil != err {
	    logs.Error(err)
	    return
    }

    defer ch.Close()
    // 定義交換“direct”、“fanout”、“topic”和“headers”
    err = ch.ExchangeDeclare("happy", amqp.ExchangeTopic, true, false, false, false, nil)
    if nil != err {
	    logs.Error(err)
	    return
    }

    data = fmt.Sprintf("hello,world!!!")
        //a.b.c.d.e 為發布key,以.分割；
    err = ch.Publish("happy", "a.b.c.d.e", false, false,
	    amqp.Publishing{
		    ContentType:  "text/plain",
		    Body:         []byte(data),
		    DeliveryMode: amqp.Transient, // 1=non-persistent, 2=persistent
	    })
    if nil != err {
	    logs.Error(err)
	    return
    }

}

Recover 接收

// amqp://<user>:<password>@<ip>:<port>
var addr = "amqp://test:test123@127.0.0.1:5672" //test

func main() {

    // 建立連接
    conn, err := amqp.Dial(addr)
    if nil != err {
            logs.Error(err)
	    return
    }

    defer conn.Close()
    // 申請通道
    ch, err := conn.Channel()
    if nil != err {
	    logs.Error(err)
	    return
    }

    defer ch.Close()
    // 定義交換
    err = ch.ExchangeDeclare("happy", amqp.ExchangeTopic, true, false, false, false, nil)
    if nil != err {
	    logs.Error(err)
	    return
    }

    queName := "test.test1.test2"
    topic := "a.#"


    // 定義通道
    que, err := ch.QueueDeclare(queName, false, false, false, false, nil)
    if nil != err {
    	    logs.Error(err)
    }

    err = ch.QueueBind(que.Name, topic, "happy", false, nil)
    if nil != err {
	    logs.Error(err)
	    return
    }

    msges, err := ch.Consume(que.Name, "", true, false, false, false, nil)
    if nil != err {
	    logs.Error(err)
	    return
    }

    logs.Info("start recv")

    for msg := range msges {
	    fmt.Println(">>> %s", string(msg.Body))
	
    }

}

http管理端口是15672

注：隊列應先注冊一次，才能收到消息