RabbitMQ入门之Go语言教程-(4)

RabbitMQ官网的消息模型的第四种-路由模式

路由(routing)

(using the Go RabbitMQ client)

在上一教程中，我们构建了一个简单的日志记录系统。我们能够向许多接收者广播日志消息。

在本教程中，我们将向它添加一个特性-我们将使它能够只订阅消息的一个子集。例如，我们将只能将关键错误消息定向到日志文件（以节省磁盘空间），同时仍然能够在控制台上打印所有日志消息。

绑定

在前面的示例中，我们已经在创建绑定。你可能会想起以下代码：

1
2
3
4
5
6


err = ch.QueueBind(
  q.Name, // queue name
  "",     // routing key
  "logs", // exchange
  false,
  nil)

绑定是交换器和队列之间的关系。这可以简单地理解为：队列对来自此交换器的消息感兴趣。

绑定可以采用额外的routing_key参数。为了避免与Channel.Publish参数混淆，我们将其称为binding key。这是我们如何使用键创建绑定的方法：

1
2
3
4
5
6


err = ch.QueueBind(
  q.Name,    // queue name
  "black",   // routing key
  "logs",    // exchange
  false,
  nil)

绑定密钥的含义取决于交换器的类型。我们以前使用的fanout交换器只是忽略了这个值。

Direct交换器

我们上一个教程中的日志系统将所有消息广播给所有消费者。我们希望扩展这一点，允许根据消息的严重性过滤消息。例如，我们可能希望将日志消息写入磁盘的脚本只接收严重错误，而不会在warning或info日志消息上浪费磁盘空间。

我们使用fanout交换器，这并没有给我们很大的灵活性——它只能进行无脑广播。

我们将使用direct交换器。direct交换器背后的路由算法很简单——消息进入其binding key与消息的routing key完全匹配的队列。

为了说明这一点，请考虑以下设置：

在此设置中，我们可以看到绑定了两个队列的direct交换器X。第一个队列绑定键为orange，第二个队列绑定为两个，一个绑定键为black，另一个为green。

在这种设置中，使用orange路由键发布到交换器的消息将被路由到队列Q1。路由键为black或green的消息将转到Q2。所有其他消息将被丢弃。

多重绑定

用相同的绑定键绑定多个队列是完全合法的。在我们的示例中，我们可以使用绑定键black在X和Q1之间添加绑定。在这种情况下，direct交换器的行为将类似fanout，并将消息广播到所有匹配的队列。带有black路由键的消息将同时传递给Q1和Q2。

发送日志

我们将在日志系统中使用这个模型。我们将发送消息到direct交换器，而不是fanout。我们将提供严重性（译注：通常我们使用日志级别划分日志信息的严重程度）作为路由键。这样，接收脚本将能够选择其想要接收的日志级别。让我们首先关注发送日志。

与以前一样，我们需要首先创建一个交换器：

1
2
3
4
5
6
7
8
9


err = ch.ExchangeDeclare(
  "logs_direct", // name
  "direct",      // type
  true,          // durable
  false,         // auto-deleted
  false,         // internal
  false,         // no-wait
  nil,           // arguments
)

我们已经准备好发送一条消息：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21


err = ch.ExchangeDeclare(
  "logs_direct", // name
  "direct",      // type
  true,          // durable
  false,         // auto-deleted
  false,         // internal
  false,         // no-wait
  nil,           // arguments
)
failOnError(err, "Failed to declare an exchange")

body := bodyFrom(os.Args)
err = ch.Publish(
  "logs_direct",         // exchange
  severityFrom(os.Args), // routing key
  false, // mandatory
  false, // immediate
  amqp.Publishing{
    ContentType: "text/plain",
    Body:        []byte(body),
})

为了简化问题，我们假设“严重性”可以是“info”、“warning”、“error”之一。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26


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

if len(os.Args) < 2 {
  log.Printf("Usage: %s [info] [warning] [error]", os.Args[0])
  os.Exit(0)
}
// 建立多个绑定关系
for _, s := range os.Args[1:] {
  log.Printf("Binding queue %s to exchange %s with routing key %s",
     q.Name, "logs_direct", s)
  err = ch.QueueBind(
    q.Name,        // queue name
    s,             // routing key
    "logs_direct", // exchange
    false,
    nil)
  failOnError(err, "Failed to bind a queue")
}

完整demo

1.Routing之订阅模型-Direct(直连)

在Fanout模式中，一条消息，会被所有订阅的队列所消费。但是，在某些场景下，我们希望不同的消息被不同的队列消费，这时就要用到Direct类型的Exchange。

在Direct模型下：

队列与交换机的绑定，不能是任意绑定了，而是要制定一个Routing Key（路由key）
消息的发送方在向Exchange发送消息时，也必须指定消息的Routing Key
Exchange不再把消息交给每一个绑定的队列，而是根据消息的Routing Key进行判断，只有队列的Routing Key与消息的Routing Key完全一致，才会接收到消息

流程：

Putting it all together

图解：

P: 生产者，向Exchange发送消息，发送消息时，会指定一个routing key.
X: Exchange（交换机），接收生产者的消息，然后把消息递交给与routing key完全匹配的队列；
C1：消费者，其所在的队列指定了需要routing key为error的消息
C2：消费者，其所在的队列指定了需要routing key为info, error，warning的消息

producer.go

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72


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://admin:admin@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(
		"logs_direct", // name
		"direct",      // type
		true,          // durable
		false,         // auto-deleted
		false,         // internal
		false,         // no-wait
		nil,           // arguments
	)

	failOnError(err, "Failed to declare an exchange")

	body := bodyFrom(os.Args)
	err = ch.Publish(
		"logs_direct",         // exchange
		severityFrom(os.Args), // routing key
		false, // mandatory
		false, // immediate
		amqp.Publishing{
			ContentType: "text/plain",
			Body:        []byte(body),
		})

	failOnError(err, "Failed to publish a message")

	log.Printf(" [x] Sent %s", body)
}

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

func severityFrom(args []string) string  {
	var s string
	if (len(args) < 2) || os.Args[1] == ""{
		s = "info"
	} else {
		s = os.Args[1]
	}
	return s
}

consumer.go

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82


package main

import (
	"log"
	"os"
	"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://admin:admin@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(
		"logs_direct", // name
		"direct",      // type
		true,          // durable
		false,         // auto-deleted
		false,         // internal
		false,         // no-wait
		nil,           // arguments
	)
	failOnError(err, "Failed to declare an exchange")

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

	if len(os.Args) < 2 {
		log.Printf("Usage: %s [info] [warning] [error]", os.Args[0])
		os.Exit(0)
	}
	for _, s := range os.Args[1:] {
		log.Printf("Binding queue %s to exchange %s with routing key %s",
			q.Name, "logs_direct", s)
		err = ch.QueueBind(
			q.Name,        // queue name
			s,             // routing key
			"logs_direct", // exchange
			false,
			nil)
		failOnError(err, "Failed to bind 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")

	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
}

测试

如果我们只想将“warning”和“err”（而不是“info”）级别的日志消息保存到文件中，只需打开控制台并输入：

go run consumer1.go warning error > logs_from_rabbit.log

如果你想在屏幕上查看所有日志消息，请打开一个新终端并执行以下操作：

go run consumer2.go info warning error

例如，我们要发出error日志消息，只需输入：

go run producer.go error "Run. Run. Or it will explode."

我们发出“info”级别的日志消息，输入：

go run producer.go info "Run. Info Test"

创建的logs_direct的交换机：

1630853716172

绑定的两个队列：

1630853808518

消费者1

1630852986604

消费者2

1630853312124

生产者

1630852889585

如同我们所期望的，消费者1和消费者2都能获取到error级别的日志消息，对于info级别的日志消息，只有消费者2能获取到；