RabbitMqBase消息队列小结
虽然这个消息队列我只是知道应用场景在哪里，还没有实际操作到，但是原理还是要知道的。这些知识点就像珍珠，万一哪天就用到了。在没具体学习之前，我一直在想: 这玩意不就是个队列吗。还能玩出花来？结果，一研究，还真的打脸了。
首先，这玩意用的是AMQP协议，并且只是占用了一个tcp连接，然后就会问到，多个消息，那不得卡死啊，不。他用了channel,这个东西可以有多个，可以理解为他开辟了线程，这样就避免了网络连接的资源浪费。
消息流程
消息 => 交换机 => 通过绑定的key => 队列 => 消费者
交换机发布消息的方式有
- 直接发送
direct: 消息一根肠子到队列 - 广播
fanout: 交换机发布消息，无需指定key,绑定在交换机上面的队列就会知道消息 - 匹配消息
topic: 交换机发布消息根据规则匹配到满足规则的key上， 和正则匹配差不多一个意思
延时队列
对于我们想发布一个定时任务，我们不想让这个任务占用我们自己的电脑资源，就要发布完任务，然后就关电脑睡觉。这时候，我们就需要一个mq服务器，把消息塞进去，首先想的就是让消息自己根据时间戳排序，不好意思，想多了，很难实现。那就换一种思路，消息设置过期时间，然后就把消息发到别的转换机上，再去发到指定队列上，任务监听队列就好了。
代码实现
-
发布消息
send.gopackage dead_q import ( "github.com/streadway/amqp" "go-mq/conf" "log" ) func failOnError(err error, msg string) { if err != nil { log.Fatalf("%s: %s", msg, err) } } func Send() { url := conf.Host conn, err := amqp.Dial(url) 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 := "dead-last" // 将消息发送到延时队列上 err = ch.Publish( "delay-exchange", // exchange 这里为空则不选择 exchange "", // routing key true, // mandatory false, // immediate amqp.Publishing{ ContentType: "text/plain", Body: []byte(body), Expiration: "5000", // 设置五秒的过期时间 }) failOnError(err, "Failed to publish a message") log.Printf(" [x] Sent %s", body) } -
第一个交换机为了设置超时时间和下一个交换机
rec.gopackage rec import ( "github.com/streadway/amqp" "go-mq/conf" "log" ) func failOnError(err error, msg string) { if err != nil { log.Fatalf("%s: %s", msg, err) } } func Rec() { url := conf.Host conn, err := amqp.Dial(url) failOnError(err, "Failed to connect to RabbitMQ") defer conn.Close() ch, err := conn.Channel() failOnError(err, "Failed to open a channel") defer ch.Close() // 声明一个主要使用的 exchange err = ch.ExchangeDeclare( "delay-exchange", // name "fanout", // type true, // durable false, // auto-deleted false, // internal false, // no-wait nil, // arguments ) failOnError(err, "Failed to declare an exchange") /** * 注意,这里是重点!!!!! * 声明一个延时队列, ß我们的延时消息就是要发送到这里 */ q, errDelay := ch.QueueDeclare( "delay-qune", // name true, // durable false, // delete when unused false, // exclusive false, // no-wait amqp.Table{ // 当消息过期时把消息发送到 logs 这个 exchange "x-dead-letter-exchange":"dead-exchange", }, // arguments ) failOnError(errDelay, "Failed to declare a delay_queue") err = ch.QueueBind( q.Name, // queue name, 这里指的是 test_logs "", // routing key "delay-exchange", // exchange false, nil) failOnError(err, "Failed to bind a queue") // 这里监听的是 test_logs msgs, err := ch.Consume( q.Name, // queue name, 这里指的是 test_logs "", // consumer false, // 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(" [x] %s", d.Body) d.Ack(false) } }() log.Printf(" [*] Waiting for logs. To exit press CTRL+C") <-forever } -
最后的队列用于任务消费
dedrec.gopackage dead_rec import ( "github.com/streadway/amqp" "go-mq/conf" "log" ) func failOnError(err error, msg string) { if err != nil { log.Fatalf("%s: %s", msg, err) } } func Rec() { url := conf.Host conn, err := amqp.Dial(url) failOnError(err, "Failed to connect to RabbitMQ") defer conn.Close() ch, err := conn.Channel() failOnError(err, "Failed to open a channel") defer ch.Close() // 声明一个主要使用的 exchange err = ch.ExchangeDeclare( "dead-exchange", // name amqp.ExchangeFanout, // type true, // durable false, // auto-deleted false, // internal false, // no-wait nil, // arguments ) failOnError(err, "Failed to declare an exchange") /** * 注意,这里是重点!!!!! * 声明一个延时队列, ß我们的延时消息就是要发送到这里 */ arg2 := make(map[string]interface{}) arg2["x-dead-letter-exchange"] = "dead-q" arg2["x-max-length"] = 3 delayQune, _ := ch.QueueDeclare( "dead-qune", // name true, // durable false, // delete when unused false, // exclusive false, // no-wait arg2, // arguments ) err = ch.QueueBind( delayQune.Name, // queue name, 这里指的是 test_logs "", // routing key "dead-exchange", // exchange false, nil) failOnError(err, "Failed to bind a queue") // 这里监听的是 test_logs msgs, err := ch.Consume( delayQune.Name, // queue name, 这里指的是 test_logs "", // consumer false, // 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(" [x] %s", d.Body) d.Ack(false) } }() log.Printf(" [*] Waiting for logs. To exit press CTRL+C") <-forever }
附加知识点
-
消费者公平消费
if err := ch.Qos(1, 0, false); err != nil { failOnError(err, "Failed to set QoS") } -
设置队列属性
arg2 := make(map[string]interface{}) arg2["x-dead-letter-exchange"] = "dead-q" arg2["x-max-length"] = 3 delayQune, _ := ch.QueueDeclare( "dead-qune", // name true, // durable false, // delete when unused false, // exclusive false, // no-wait arg2, // arguments )