quick start

package amqp_test

import (
	"fmt"
	"github.com/streadway/amqp"
	"testing"
)

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

func TestAmqp(t *testing.T) {
	// 创建连接
	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()

	q, err := ch.QueueDeclare(
		"hello-streadway-amqp", // name
		false,                  // durable
		false,                  // delete when unused
		false,                  // exclusive
		false,                  // no-wait
		nil,                    // arguments
	)
	failOnError(err, "Failed to declare a queue")

	msgs, err := ch.Consume(
		q.Name, // queue
		"",     // consumer
		true,   // auto-ack
		false,  // exclusive
		false,  // no-local
		false,  // no-wait
		nil,    // args
	)
	failOnError(err, "Failed to register a consumer")

	go func() {
		for d := range msgs {
			fmt.Println("---", string(d.Body))
		}
	}()

	body := "Hello World!"
	err = ch.Publish(
		"",     // exchange
		q.Name, // routing key
		false,  // mandatory
		false,  // immediate
		amqp.Publishing{
			ContentType: "text/plain",
			Body:        []byte(body),
		})
	failOnError(err, "Failed to publish a message")

	select {}
}

延时队列

• x-dead-letter-exchange 将过期的消息发送到指定的 exchange 中
• x-dead-letter-routing-key 将过期的消息发送到自定的 route当中

消费者comsumer.go

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() {
    // 建立链接
    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()

    // 声明一个主要使用的 exchange
    err = ch.ExchangeDeclare(
        "logs",   // name
        "fanout", // type
        true,     // durable
        false,    // auto-deleted
        false,    // internal
        false,    // no-wait
        nil,      // arguments
    )
    failOnError(err, "Failed to declare an exchange")

    // 声明一个常规的队列, 其实这个也没必要声明,因为 exchange 会默认绑定一个队列
    q, err := ch.QueueDeclare(
        "test_logs",    // name
        false, // durable
        false, // delete when unused
        true,  // exclusive
        false, // no-wait
        nil,   // arguments
    )
    failOnError(err, "Failed to declare a queue")

    /**
     * 注意,这里是重点!!!!!
     * 声明一个延时队列, ß我们的延时消息就是要发送到这里
     */
    _, errDelay := ch.QueueDeclare(
        "test_delay",    // name
        false, // durable
        false, // delete when unused
        true,  // exclusive
        false, // no-wait
        amqp.Table{
            // 当消息过期时把消息发送到 logs 这个 exchange
            "x-dead-letter-exchange":"logs",
        },   // arguments
    )
    failOnError(errDelay, "Failed to declare a delay_queue")

    err = ch.QueueBind(
        q.Name, // queue name, 这里指的是 test_logs
        "",     // routing key
        "logs", // exchange
        false,
        nil)
    failOnError(err, "Failed to bind a queue")

    // 这里监听的是 test_logs
    msgs, err := ch.Consume(
        q.Name, // queue name, 这里指的是 test_logs
        "",     // 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(" [x] %s", d.Body)
        }
    }()

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

生产者productor.go

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() {
    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(
        "",                 // exchange 这里为空则不选择 exchange
        "test_delay",         // routing key
        false,              // 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)
}

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