rabbitmq消息队列——"路由"

在之前的教程中,我们创建了一个简单的日志系统。我们能够向许多交换器转发日志消息。

在本教程中,我们将添加一个功能——我们让它仅仅接收我们感兴趣的日志类别。举例:我们 实现仅将严重级别的错误日志写入磁盘(为了节省磁盘空间),其余日志级别的日志直接打印到控制台。

绑定

之前的章节中我们已经创建过绑定,你可能还会记得:

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

绑定是用来维系交换器和队列关系的,这可以被简单地理解为:队列仅仅对从交换器中传的消息感兴趣。

绑定有个额外参数叫做routing_key,为了避免与Channel.Publish方法中的参数相混淆,我们称之为binding key(绑定键)。使用绑定键创建绑定如下:

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

绑定键的含义取决于交换器的类型。我们之前使用的fanout类型的交换器,就会直接忽略这个参数。

Direct型交换器

我们之前的教程中的日志系统是广播所有的消息到所有消费者。我们希望以此拓展来实现根据消息严重性来过滤消息。比如我们希望 写日志到硬盘的代码仅仅接收严重级别的,不要浪费磁盘存储在warning或者info级别的日志。

之前使用的是fanout类型交换器,没有更好的拓展性或者说灵活性——它只能盲目的广播。

现在 使用direct型交换器替代。Direct型的路由算法 比较简单——消息会被派发到某个队列,该队列的绑定键恰好和消息的路由键一致。

为了阐述,考虑如下设置:

该设置中,可以看到direct型的交换器X被绑定到了两个队列:Q1、Q2。Q1使用绑定键orange绑定,Q2包含两个绑定键:black和green。

基于如上设置的话,使用路由键orange发布的消息会被路由到Q1队列,而使用black或者green路由键的消息均会被路由到Q2,所有其余消息将被丢弃。

备注:这里的交换器X和队列的绑定是多对多的关系,也就是说一个交换器可以到绑定多个队列,一个队列也可以被多个交换器绑定,消息只会被路由一次,不能因为两个绑定键都匹配上了路由键消息就会被路由两次,这种是不存在的。

多个绑定

用相同的绑定键去绑定多个队列是完全合法的,我们可以再添加一个black绑定键来绑定X和Q1,这样Q1和Q2都使用black绑定到了交换器X,这其实和fanout类型的交换器直接绑定到队列Q1、Q2功能相同:使用black路由键的消息会被直接路由到Q1和Q2。

发送日志

我们将使用该模型来构建日志系统。使用direct型的交换器替换fanout型的,我们将日志的严重级别作为路由键,这样的话接收端程序可以选择日志接收级别进行接收,首先聚焦下日志发送端:

首先创建一个交换器:

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

然后是发送消息:

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'。

订阅

接收还和之前章节接收一样,只有一个例外:我们将为每一个感兴趣的严重级别创建一个绑定:

q, err := ch.QueueDeclare(
  "",    // name
  false, // durable
  false, // delete when usused
  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")
}

糅合在一起

发送端:

// rabbitmq_4_emit_log_direct.go project main.go
package main

import (
	"fmt"
	"log"
	"os"
	"strings"

	"github.com/streadway/amqp"
)

func failOnError(err error, msg string) {
	if err != nil {
		log.Fatalf("%s: %s", msg, err)
		panic(fmt.Sprintf("%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()

	//声明一个channel
	ch, err := conn.Channel()
	failOnError(err, "Failed to open a channel")
	defer ch.Close()

	//声明一个direct类型交换器
	err = ch.ExchangeDeclare("logs_direct", "direct", true, false, false, false, nil)
	failOnError(err, "Failed to declare an exchange")

	body := bodyFrom(os.Args)
	ch.Publish("logs_direct", severityFrom(os.Args), false, false, 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 || args[1] == "" {
		s = "info"
	} else {
		s = args[1]
	}
	return s
}

接收端:

// rabbitmq_4_receive_logs_direct.go project main.go
package main

import (
	"fmt"
	"log"
	"os"

	"github.com/streadway/amqp"
)

func failOnError(err error, msg string) {
	if err != nil {
		log.Fatalf("%s: %s", msg, err)
		panic(fmt.Sprintf("%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()

	//声明一个channel
	ch, err := conn.Channel()
	failOnError(err, "Failed to open a channel")
	defer ch.Close()

	//声明一个direct类型交换器
	err = ch.ExchangeDeclare("logs_direct", "direct", true, false, false, false, nil)
	failOnError(err, "Failed to declare an exchange")

	//声明一个队列
	q, err := ch.QueueDeclare("", false, false, true, false, nil)
	failOnError(err, "Failed to declare a queue")

	//判断cmd窗口接收参数是否足够
	if len(os.Args) < 2 {
		log.Printf("Usage:%s [info] [warning] [error]", os.Args[0])
		os.Exit(0)
	}

	//cmd窗口输入的多个日志级别,分别循环处理—进行绑定
	for _, s := range os.Args[1:] {
		log.Printf("Binding queue %s to exchange %s with routing key %s", q.Name, "logs_direct", s)
		ch.QueueBind(q.Name, s, "logs_direct", false, nil)
		failOnError(err, "Failed to bind a queue")
	}

	msgs, err := ch.Consume(q.Name, "", true, false, false, false, nil)
	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
}

如果您只想保存“警告”和“错误”(而不是“信息”)日志消息到文件,只需要打开一个控制台然后输入:

go run receive_logs_direct.go warning error > logs_from_rabbit.log

如果你想看到所有的日志消息在你的屏幕上,打开一个新的终端,输入:

go run receive_logs_direct.go info warning error

发出一个错误日志消息类型如下:

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

可以观察到:

消息可以进行分类接收了, 只有error级别的消息才会被存入log日志文件,而info、warning级别的都不存入。

实际效果如下:

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏鬼谷君

kvm解决1000M网卡问题

3893
来自专栏码洞

HTTP协议冷知识大全

HTTP协议是纯文本协议,没有任何加密措施。通过HTTP协议传输的数据都可以在网络上被完全监听。如果用户登陆时将用户名和密码直接明文通过HTTP协议传输过去了,...

842
来自专栏三杯水

系统重启后nginx reload不生效原因分析

这是一种比较少见,困扰我很久的问题,虽然这个问题很简单,但是找到根本原因还是费了不少时间,现在把分析过程分享如下。

2142
来自专栏小车博客

【开源】采集1024的图片到WordPress,图床只支持chevereto

1773
来自专栏杨建荣的学习笔记

zabbix中配置dg的监控(r6笔记第62天)

在zabbix中有了orabbix的辅助,监控效率大大提高,但是因为orabbix是基于jdbc的方式,有些监控还是有一些限制。 比如dataguard的检查,...

2834
来自专栏FreeBuf

如何在安卓移动终端进行一次渗透实验?

下面的实验将通过使用一个虚拟Android系统作为实验主体来进行展示。在这个实验中,通过使用虚拟Android系统,模拟移动终端的操作,来帮助我们逐步实现使用一...

2586
来自专栏网络

CobaltStrike团体服务器部署并后台运行

大家好,我是你们的老朋友Alex。Cobaltstrike的部署安装很简单,但是在实际使用中出现了问题。我把团队服务器放在ECS上,出现了两个问题:1.客户端无...

2426
来自专栏java学习

maven常用命令集合(收藏大全)

如果你是初学者,或者是自学者!你可以加小编微信(xxf960326)!小编可以给你学习上,工作上的一些建议以及可以给你(免费)提供学习资料!最重要我们还可以交个...

681
来自专栏木子昭的博客

nodejs爬虫获取漫威超级英雄电影海报

1453
来自专栏丑胖侠

maven配置文件不一致导致构建失败

问题 在使用maven构建项目时出现一个神奇的问题。通过Intellij IDEA的Maven Projects中进行构建,没问题,能够成功打包项目。 可是...

1937

扫码关注云+社区