我们构建了一个简单的日志记录系统。我们能够向许多接收者广播日志消息。
在本教程中，我们将向它添加一个特性-我们将使它能够只订阅消息的一个子集。例如，我们将只能将关键错误消息定向到日志文件（以节省磁盘空间），同时仍然能够在控制台上打印所有日志消息。

绑定

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

  1. err = ch.QueueBind(
  2.   q.Name, // queue name
  3.   "",     // routing key
  4.   "logs", // exchange
  5.   false,
  6.   nil)

绑定是交换器和队列之间的关系。这可以简单地理解为：队列对来自此交换器的消息感兴趣。
绑定可以采用额外的routing_key参数。为了避免与Channel.Publish参数混淆，我们将其称为binding key。这是我们如何使用键创建绑定的方法：

  1. err = ch.QueueBind(
  2.   q.Name,    // queue name
  3.   "black",   // routing key
  4.   "logs",    // exchange
  5.   false,
  6.   nil)

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

直连交换器

我们上一个教程中的日志系统将所有消息广播给所有消费者。我们希望扩展这一点，允许根据消息的严重性过滤消息。例如，我们可能希望将日志消息写入磁盘的脚本只接收严重错误，而不会在warning或info日志消息上浪费磁盘空间。
我们使用fanout交换器，这并没有给我们很大的灵活性——它只能进行无脑广播。
我们将使用direct交换器。direct交换器背后的路由算法很简单——消息进入其binding key与消息的routing key完全匹配的队列。
为了说明这一点，请考虑以下设置：
路由 - 图1
在此设置中，我们可以看到绑定了两个队列的direct交换器X。第一个队列绑定键为orange，第二个队列绑定为两个，一个绑定键为black，另一个为green。
在这种设置中，使用orange路由键发布到交换器的消息将被路由到队列Q1。路由键为black或green的消息将转到Q2。所有其他消息将被丢弃。

多重绑定

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

发送日志

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

  1. err = ch.ExchangeDeclare(
  2.   "logs_direct", // name
  3.   "direct",      // type
  4.   true,          // durable
  5.   false,         // auto-deleted
  6.   false,         // internal
  7.   false,         // no-wait
  8.   nil,           // arguments
  9. )

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

  1. err = ch.ExchangeDeclare(
  2.   "logs_direct", // name
  3.   "direct",      // type
  4.   true,          // durable
  5.   false,         // auto-deleted
  6.   false,         // internal
  7.   false,         // no-wait
  8.   nil,           // arguments
  9. )
  10. failOnError(err, "Failed to declare an exchange")
  11. body := bodyFrom(os.Args)
  12. err = ch.Publish(
  13.   "logs_direct",         // exchange
  14.   severityFrom(os.Args), // routing key
  15.   false, // mandatory
  16.   false, // immediate
  17.   amqp.Publishing{
  18.     ContentType: "text/plain",
  19.     Body:        []byte(body),
  20. })

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

订阅

接收消息的工作方式与上一教程一样，但有一个例外——我们将为感兴趣的每种严重性（日志级别）创建一个新的绑定。

  1. q, err := ch.QueueDeclare(
  2.   "",    // name
  3.   false, // durable
  4.   false, // delete when unused
  5.   true,  // exclusive
  6.   false, // no-wait
  7.   nil,   // arguments
  8. )
  9. failOnError(err, "Failed to declare a queue")
  10. if len(os.Args) < 2 {
  11.   log.Printf("Usage: %s [info] [warning] [error]", os.Args[0])
  12.   os.Exit(0)
  13. }
  14. // 建立多个绑定关系
  15. for _, s := range os.Args[1:] {
  16.   log.Printf("Binding queue %s to exchange %s with routing key %s",
  17.      q.Name, "logs_direct", s)
  18.   err = ch.QueueBind(
  19.     q.Name,        // queue name
  20.     s,             // routing key
  21.     "logs_direct", // exchange
  22.     false,
  23.     nil)
  24.   failOnError(err, "Failed to bind a queue")
  25. }

完整示例

路由 - 图3
emit_log_direct.go脚本的代码：

  1. package main
  2. import (
  3.         "log"
  4.         "os"
  5.         "strings"
  6.         "github.com/streadway/amqp"
  7. )
  8. func failOnError(err error, msg string) {
  9.         if err != nil {
  10.                 log.Fatalf("%s: %s", msg, err)
  11.         }
  12. }
  13. func main() {
  14.         conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
  15.         failOnError(err, "Failed to connect to RabbitMQ")
  16.         defer conn.Close()
  17.         ch, err := conn.Channel()
  18.         failOnError(err, "Failed to open a channel")
  19.         defer ch.Close()
  20.         err = ch.ExchangeDeclare(
  21.                 "logs_direct", // name
  22.                 "direct",      // type
  23.                 true,          // durable
  24.                 false,         // auto-deleted
  25.                 false,         // internal
  26.                 false,         // no-wait
  27.                 nil,           // arguments
  28.         )
  29.         failOnError(err, "Failed to declare an exchange")
  30.         body := bodyFrom(os.Args)
  31.         err = ch.Publish(
  32.                 "logs_direct",         // exchange
  33.                 severityFrom(os.Args), // routing key
  34.                 false, // mandatory
  35.                 false, // immediate
  36.                 amqp.Publishing{
  37.                         ContentType: "text/plain",
  38.                         Body:        []byte(body),
  39.                 })
  40.         failOnError(err, "Failed to publish a message")
  41.         log.Printf(" [x] Sent %s", body)
  42. }
  43. func bodyFrom(args []string) string {
  44.         var s string
  45.         if (len(args) < 3) || os.Args[2] == "" {
  46.                 s = "hello"
  47.         } else {
  48.                 s = strings.Join(args[2:], " ")
  49.         }
  50.         return s
  51. }
  52. func severityFrom(args []string) string {
  53.         var s string
  54.         if (len(args) < 2) || os.Args[1] == "" {
  55.                 s = "info"
  56.         } else {
  57.                 s = os.Args[1]
  58.         }
  59.         return s
  60. }

receive_logs_direct.go的代码：

  1. package main
  2. import (
  3.         "log"
  4.         "os"
  5.         "github.com/streadway/amqp"
  6. )
  7. func failOnError(err error, msg string) {
  8.         if err != nil {
  9.                 log.Fatalf("%s: %s", msg, err)
  10.         }
  11. }
  12. func main() {
  13.         conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
  14.         failOnError(err, "Failed to connect to RabbitMQ")
  15.         defer conn.Close()
  16.         ch, err := conn.Channel()
  17.         failOnError(err, "Failed to open a channel")
  18.         defer ch.Close()
  19.         err = ch.ExchangeDeclare(
  20.                 "logs_direct", // name
  21.                 "direct",      // type
  22.                 true,          // durable
  23.                 false,         // auto-deleted
  24.                 false,         // internal
  25.                 false,         // no-wait
  26.                 nil,           // arguments
  27.         )
  28.         failOnError(err, "Failed to declare an exchange")
  29.         q, err := ch.QueueDeclare(
  30.                 "",    // name
  31.                 false, // durable
  32.                 false, // delete when unused
  33.                 true,  // exclusive
  34.                 false, // no-wait
  35.                 nil,   // arguments
  36.         )
  37.         failOnError(err, "Failed to declare a queue")
  38.         if len(os.Args) < 2 {
  39.                 log.Printf("Usage: %s [info] [warning] [error]", os.Args[0])
  40.                 os.Exit(0)
  41.         }
  42.         for _, s := range os.Args[1:] {
  43.                 log.Printf("Binding queue %s to exchange %s with routing key %s",
  44.                         q.Name, "logs_direct", s)
  45.                 err = ch.QueueBind(
  46.                         q.Name,        // queue name
  47.                         s,             // routing key
  48.                         "logs_direct", // exchange
  49.                         false,
  50.                         nil)
  51.                 failOnError(err, "Failed to bind a queue")
  52.         }
  53.         msgs, err := ch.Consume(
  54.                 q.Name, // queue
  55.                 "",     // consumer
  56.                 true,   // auto ack
  57.                 false,  // exclusive
  58.                 false,  // no local
  59.                 false,  // no wait
  60.                 nil,    // args
  61.         )
  62.         failOnError(err, "Failed to register a consumer")
  63.         forever := make(chan bool)
  64.         go func() {
  65.                 for d := range msgs {
  66.                         log.Printf(" [x] %s", d.Body)
  67.                 }
  68.         }()
  69.         log.Printf(" [*] Waiting for logs. To exit press CTRL+C")
  70.         <-forever
  71. }

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

  1. go run receive_logs_direct.go warning error > logs_from_rabbit.log

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

  1. go run receive_logs_direct.go info warning error
  2. # => [*] Waiting for logs. To exit press CTRL+C

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

  1. go run emit_log_direct.go error "Run. Run. Or it will explode."
  2. # => [x] Sent 'error':'Run. Run. Or it will explode.'

（这里是（emit_log_direct.go)和（receive_logs_direct.go）的完整源码）