首页 > 代码库 > rabbitmq消息队列
rabbitmq消息队列
rabbitmq消息队列
在之前的教程中,我们创建了一个简单的日志系统。我们能够向许多交换器转发日志消息。
在本教程中,我们将添加一个功能——我们让它仅仅接收我们感兴趣的日志类别。举例:我们 实现仅将严重级别的错误日志写入磁盘(为了节省磁盘空间),其余日志级别的日志直接打印到控制台。
绑定
之前的章节中我们已经创建过绑定,你可能还会记得:
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型的路由算法 比较简单——消息会被派发到某个队列,该队列的绑定键恰好和消息的路由键一致。
为了阐述,考虑如下设置:
该设置中,可以看到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型的,我们将日志的严重级别作为路由键,这样的话接收端程序可以选择日志接收级别进行接收,首先聚焦下日志发送端:
首先创建一个交换器:
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
|
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" ) } |
糅合在一起
发送端:
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
|
// 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 } |
接收端:
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
|
// 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 } |
如果您只想保存“警告”和“错误”(而不是“信息”)日志消息到文件,只需要打开一个控制台然后输入:
1
|
go run receive_logs_direct.go warning error > logs_from_rabbit.log |
如果你想看到所有的日志消息在你的屏幕上,打开一个新的终端,输入:
1
|
go run receive_logs_direct.go info warning error |
发出一个错误日志消息类型如下:
1
|
go run emit_log_direct.go error "Run. Run. Or it will explode." |
可以观察到:
消息可以进行分类接收了, 只有error级别的消息才会被存入log日志文件,而info、warning级别的都不存入。
实际效果如下:
rabbitmq消息队列