一：消息重試機制

如是有了如下思路：

消息進入隊列前，header默認有參數(shù) retry_num=0 表示嘗試次數(shù)；

消費者在消費時候的，如果消息失敗，就把消息插入另外一個隊列（隊列abc）；該隊列abc 綁定一個死信隊列（原始消費的隊列），這樣形成一個回路；

當消息失敗后，消息就進入隊列abc，隊列abc擁有ttl過期時間，ttl過期時間到了后，該消息進入死信隊列（死信隊列剛好是剛開始我們消費的隊列）；

這樣消息就又回到原始消費隊列尾部了；

最后可以通過隊列消息頭部的header參數(shù)retry_num 可以控制消息消費多少次后，直接插入db日志；

db日志可以記錄交換機路由，queuename，這樣，可以做一個后臺管理，可以手動一次把消息重新放入隊列，進行消息（因為有時間消費隊列里面可能在請求其它服務(wù)，其它服務(wù)也可能會掛掉）

這時候消息無論你消費多少次都沒有用，但是入庫db后，可以一鍵重回隊列消息（當我們知道服務(wù)已經(jīng)正常后）

圖解：

附上代碼

git clone https://github.com/sunlongv520/golang-rabbitmq

send.go 消費者

package main

import (
    "fmt"
    _ "fmt"
    "https://github.com/sunlongv520/golang-rabbitmq/utils/rabbitmq"
)

func main() {


    for i := 0;i<20;i++{
        body := fmt.Sprintf("{\"order_id\":%d}",i)
        fmt.Println(body)

        /**
            使用默認的交換機
            如果是默認交換機
            type QueueExchange struct {
            QuName  string           // 隊列名稱
            RtKey   string           // key值
            ExName  string           // 交換機名稱
            ExType  string           // 交換機類型
            Dns     string              //鏈接地址
            }
            如果你喜歡使用默認交換機
            RtKey  此處建議填寫成 RtKey 和 QuName 一樣的值
         */

        queueExchange := rabbitmq.QueueExchange{
            "a_test_0001",
            "a_test_0001",
            "hello_go",
            "direct",
            "amqp://guest:guest@192.168.1.169:5672/",
        }

        _ = rabbitmq.Send(queueExchange,body)

    }
}

View Code

recv.go消費者

package main

import (
    "fmt"
    "https://github.com/sunlongv520/golang-rabbitmq/utils/rabbitmq"
    "time"
)

type RecvPro struct {

}

//// 實現(xiàn)消費者 消費消息失敗 自動進入延時嘗試  嘗試3次之后入庫db
/*
返回值 error 為nil  則表示該消息消費成功
否則消息會進入ttl延時隊列  重復(fù)嘗試消費3次
3次后消息如果還是失敗 消息就執(zhí)行失敗  進入告警 FailAction
 */
func (t *RecvPro) Consumer(dataByte []byte) error {
    time.Sleep(time.Second*1)
    //return errors.New("頂頂頂頂")
    fmt.Println(string(dataByte))
    //time.Sleep(1*time.Second)
    //return errors.New("頂頂頂頂")
    return nil
}

//消息已經(jīng)消費3次 失敗了 請進行處理
/*
如果消息 消費3次后 仍然失敗  此處可以根據(jù)情況 對消息進行告警提醒 或者 補償  入庫db  釘釘告警等等
 */
func (t *RecvPro) FailAction(err error,dataByte []byte) error {
    fmt.Println(string(dataByte))
    fmt.Println(err)
    fmt.Println("任務(wù)處理失敗了，我要進入db日志庫了")
    fmt.Println("任務(wù)處理失敗了，發(fā)送釘釘消息通知主人")
    return nil
}



func main() {
    processTask := &RecvPro{}

    /*
        runNums: 表示任務(wù)并發(fā)處理數(shù)量  一般建議 普通任務(wù)1-3    就可以了
        maxTryConnTimeFromMinute:表示最大嘗試時間  分鐘
     */
    err := rabbitmq.Recv(rabbitmq.QueueExchange{
        "a_test_0001",
        "a_test_0001",
        "hello_go",
        "direct",
        "amqp://guest:guest@192.168.1.169:5672/",
    },
    processTask,4,2)
    if(err != nil){
        fmt.Println(err)
    }

}

View Code

utils/rabbitmq包

package rabbitmq

import (
    "errors"
    "strconv"
    "time"

    //"errors"
    "fmt"
    "github.com/streadway/amqp"
    "log"
)


// 定義全局變量,指針類型
var mqConn *amqp.Connection
var mqChan *amqp.Channel

// 定義生產(chǎn)者接口
type Producer interface {
    MsgContent() string
}

// 定義生產(chǎn)者接口
type RetryProducer interface {
    MsgContent() string
}

// 定義接收者接口
type Receiver interface {
    Consumer([]byte)    error
    FailAction(error , []byte)  error
}

// 定義RabbitMQ對象
type RabbitMQ struct {
    connection *amqp.Connection
    Channel *amqp.Channel
    dns string
    QueueName   string            // 隊列名稱
    RoutingKey  string            // key名稱
    ExchangeName string           // 交換機名稱
    ExchangeType string           // 交換機類型
    producerList []Producer
    retryProducerList []RetryProducer
    receiverList []Receiver
}

// 定義隊列交換機對象
type QueueExchange struct {
    QuName  string           // 隊列名稱
    RtKey   string           // key值
    ExName  string           // 交換機名稱
    ExType  string           // 交換機類型
    Dns     string              //鏈接地址
}



// 鏈接rabbitMQ
func (r *RabbitMQ)MqConnect() (err error){

    mqConn, err = amqp.Dial(r.dns)
    r.connection = mqConn   // 賦值給RabbitMQ對象

    if err != nil {
        fmt.Printf("rbmq鏈接失敗  :%s \n", err)
    }

    return
}

// 關(guān)閉mq鏈接
func (r *RabbitMQ)CloseMqConnect() (err error){

    err = r.connection.Close()
    if err != nil{
        fmt.Printf("關(guān)閉mq鏈接失敗  :%s \n", err)
    }
    return
}

// 鏈接rabbitMQ
func (r *RabbitMQ)MqOpenChannel() (err error){
    mqConn := r.connection
    r.Channel, err = mqConn.Channel()
    //defer mqChan.Close()
    if err != nil {
        fmt.Printf("MQ打開管道失敗:%s \n", err)
    }
    return err
}

// 鏈接rabbitMQ
func (r *RabbitMQ)CloseMqChannel() (err error){
    r.Channel.Close()
    if err != nil {
        fmt.Printf("關(guān)閉mq鏈接失敗  :%s \n", err)
    }
    return err
}




// 創(chuàng)建一個新的操作對象
func NewMq(q QueueExchange) RabbitMQ {
    return RabbitMQ{
        QueueName:q.QuName,
        RoutingKey:q.RtKey,
        ExchangeName: q.ExName,
        ExchangeType: q.ExType,
        dns:q.Dns,
    }
}

func (mq *RabbitMQ) sendMsg (body string) (err error)  {
    err = mq.MqOpenChannel()
    ch := mq.Channel
    if err != nil{
        log.Printf("Channel err  :%s \n", err)
    }

    defer func() {
        _ = mq.Channel.Close()
    }()
    if mq.ExchangeName != "" {
        if mq.ExchangeType == ""{
            mq.ExchangeType = "direct"
        }
        err =  ch.ExchangeDeclare(mq.ExchangeName, mq.ExchangeType, true, false, false, false, nil)
        if err != nil {
            log.Printf("ExchangeDeclare err  :%s \n", err)
        }
    }


    // 用于檢查隊列是否存在,已經(jīng)存在不需要重復(fù)聲明
    _, err = ch.QueueDeclare(mq.QueueName, true, false, false, false, nil)
    if err != nil {
        log.Printf("QueueDeclare err :%s \n", err)
    }
    // 綁定任務(wù)
    if mq.RoutingKey != "" && mq.ExchangeName != "" {
        err = ch.QueueBind(mq.QueueName, mq.RoutingKey, mq.ExchangeName, false, nil)
        if err != nil {
            log.Printf("QueueBind err :%s \n", err)
        }
    }

    if mq.ExchangeName != "" && mq.RoutingKey != ""{
        err = mq.Channel.Publish(
            mq.ExchangeName,     // exchange
            mq.RoutingKey, // routing key
            false,  // mandatory
            false,  // immediate
            amqp.Publishing {
                ContentType: "text/plain",
                Body:        []byte(body),
                DeliveryMode: 2,
            })
    }else{
        err = mq.Channel.Publish(
            "",     // exchange
            mq.QueueName, // routing key
            false,  // mandatory
            false,  // immediate
            amqp.Publishing {
                ContentType: "text/plain",
                Body:        []byte(body),
                DeliveryMode: 2,
            })
    }
    return

}


/*
發(fā)送延時消息
 */
func (mq *RabbitMQ)sendDelayMsg(body string,ttl int64) (err error){
    err =mq.MqOpenChannel()
    ch := mq.Channel
    if err != nil{
        log.Printf("Channel err  :%s \n", err)
    }
    defer mq.Channel.Close()

    if mq.ExchangeName != "" {
        if mq.ExchangeType == ""{
            mq.ExchangeType = "direct"
        }
        err =  ch.ExchangeDeclare(mq.ExchangeName, mq.ExchangeType, true, false, false, false, nil)
        if err != nil {
            return
        }
    }


    if ttl <= 0{
        return errors.New("發(fā)送延時消息，ttl參數(shù)是必須的")
    }

    table := make(map[string]interface{},3)
    table["x-dead-letter-routing-key"] = mq.RoutingKey
    table["x-dead-letter-exchange"] = mq.ExchangeName
    table["x-message-ttl"] = ttl*1000

    //fmt.Printf("%+v",table)
    //fmt.Printf("%+v",mq)
    // 用于檢查隊列是否存在,已經(jīng)存在不需要重復(fù)聲明
    ttlstring := strconv.FormatInt(ttl,10)
    queueName := fmt.Sprintf("%s_delay_%s",mq.QueueName ,ttlstring)
    routingKey := fmt.Sprintf("%s_delay_%s",mq.QueueName ,ttlstring)
    _, err = ch.QueueDeclare(queueName, true, false, false, false, table)
    if err != nil {
        return
    }
    // 綁定任務(wù)
    if routingKey != "" && mq.ExchangeName != "" {
        err = ch.QueueBind(queueName, routingKey, mq.ExchangeName, false, nil)
        if err != nil {
            return
        }
    }

    header := make(map[string]interface{},1)

    header["retry_nums"] = 0

    var ttl_exchange string
    var ttl_routkey string

    if(mq.ExchangeName != "" ){
        ttl_exchange = mq.ExchangeName
    }else{
        ttl_exchange = ""
    }


    if mq.RoutingKey != "" && mq.ExchangeName != ""{
        ttl_routkey = routingKey
    }else{
        ttl_routkey = queueName
    }

    err = mq.Channel.Publish(
        ttl_exchange,     // exchange
        ttl_routkey, // routing key
        false,  // mandatory
        false,  // immediate
        amqp.Publishing {
            ContentType: "text/plain",
            Body:        []byte(body),
            Headers:header,
        })
    if err != nil {
        return

    }
    return
}


func (mq *RabbitMQ) sendRetryMsg (body string,retry_nums int32,args ...string)  {
    err :=mq.MqOpenChannel()
    ch := mq.Channel
    if err != nil{
        log.Printf("Channel err  :%s \n", err)
    }
    defer mq.Channel.Close()

    if mq.ExchangeName != "" {
        if mq.ExchangeType == ""{
            mq.ExchangeType = "direct"
        }
        err =  ch.ExchangeDeclare(mq.ExchangeName, mq.ExchangeType, true, false, false, false, nil)
        if err != nil {
            log.Printf("ExchangeDeclare err  :%s \n", err)
        }
    }

    //原始路由key
    oldRoutingKey := args[0]
    //原始交換機名
    oldExchangeName := args[1]

    table := make(map[string]interface{},3)
    table["x-dead-letter-routing-key"] = oldRoutingKey
    if oldExchangeName != "" {
        table["x-dead-letter-exchange"] = oldExchangeName
    }else{
        mq.ExchangeName = ""
        table["x-dead-letter-exchange"] = ""
    }

    table["x-message-ttl"] = int64(20000)

    //fmt.Printf("%+v",table)
    //fmt.Printf("%+v",mq)
    // 用于檢查隊列是否存在,已經(jīng)存在不需要重復(fù)聲明
    _, err = ch.QueueDeclare(mq.QueueName, true, false, false, false, table)
    if err != nil {
        log.Printf("QueueDeclare err :%s \n", err)
    }
    // 綁定任務(wù)
    if mq.RoutingKey != "" && mq.ExchangeName != "" {
        err = ch.QueueBind(mq.QueueName, mq.RoutingKey, mq.ExchangeName, false, nil)
        if err != nil {
            log.Printf("QueueBind err :%s \n", err)
        }
    }

    header := make(map[string]interface{},1)

    header["retry_nums"] = retry_nums + int32(1)

    var ttl_exchange string
    var ttl_routkey string

    if(mq.ExchangeName != "" ){
        ttl_exchange = mq.ExchangeName
    }else{
        ttl_exchange = ""
    }


    if mq.RoutingKey != "" && mq.ExchangeName != ""{
        ttl_routkey = mq.RoutingKey
    }else{
        ttl_routkey = mq.QueueName
    }

    //fmt.Printf("ttl_exchange:%s,ttl_routkey:%s \n",ttl_exchange,ttl_routkey)
    err = mq.Channel.Publish(
        ttl_exchange,     // exchange
        ttl_routkey, // routing key
        false,  // mandatory
        false,  // immediate
        amqp.Publishing {
            ContentType: "text/plain",
            Body:        []byte(body),
            Headers:header,
        })
    if err != nil {
        fmt.Printf("MQ任務(wù)發(fā)送失敗:%s \n", err)

    }

}


// 監(jiān)聽接收者接收任務(wù) 消費者
func (mq *RabbitMQ) ListenReceiver(receiver Receiver) {
    err :=mq.MqOpenChannel()
    ch := mq.Channel
    if err != nil{
        log.Printf("Channel err  :%s \n", err)
    }
    defer mq.Channel.Close()
    if mq.ExchangeName != "" {
        if mq.ExchangeType == ""{
            mq.ExchangeType = "direct"
        }
        err =  ch.ExchangeDeclare(mq.ExchangeName, mq.ExchangeType, true, false, false, false, nil)
        if err != nil {
            log.Printf("ExchangeDeclare err  :%s \n", err)
        }
    }


    // 用于檢查隊列是否存在,已經(jīng)存在不需要重復(fù)聲明
    _, err = ch.QueueDeclare(mq.QueueName, true, false, false, false, nil)
    if err != nil {
        log.Printf("QueueDeclare err :%s \n", err)
    }
    // 綁定任務(wù)
    if mq.RoutingKey != "" && mq.ExchangeName != "" {
        err = ch.QueueBind(mq.QueueName, mq.RoutingKey, mq.ExchangeName, false, nil)
        if err != nil {
            log.Printf("QueueBind err :%s \n", err)
        }
    }
    // 獲取消費通道,確保rabbitMQ一個一個發(fā)送消息
    err =  ch.Qos(1, 0, false)
    msgList, err :=  ch.Consume(mq.QueueName, "", false, false, false, false, nil)
    if err != nil {
        log.Printf("Consume err :%s \n", err)
    }
    for msg := range msgList {
        retry_nums,ok := msg.Headers["retry_nums"].(int32)
        if(!ok){
            retry_nums = int32(0)
        }
        // 處理數(shù)據(jù)
        err := receiver.Consumer(msg.Body)
        if err!=nil {
            //消息處理失敗 進入延時嘗試機制
            if retry_nums < 3{
                fmt.Println(string(msg.Body))
                fmt.Printf("消息處理失敗 消息開始進入嘗試  ttl延時隊列 \n")
                retry_msg(msg.Body,retry_nums,QueueExchange{
                        mq.QueueName,
                        mq.RoutingKey,
                        mq.ExchangeName,
                        mq.ExchangeType,
                        mq.dns,
                    })
            }else{
                //消息失敗 入庫db
                fmt.Printf("消息處理3次后還是失敗了 入庫db 釘釘告警 \n")
                receiver.FailAction(err,msg.Body)
            }
            err = msg.Ack(true)
            if err != nil {
                fmt.Printf("確認消息未完成異常:%s \n", err)
            }
        }else {
            // 確認消息,必須為false
            err = msg.Ack(true)

            if err != nil {
                fmt.Printf("消息消費ack失敗 err :%s \n", err)
            }
        }

    }
}

//消息處理失敗之后 延時嘗試
func retry_msg(msg []byte,retry_nums int32,queueExchange QueueExchange){
    //原始隊列名稱 交換機名稱
    oldQName := queueExchange.QuName
    oldExchangeName := queueExchange.ExName
    oldRoutingKey := queueExchange.RtKey
    if oldRoutingKey == "" || oldExchangeName == ""{
        oldRoutingKey = oldQName
    }

    if queueExchange.QuName != "" {
        queueExchange.QuName = queueExchange.QuName + "_retry_3";
    }

    if queueExchange.RtKey != "" {
        queueExchange.RtKey = queueExchange.RtKey + "_retry_3";
    }else{
        queueExchange.RtKey = queueExchange.QuName + "_retry_3";
    }

//fmt.Printf("%+v",queueExchange)

    mq := NewMq(queueExchange)
    _ = mq.MqConnect()

    defer func(){
        _ = mq.CloseMqConnect()
    }()
    //fmt.Printf("%+v",queueExchange)
    mq.sendRetryMsg(string(msg),retry_nums,oldRoutingKey,oldExchangeName)


}


func Send(queueExchange QueueExchange,msg string) (err error){
    mq := NewMq(queueExchange)
    err = mq.MqConnect()
    if err != nil{
        return
    }

    defer func(){
        _ = mq.CloseMqConnect()
    }()

    err = mq.sendMsg(msg)

    return
}

//發(fā)送延時消息
func SendDelay(queueExchange QueueExchange,msg string,ttl int64)(err error){
    mq := NewMq(queueExchange)
    err = mq.MqConnect()
    if err != nil{
        return
    }
    defer func(){
        _ = mq.CloseMqConnect()
    }()
    err = mq.sendDelayMsg(msg,ttl)
    return
}


/*
runNums  開啟并發(fā)執(zhí)行任務(wù)數(shù)量
 */
func Recv(queueExchange QueueExchange,receiver Receiver,otherParams ...int) (err error){
    var (
        exitTask bool
        maxTryConnNums int  //rbmq鏈接失敗后多久嘗試一次
        runNums int
        maxTryConnTimeFromMinute int
    )

    if(len(otherParams) <= 0){
        runNums = 1
        maxTryConnTimeFromMinute = 0
    }else if(len(otherParams) == 1){
        runNums = otherParams[0]
        maxTryConnTimeFromMinute = 0
    }else if(len(otherParams) == 2){
        runNums = otherParams[0]
        maxTryConnTimeFromMinute = otherParams[1]
    }


    //maxTryConnNums := 360 //rbmq鏈接失敗后最大嘗試次數(shù)
    //maxTryConnTime := time.Duration(10) //rbmq鏈接失敗后多久嘗試一次
    maxTryConnNums = maxTryConnTimeFromMinute * 10 * maxTryConnTimeFromMinute//rbmq鏈接失敗后最大嘗試次數(shù)
    maxTryConnTime := time.Duration(6) //rbmq鏈接失敗后多久嘗試一次
    mq := NewMq(queueExchange)
    //鏈接rabbitMQ
    err = mq.MqConnect()
    if(err != nil){
        return
    }

    defer func() {
        if panicErr := recover(); panicErr != nil{
            fmt.Println(recover())
            err = errors.New(fmt.Sprintf("%s",panicErr))
        }
    }()

    //rbmq斷開鏈接后 協(xié)程退出釋放信號
    taskQuit:= make(chan struct{}, 1)
    //嘗試鏈接rbmq
    tryToLinkC := make(chan struct{}, 1)

    //最大嘗試次數(shù)
    tryToLinkMaxNums := make(chan struct{}, 1)

    maxTryNums := 0 //嘗試重啟次數(shù)

    //開始執(zhí)行任務(wù)
    for i:=1;i<=runNums;i++{
        go Recv2(mq,receiver,taskQuit);
    }

    //如果rbmq斷開連接后 嘗試重新建立鏈接
    var tryToLink = func() {
        for {
            maxTryNums += 1
            err = mq.MqConnect()
            if(err == nil){
                tryToLinkC <- struct{}{}
                break
            }
            if(maxTryNums > maxTryConnNums){
                tryToLinkMaxNums <- struct{}{}
                break
            }
            //如果鏈接斷開了 10秒重新嘗試鏈接一次
            time.Sleep(time.Second * maxTryConnTime)
        }
        return
    }
    scheduleTimer := time.NewTimer(time.Millisecond*300)
    exitTask = true
    for{
        select {
        case <-tryToLinkC: //建立鏈接成功后 重新開啟協(xié)程執(zhí)行任務(wù)
            fmt.Println("重新開啟新的協(xié)程執(zhí)行任務(wù)")
            go Recv2(mq,receiver,taskQuit);
        case <-tryToLinkMaxNums://rbmq超出最大鏈接次數(shù) 退出任務(wù)
            fmt.Println("rbmq鏈接超過最大嘗試次數(shù)!")
            exitTask = false
            err = errors.New("rbmq鏈接超過最大嘗試次數(shù)!")
        case <- taskQuit ://rbmq斷開連接后 開始嘗試重新建立鏈接
            fmt.Println("rbmq斷開連接后 開始嘗試重新建立鏈接")
             go tryToLink()
        case <- scheduleTimer.C:
            //fmt.Println("~~~~~~~~~~~~~~~~~~~~~~~")
        }
        // 重置調(diào)度間隔
        scheduleTimer.Reset(time.Millisecond*300)
        if !exitTask{
            break
        }
    }
    fmt.Println("exit")
    return
}


func Recv2(mq RabbitMQ,receiver Receiver,taskQuit chan<- struct{}){
        defer func() {
            fmt.Println("rbmq鏈接失敗,協(xié)程任務(wù)退出~~~~~~~~~~~~~~~~~~~~")
            taskQuit <- struct{}{}
            return
        }()
        // 驗證鏈接是否正常
        err := mq.MqOpenChannel()
        if(err != nil){
            return
        }
        mq.ListenReceiver(receiver)
}


type retryPro struct {
    msgContent   string
}

View Code

二，延時隊列

場景一：物聯(lián)網(wǎng)系統(tǒng)經(jīng)常會遇到向終端下發(fā)命令，如果命令一段時間沒有應(yīng)答，就需要設(shè)置成超時。

場景二：訂單下單之后30分鐘后，如果用戶沒有付錢，則系統(tǒng)自動取消訂單。

最近的一個項目遇到了這種情況，如果運單30分鐘還沒有被接單，則狀態(tài)自動變?yōu)橐讶∠崿F(xiàn)延遲消息原理如下，借用一張圖：

實現(xiàn)方案

定時任務(wù)輪詢數(shù)據(jù)庫，看是否有產(chǎn)生新任務(wù)，如果產(chǎn)生則消費任務(wù)
pcntl_alarm為進程設(shè)置一個鬧鐘信號
swoole的異步高精度定時器:swoole_time_tick(類似javascript的setInterval)和swoole_time_after(相當于javascript的setTimeout)
rabbitmq延遲任務(wù)

以上四種方案,如果生產(chǎn)環(huán)境有使用到swoole建議使用第三種方案。此篇文章重點講述第四種方案實現(xiàn)

生產(chǎn)者：

1 <?php
 2 require_once __DIR__ . '/../vendor/autoload.php';
 3 use PhpAmqpLib\Connection\AMQPStreamConnection;
 4 use PhpAmqpLib\Message\AMQPMessage;
 5 
 6 
 7 $queue = "test_ack_queue";
 8 $exchange = "test_ack_queue";
 9 //獲取連接
10 $connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
11 //從連接中創(chuàng)建通道
12 $channel = $connection->channel();
13 
14 $channel->exchange_declare('delay_exchange', 'direct',false,true,false);
15 $channel->exchange_declare('cache_exchange', 'direct',false,true,false);
16 
17 $tale = new \PhpAmqpLib\Wire\AMQPTable();
18 $tale->set('x-dead-letter-exchange', 'delay_exchange');
19 $tale->set('x-dead-letter-routing-key','delay_exchange');
20 //$tale->set('x-message-ttl',10000);
21 
22 $channel->queue_declare('cache_queue',false,true,false,false,false,$tale);
23 $channel->queue_bind('cache_queue', 'cache_exchange','cache_exchange');
24 
25 $channel->queue_declare('delay_queue',false,true,false,false,false);
26 $channel->queue_bind('delay_queue', 'delay_exchange','delay_exchange');
27 
28 
29 $msg = new AMQPMessage('Hello World',array(
30     'expiration' => 10000,
31     'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT
32 
33 ));
34 
35 $channel->basic_publish($msg,'cache_exchange','cache_exchange');
36 echo date('Y-m-d H:i:s')." [x] Sent 'Hello World!' ".PHP_EOL;
37 
38 
39 
40 
41 //while ($wait) {
42 //    $channel->wait();
43 //}
44 
45 $channel->close();
46 $connection->close();

task

View Code

消費者：

1 <?php
 2 require_once __DIR__ . '/../vendor/autoload.php';
 3 use PhpAmqpLib\Connection\AMQPStreamConnection;
 4 use PhpAmqpLib\Message\AMQPMessage;
 5 
 6 
 7 //獲取連接
 8 $connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
 9 //從連接中創(chuàng)建通道
10 $channel = $connection->channel();
11 
12 
13 //$channel->queue_declare($queue, false, true, false, false);
14 //$channel->exchange_declare($exchange, 'topic', false, true, false);
15 //$channel->queue_bind($queue, $exchange);
16 
17 
18 
19 $channel->exchange_declare('delay_exchange', 'direct',false,false,false);
20 $channel->queue_declare('delay_queue',false,true,false,false,false);
21 $channel->queue_bind('delay_queue', 'delay_exchange','delay_exchange');
22 
23 
24 
25 function process_message(AMQPMessage $message)
26 {
27     $headers = $message->get('application_headers');
28     $nativeData = $headers->getNativeData();
29 //    var_dump($nativeData['x-delay']);
30     echo date('Y-m-d H:i:s')." [x] Received",$message->body,PHP_EOL;
31     $message->delivery_info['channel']->basic_ack($message->delivery_info['delivery_tag']);
32 
33 }
34 
35 
36 $channel->basic_qos(null, 1, null);
37 $channel->basic_consume('delay_queue', '', false, false, false, false, 'process_message');
38 
39 function shutdown($channel, $connection)
40 {
41     $channel->close();
42     $connection->close();
43 }
44 register_shutdown_function('shutdown', $channel, $connection);
45 
46 while (count($channel->callbacks)) {
47     $channel->wait();
48 }
work

View Code

延時隊列實現(xiàn)和上面所講的消息重試有異曲同工之處，都是利用了延時時間和死信隊列這一特性實現(xiàn)

最新源碼倉庫地址：https://github.com/sunlongv520/golang-rabbitmq

其它：該rabbitmq包實現(xiàn)中包含了，rabbitmq斷線重連，有興趣的同學(xué)可以看看

　　（重試和重連接是兩個概念）

　　重連接：rabbitmq鏈接失敗導(dǎo)致任務(wù)失敗，此時要等待rabbitmq服務(wù)器恢復(fù)正常后才能再次啟動協(xié)程處理任務(wù)

　　重試：rabbitmq服務(wù)正常，消息消費進程也正常，但是消息處理失敗。嘗試多次消費消息后還是失敗就ack消息，在整個重試過程中不會阻塞消費

golang監(jiān)聽rabbitmq消息隊列任務(wù)斷線自動重連接：http://www.rzrgm.cn/sunlong88/p/15959476.html

posted on 2019-12-04 14:38 孫龍-程序員閱讀(2692) 評論(0) 收藏舉報

刷新頁面返回頂部