生产者
package main
import (
"fmt"
"github.com/streadway/amqp"
"strconv"
"time"
)
// 启动生产者，n秒之内，如果consume没有消费，那么就会被die-死信队列消费
func main() {
// # ========== 1.创建连接 ==========
mq, _ := amqp.Dial("amqp://guest:guest@192.168.252.128:5672/")
mqCh, _ := mq.Channel()
defer mq.Close()
// # ========== 2.设置队列（队列、交换机、绑定） ==========
// 1：声明队列
var err error
//1:队列名 2：是否持久化3：是否自动删除4：是不是私有5是否等待6：额外参数 x-message-ttl: 消息过期时间，单位是毫秒。超过该时间的消息会被认为是过期消息。
//x-dead-letter-exchange: 指定死信交换机，用于处理变成死信的消息。
//x-dead-letter-routing-key: 指定死信的路由键，用于将死信路由到指定的队列中进行处理。
//x-max-priority: 指定队列支持的最大消息优先级。
_, err = mqCh.QueueDeclare("constant.NormalQueue", true, false, false, false, amqp.Table{
"x-message-ttl": 5000, // 消息过期时间,毫秒
"x-dead-letter-exchange": "constant.DeadExchange", // 指定死信交换机
"x-dead-letter-routing-key": "constant.DeadRoutingKey", // 指定死信routing-key
})
fmt.Println(err, "创建normal队列失败")
// 1.1 声明交换机
//1：交换机名字2：交换机类型（直接，订阅，主题。。）3持久化 4自动删除 5
err = mqCh.ExchangeDeclare("constant.NormalExchange", amqp.ExchangeDirect, true, false, false, false, nil)
fmt.Println(err, "创建normal交换机失败")
// 1.2 队列绑定（将队列、routing-key、交换机三者绑定到一起）
err = mqCh.QueueBind("constant.NormalQueue", "constant.NormalRoutingKey", "constant.NormalExchange", false, nil)
fmt.Println(err, "normal：队列、交换机、routing-key 绑定失败")
// # ========== 3.设置死信队列（队列、交换机、绑定） ==========
// 2 声明死信队列
// args 为 nil。切记不要给死信队列设置消息过期时间,否则失效的消息进入死信队列后会再次过期。
_, err = mqCh.QueueDeclare("constant.DeadQueue", true, false, false, false, nil)
fmt.Println(err, "创建dead队列失败")
// 2.1 声明交换机
err = mqCh.ExchangeDeclare("constant.DeadExchange", amqp.ExchangeDirect, true, false, false, false, nil)
fmt.Println(err, "创建dead队列失败")
// 2.2 队列绑定（将队列、routing-key、交换机三者绑定到一起）
err = mqCh.QueueBind("constant.DeadQueue", "constant.DeadRoutingKey", "constant.DeadExchange", false, nil)
fmt.Println(err, "dead：队列、交换机、routing-key 绑定失败")
// # ========== 4.发布消息 ==========
message := "msg" + strconv.Itoa(int(time.Now().Unix()))
fmt.Println(message)
// 发布消息
//mandatory 如果为true回根据exchange和routekey的规则，如果无法找到符合规则的队列，就会把消息返回生产者
//immediate 如果为true，当exchange发现队列没有绑定消费者，就会把消息返回给生产者
err = mqCh.Publish("constant.NormalExchange", "constant.NormalRoutingKey", false, false, amqp.Publishing{
ContentType: "text/plain",
Body: []byte(message),
})
fmt.Println(err, "消息发布失败")
}
消费者
package main
import (
"fmt"
"github.com/streadway/amqp"
"log"
)
func main() {
// # ========== 1.创建连接 ==========
mq, _ := amqp.Dial("amqp://guest:guest@192.168.252.128:5672/")
mqCh, _ := mq.Channel()
defer mq.Close()
// # ========== 2.消费消息 ==========
msgsCh, err := mqCh.Consume("constant.NormalQueue", "", false, false, false, false, nil)
fmt.Println(err, "消费normal队列失败")
forever := make(chan bool)
go func() {
for d := range msgsCh {
// 要实现的逻辑
log.Printf("接收的消息: %s", d.Body)
// 手动应答
//d.Ack(true)
d.Reject(true) //是否拒绝,如果拒绝的话，n秒之后，就会被死信消费,或者直接不消费
}
}()
log.Printf("[*] Waiting for message, To exit press CTRL+C")
<-forever
}
死信消费者
package main
import (
"fmt"
"github.com/streadway/amqp"
"log"
)
func main() {
// # ========== 1.创建连接 ==========
mq, _ := amqp.Dial("amqp://guest:guest@192.168.252.128:5672/")
mqCh, _ := mq.Channel()
defer mq.Close()
// # ========== 2.消费死信消息 ==========
msgsCh, err := mqCh.Consume("constant.DeadQueue", "", false, false, false, false, nil)
fmt.Println(err, "消费dead队列失败")
forever := make(chan bool)
go func() {
for d := range msgsCh {
// 要实现的逻辑
log.Printf("接收的消息: %s", d.Body)
// 手动应答
d.Ack(true)
}
}()
log.Printf("[*] Waiting for message, To exit press CTRL+C")
<-forever
}