利用rabbitmq延迟消息实现订单过期取消

过期订单消费端：

package main

import (
	"log"

	"github.com/streadway/amqp"
)

func failOnError(err error, msg string) {
	if err != nil {
		log.Fatalf("%s: %s", msg, err)
	}
}


func main() {
	orderExchange:="order_exchange"//订单交换机
	orderQueue := "order_queue"//订单队列，消息发送到这个队列，但没有消费者消费这个队列的消息
	orderRoutingKey := "order"//订单路由key
	deadExchange:="dead_exchange"//死信交换机
	deadQueue := "dead_queue"//死信队列

	// 建立链接
	conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
	failOnError(err, "Failed to connect to RabbitMQ")
	defer conn.Close()

	ch, err := conn.Channel()
	failOnError(err, "Failed to open a channel")
	defer ch.Close()

	// 声明订单交换机
	err = ch.ExchangeDeclare(
		orderExchange,   // name
		"direct", // type
		true,     // durable
		false,    // auto-deleted
		false,    // internal
		false,    // no-wait
		nil,      // arguments
	)
	failOnError(err, "Failed to declare an exchange")

	// 声明订单队列，延迟消息发送到这个队列
	_, err = ch.QueueDeclare(
		orderQueue,    // name
		false, // durable
		false, // delete when unused
		true,  // exclusive
		false, // no-wait
		amqp.Table{
			// 当消息过期时把消息发送到 死信 exchange
			"x-dead-letter-exchange":deadExchange,
		},   // arguments
	)
	failOnError(err, "Failed to declare a queue")

	//绑定订单队列与订单交换机
	err = ch.QueueBind(
		orderQueue, // queue name, 这里指的是 test_logs
		orderRoutingKey,     // routing key
		orderExchange, // exchange
		false,
		nil)
	failOnError(err, "Failed to bind a queue")

	// 声明死信交换机
	err = ch.ExchangeDeclare(
		deadExchange,   // name
		"fanout", // type
		true,     // durable
		false,    // auto-deleted
		false,    // internal
		false,    // no-wait
		nil,      // arguments
	)
	failOnError(err, "Failed to declare an exchange")

	// 声明死信队列，消息过期会到这个队列
	_, err = ch.QueueDeclare(
		deadQueue,    // name
		false, // durable
		false, // delete when unused
		true,  // exclusive
		false, // no-wait
		nil,   // arguments
	)
	failOnError(err, "Failed to declare a queue")

	//绑定死信队列与死信交换机
	err = ch.QueueBind(
		deadQueue, // queue name,
		"",     // routing key
		deadExchange, // exchange
		false,
		nil)
	failOnError(err, "Failed to bind a queue")


	// 消费死信队列中的消息
	msgs, err := ch.Consume(
		deadQueue, // queue name,
		"",     // consumer
		true,   // auto-ack
		false,  // exclusive
		false,  // no-local
		false,  // no-wait
		nil,    // args
	)
	failOnError(err, "Failed to register a consumer")

	forever := make(chan bool)

	go func() {
		for d := range msgs {
			log.Printf("get msg, [订单号] %s", d.Body)
			//判断订单是否支付，未支付则取消订单，归还库存。。。
		}
	}()

	log.Printf(" [*] Waiting for msg. To exit press CTRL+C")
	<-forever
}

消息发送端：

package main

import (
	"log"
	"os"
	"strings"

	"github.com/streadway/amqp"
)

func failOnError(err error, msg string) {
	if err != nil {
		log.Fatalf("%s: %s", msg, err)
	}
}

func main() {
	orderExchange:="order_exchange"//订单交换机
	orderRoutingKey := "order"//订单路由key

	conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
	failOnError(err, "Failed to connect to RabbitMQ")
	defer conn.Close()

	ch, err := conn.Channel()
	failOnError(err, "Failed to open a channel")
	defer ch.Close()

	body := bodyFrom(os.Args)
	// 将消息发送到延时队列上
	err = ch.Publish(
		orderExchange,                 // exchange
		orderRoutingKey,       // routing key
		false,              // mandatory
		false,              // immediate
		amqp.Publishing{
			ContentType: "text/plain",
			Body:        []byte(body),
			Expiration: "10000", // 设置10秒的过期时间
		})
	failOnError(err, "Failed to publish a message")

	log.Printf("  Send msg: %s", body)
}

func bodyFrom(args []string) string {
	var s string
	if (len(args) < 2) || os.Args[1] == "" {
		s = "10086"
	} else {
		s = strings.Join(args[1:], " ")
	}
	return s
}

执行结果

发送端：

PS E:gocoderabbitmqproducer> go run .main.go 6666
2022/06/27 21:52:41   Send msg: 6666

消费端：

PS E:gocoderabbitmqconsumer> go run .main.go
2022/06/27 21:52:30  [*] Waiting for msg. To exit press CTRL+C
2022/06/27 21:52:51 get msg, [订单号] 6666

可以看到发送和消费消息的时间差刚好是延迟时间10秒