生產者代碼:

package main

import (

uuid "github.com/satori/go.uuid"

"github.com/streadway/amqp"

"github.com/wonderivan/logger"

"rmq/db/rmq"

"time"

)

const (

DeadLettersExchangeName = "dlx_exchange_packet" // 死信交換機

DeadLettersQueueName = "dlx_queue_packet" // 死信隊列

QueueName = "queue_packet" // 目標隊列

ExchangeName = "exchange_packet" // 目標交換機

)

var (

ch *amqp.Channel

err error

conn *amqp.Connection

queue amqp.Queue

dlxQueue amqp.Queue

)

func main() {

if conn, err = rmq.GetConn(); err != nil {

logger.Error("連接RabbitMQ服務器失敗:%s", err.Error())

return

}

defer conn.Close()

if ch, err = conn.Channel(); err != nil {

logger.Error("獲取Channel失敗:%s", err.Error())

return

}

defer ch.Close()

// 聲明隊列交換機

if err = ch.ExchangeDeclare(ExchangeName, amqp.ExchangeFanout, true, false, false, false, nil); err != nil {

logger.Error("聲明業務交換機失敗:%s", err.Error())

return

}

// 創建死信交換機

if err = ch.ExchangeDeclare(DeadLettersExchangeName, amqp.ExchangeDirect, true, false, false, false, nil); err != nil {

logger.Error("創建死信交換機:%s", err.Error())

return

}

// 創建死信隊列

if dlxQueue, err = ch.QueueDeclare(DeadLettersQueueName, true, false, false, false, nil); err != nil {

logger.Error("創建死信隊列失敗:%s", err.Error())

return

}

// 創建業務隊列

if queue, err = ch.QueueDeclare(QueueName, true, false, false, false, amqp.Table{

"x-message-ttl": 6000, // 消息過期時間 毫秒

"x-dead-letter-exchange": DeadLettersExchangeName, // 死信交換機

// "x-dead-letter-routing-key": "dlxKey", // 死信路由key

}); err != nil {

logger.Warn("創建業務隊列失敗:%s", err.Error())

return

}

// 業務隊列綁定交換機

if err = ch.QueueBind(queue.Name, "", ExchangeName, false, nil); err != nil {

logger.Error("綁定業務交換機失敗:%s", err.Error())

return

}

// 死信隊列綁定死信交換機

if err = ch.QueueBind(dlxQueue.Name, "", DeadLettersExchangeName, false, nil); err != nil {

logger.Error("綁定死信交換機失敗:%s", err.Error())

}

for i := 1; i <= 10; i++ {

msg := amqp.Publishing{

MessageId: uuid.NewV4().String(),

ContentType: "text/plain",

Body: []byte("紅包退回"),

}

// 發布消息

err = ch.Publish(

ExchangeName,

"",

false,

false,

msg,

)

if err != nil {

logger.Error("發送失敗: %s", err.Error())

return

} else {

logger.Info("發送成功:%s", msg.MessageId)

}

}

}

消費者代碼

package main

import (

uuid "github.com/satori/go.uuid"

"github.com/streadway/amqp"

"github.com/wonderivan/logger"

"rmq/db/rmq"

"time"

)

const (

DeadLettersExchangeName = "dlx_exchange_packet" // 死信交換機

DeadLettersQueueName = "dlx_queue_packet" // 死信隊列

QueueName = "queue_packet" // 目標隊列

ExchangeName = "exchange_packet" // 目標交換機

)

var (

ch *amqp.Channel

err error

conn *amqp.Connection

queue amqp.Queue

dlxQueue amqp.Queue

)

func main() {

if conn, err = rmq.GetConn(); err != nil {

logger.Error("連接RabbitMQ服務器失敗:%s", err.Error())

return

}

defer conn.Close()

if ch, err = conn.Channel(); err != nil {

logger.Error("獲取Channel失敗:%s", err.Error())

return

}

defer ch.Close()

// 創建死信交換機

if err = ch.ExchangeDeclare(DeadLettersExchangeName, amqp.ExchangeDirect, true, false, false, false, nil); err != nil {

logger.Error("創建死信交換機:%s", err.Error())

return

}

// 創建死信隊列

if dlxQueue, err = ch.QueueDeclare(DeadLettersQueueName, true, false, false, false, nil); err != nil {

logger.Error("創建死信隊列失敗:%s", err.Error())

return

}

// 死信隊列綁定死信交換機

if err = ch.QueueBind(dlxQueue.Name, "", DeadLettersExchangeName, false, nil); err != nil {

logger.Error("綁定死信交換機失敗:%s", err.Error())

}

msgList, err := ch.Consume(dlxQueue.Name, "", false, false, false, false, nil)

if err != nil {

logger.Error("消費者監聽失敗:%s", err.Error())

return

}

for {

select {

case message, ok :=

if !ok {

continue

}

go func(msg amqp.Delivery) {

logger.Info("messageID: %s", msg.MessageId)

logger.Info("messageBody: %s", msg.Body)

if err = msg.Ack(false); err != nil {

logger.Error("確認消息失敗")

}

}(message)

case

}

}

}

有疑問加站長微信聯系(非本文作者)

總結

以上是生活随笔為你收集整理的php 红包过期退回,RabbitMQ功能实现1- 红包未领取退回的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。