生產(chǎn)者代碼:

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" // 死信交換機(jī)

DeadLettersQueueName = "dlx_queue_packet" // 死信隊(duì)列

QueueName = "queue_packet" // 目標(biāo)隊(duì)列

ExchangeName = "exchange_packet" // 目標(biāo)交換機(jī)

)

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服務(wù)器失敗:%s", err.Error())

return

}

defer conn.Close()

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

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

return

}

defer ch.Close()

// 聲明隊(duì)列交換機(jī)

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

logger.Error("聲明業(yè)務(wù)交換機(jī)失敗:%s", err.Error())

return

}

// 創(chuàng)建死信交換機(jī)

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

logger.Error("創(chuàng)建死信交換機(jī):%s", err.Error())

return

}

// 創(chuàng)建死信隊(duì)列

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

logger.Error("創(chuàng)建死信隊(duì)列失敗:%s", err.Error())

return

}

// 創(chuàng)建業(yè)務(wù)隊(duì)列

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

"x-message-ttl": 6000, // 消息過(guò)期時(shí)間 毫秒

"x-dead-letter-exchange": DeadLettersExchangeName, // 死信交換機(jī)

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

}); err != nil {

logger.Warn("創(chuàng)建業(yè)務(wù)隊(duì)列失敗:%s", err.Error())

return

}

// 業(yè)務(wù)隊(duì)列綁定交換機(jī)

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

logger.Error("綁定業(yè)務(wù)交換機(jī)失敗:%s", err.Error())

return

}

// 死信隊(duì)列綁定死信交換機(jī)

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

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

}

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

msg := amqp.Publishing{

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

ContentType: "text/plain",

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

}

// 發(fā)布消息

err = ch.Publish(

ExchangeName,

"",

false,

false,

msg,

)

if err != nil {

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

return

} else {

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

}

}

}

消費(fèi)者代碼

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" // 死信交換機(jī)

DeadLettersQueueName = "dlx_queue_packet" // 死信隊(duì)列

QueueName = "queue_packet" // 目標(biāo)隊(duì)列

ExchangeName = "exchange_packet" // 目標(biāo)交換機(jī)

)

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服務(wù)器失敗:%s", err.Error())

return

}

defer conn.Close()

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

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

return

}

defer ch.Close()

// 創(chuàng)建死信交換機(jī)

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

logger.Error("創(chuàng)建死信交換機(jī):%s", err.Error())

return

}

// 創(chuàng)建死信隊(duì)列

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

logger.Error("創(chuàng)建死信隊(duì)列失敗:%s", err.Error())

return

}

// 死信隊(duì)列綁定死信交換機(jī)

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

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

}

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

if err != nil {

logger.Error("消費(fèi)者監(jiān)聽(tīng)失敗:%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("確認(rèn)消息失敗")

}

}(message)

case

}

}

}

有疑問(wèn)加站長(zhǎng)微信聯(lián)系(非本文作者)

總結(jié)

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

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。