前两章我们讲了RabbitMQ的direct模式和fanout模式，本章介绍topic主题模式的应用。如果对direct模式下通过routingkey来匹配消息的模式已经有一定了解那fanout也很好理解。简单的可以理解成direct是通过routingkey精准匹配的，而topic是通过routingkey来模糊匹配。
在topic模式下支持两个特殊字符的匹配。* (星号) 代表任意 一个单词# (井号) 0个或者多个单词
注意：上面说的是单词不是字符。
如下图所示，RabbitMQ direct模式通过RoutingKey来精准匹配，RoutingKey为red的投递到Queue1，RoutingKey为black和white的投递到Queue2。
我们可以假设一个场景，我们要做一个日志模块来收集处理不同的日志，日志区分包含三个维度的标准：模块、日志紧急程度、日志重要程度。模块分为：red、black、white；紧急程度分为：critical、normal；把重要程度分为：medium、low、high在RoutingKey字段中我们把这三个维度通过两个“.“连接起来。
现在我们需要对black模块，紧急程度为critical，重要程度为high的日志分配到队列1打印到屏幕；对所以模块重要程度为high的日志和white紧急程度为critical的日志发送到队列2持久化到硬盘。如下示例：
-
RoutingKey为“black.critical.high”的日志会投递到queue1和queue2,。
-
RoutingKey为“red.critical.high”的日志会只投递到queue2。
-
RoutingKey为“white.critical.high”的日志会投递到queue2，并且虽然queue2的两个匹配规则都符合但只会向queue2投递一份。
新建TopicProduct用来发布三种routingkey的消息。
using System;using System.Text;using RabbitMQ.Client;using RabbitMQ.Client.Events;namespace TopicProduct{ class Program { static void Main(string[] args) { String exchangeName = "wytExchangeTopic"; String routeKeyName1 = "black.critical.high"; String routeKeyName2 = "red.critical.high"; String routeKeyName3 = "white.critical.high"; String message1 = "black-critical-high!"; String message2 = "red-critical-high!"; String message3 = "white-critical-high!"; ConnectionFactory factory = new ConnectionFactory(); factory.HostName = "192.168.63.129"; factory.Port = 5672; factory.VirtualHost = "/wyt"; factory.UserName = "wyt"; factory.Password = "wyt"; using (IConnection connection=factory.CreateConnection()) { using (IModel channel=connection.CreateModel()) { channel.ExchangeDeclare(exchange: exchangeName, type: "topic", durable: true, autoDelete: false, arguments: null); IBasicProperties properties = channel.CreateBasicProperties(); properties.Persistent = true; Byte[] body1 = Encoding.UTF8.GetBytes(message1); Byte[] body2 = Encoding.UTF8.GetBytes(message2); Byte[] body3 = Encoding.UTF8.GetBytes(message3); //消息推送 channel.BasicPublish(exchange: exchangeName, routingKey:routeKeyName1,basicProperties: properties, body: body1); channel.BasicPublish(exchange: exchangeName, routingKey: routeKeyName2, basicProperties: properties, body: body2); channel.BasicPublish(exchange: exchangeName, routingKey: routeKeyName3, basicProperties: properties, body: body3); Console.WriteLine(" [x] Sent {0}", message1); Console.WriteLine(" [x] Sent {0}", message2); Console.WriteLine(" [x] Sent {0}", message3); } } Console.WriteLine(" Press [enter] to exit."); Console.ReadLine(); } }}
using System;using System.Text;using RabbitMQ.Client;using RabbitMQ.Client.Events;namespace TopicProduct{ class Program { static void Main(string[] args) { String exchangeName = "wytExchangeTopic"; String routeKeyName1 = "black.critical.high"; String routeKeyName2 = "red.critical.high"; String routeKeyName3 = "white.critical.high"; String message1 = "black-critical-high!"; String message2 = "red-critical-high!"; String message3 = "white-critical-high!"; ConnectionFactory factory = new ConnectionFactory(); factory.HostName = "192.168.63.129"; factory.Port = 5672; factory.VirtualHost = "/wyt"; factory.UserName = "wyt"; factory.Password = "wyt"; using (IConnection connection=factory.CreateConnection()) { using (IModel channel=connection.CreateModel()) { channel.ExchangeDeclare(exchange: exchangeName, type: "topic", durable: true, autoDelete: false, arguments: null); IBasicProperties properties = channel.CreateBasicProperties(); properties.Persistent = true; Byte[] body1 = Encoding.UTF8.GetBytes(message1); Byte[] body2 = Encoding.UTF8.GetBytes(message2); Byte[] body3 = Encoding.UTF8.GetBytes(message3); //消息推送 channel.BasicPublish(exchange: exchangeName, routingKey:routeKeyName1,basicProperties: properties, body: body1); channel.BasicPublish(exchange: exchangeName, routingKey: routeKeyName2, basicProperties: properties, body: body2); channel.BasicPublish(exchange: exchangeName, routingKey: routeKeyName3, basicProperties: properties, body: body3); Console.WriteLine(" [x] Sent {0}", message1); Console.WriteLine(" [x] Sent {0}", message2); Console.WriteLine(" [x] Sent {0}", message3); } } Console.WriteLine(" Press [enter] to exit."); Console.ReadLine(); } }} 新建TopicCustomerA接收一种消息
using System;using System.Text;using RabbitMQ.Client;using RabbitMQ.Client.Events;namespace TopicCustomerA{ class Program { static void Main(string[] args) { String exchangeName = "wytExchangeTopic"; String routeKeyName1 = "black.critical.high"; ConnectionFactory factory = new ConnectionFactory(); factory.HostName = "192.168.63.129"; factory.Port = 5672; factory.VirtualHost = "/wyt"; factory.UserName = "wyt"; factory.Password = "wyt"; using (IConnection connection=factory.CreateConnection()) { using (IModel channel=connection.CreateModel()) { channel.ExchangeDeclare(exchange: exchangeName, type: "topic", durable: true, autoDelete: false, arguments: null); String queueName = channel.QueueDeclare().QueueName; channel.QueueBind(queue: queueName, exchange: exchangeName, routingKey: routeKeyName1, arguments: null); EventingBasicConsumer consumer = new EventingBasicConsumer(channel); consumer.Received += (model, ea) => { var body = ea.Body; var message = Encoding.UTF8.GetString(body); var routingKey = ea.RoutingKey; Console.WriteLine(" [x] Received '{0}':'{1}'", routingKey, message); channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false); }; channel.BasicConsume(queue: queueName, autoAck: false, consumer: consumer); Console.WriteLine(" Press [enter] to exit."); Console.ReadLine(); } } } }}
using System;using System.Text;using RabbitMQ.Client;using RabbitMQ.Client.Events;namespace TopicCustomerA{ class Program { static void Main(string[] args) { String exchangeName = "wytExchangeTopic"; String routeKeyName1 = "black.critical.high"; ConnectionFactory factory = new ConnectionFactory(); factory.HostName = "192.168.63.129"; factory.Port = 5672; factory.VirtualHost = "/wyt"; factory.UserName = "wyt"; factory.Password = "wyt"; using (IConnection connection=factory.CreateConnection()) { using (IModel channel=connection.CreateModel()) { channel.ExchangeDeclare(exchange: exchangeName, type: "topic", durable: true, autoDelete: false, arguments: null); String queueName = channel.QueueDeclare().QueueName; channel.QueueBind(queue: queueName, exchange: exchangeName, routingKey: routeKeyName1, arguments: null); EventingBasicConsumer consumer = new EventingBasicConsumer(channel); consumer.Received += (model, ea) => { var body = ea.Body; var message = Encoding.UTF8.GetString(body); var routingKey = ea.RoutingKey; Console.WriteLine(" [x] Received '{0}':'{1}'", routingKey, message); channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false); }; channel.BasicConsume(queue: queueName, autoAck: false, consumer: consumer); Console.WriteLine(" Press [enter] to exit."); Console.ReadLine(); } } } }} 新建TopicCustomerB接收两种消息
using System;using System.Text;using RabbitMQ.Client;using RabbitMQ.Client.Events;namespace TopicCustomerB{ class Program { static void Main(string[] args) { String exchangeName = "wytExchangeTopic"; String routeKeyName1 = "red.critical.*"; String routeKeyName2 = "white.critical.*"; ConnectionFactory factory = new ConnectionFactory(); factory.HostName = "192.168.63.129"; factory.Port = 5672; factory.VirtualHost = "/wyt"; factory.UserName = "wyt"; factory.Password = "wyt"; using (IConnection connection = factory.CreateConnection()) { using (IModel channel = connection.CreateModel()) { channel.ExchangeDeclare(exchange: exchangeName, type: "topic", durable: true, autoDelete: false, arguments: null); String queueName = channel.QueueDeclare().QueueName; channel.QueueBind(queue: queueName, exchange: exchangeName, routingKey: routeKeyName1, arguments: null); channel.QueueBind(queue: queueName, exchange: exchangeName, routingKey: routeKeyName2, arguments: null); EventingBasicConsumer consumer = new EventingBasicConsumer(channel); consumer.Received += (model, ea) => { var body = ea.Body; var message = Encoding.UTF8.GetString(body); var routingKey = ea.RoutingKey; Console.WriteLine(" [x] Received '{0}':'{1}'", routingKey, message); channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false); }; channel.BasicConsume(queue: queueName, autoAck: false, consumer: consumer); Console.WriteLine(" Press [enter] to exit."); Console.ReadLine(); } } } }}
using System;using System.Text;using RabbitMQ.Client;using RabbitMQ.Client.Events;namespace TopicCustomerB{ class Program { static void Main(string[] args) { String exchangeName = "wytExchangeTopic"; String routeKeyName1 = "red.critical.*"; String routeKeyName2 = "white.critical.*"; ConnectionFactory factory = new ConnectionFactory(); factory.HostName = "192.168.63.129"; factory.Port = 5672; factory.VirtualHost = "/wyt"; factory.UserName = "wyt"; factory.Password = "wyt"; using (IConnection connection = factory.CreateConnection()) { using (IModel channel = connection.CreateModel()) { channel.ExchangeDeclare(exchange: exchangeName, type: "topic", durable: true, autoDelete: false, arguments: null); String queueName = channel.QueueDeclare().QueueName; channel.QueueBind(queue: queueName, exchange: exchangeName, routingKey: routeKeyName1, arguments: null); channel.QueueBind(queue: queueName, exchange: exchangeName, routingKey: routeKeyName2, arguments: null); EventingBasicConsumer consumer = new EventingBasicConsumer(channel); consumer.Received += (model, ea) => { var body = ea.Body; var message = Encoding.UTF8.GetString(body); var routingKey = ea.RoutingKey; Console.WriteLine(" [x] Received '{0}':'{1}'", routingKey, message); channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false); }; channel.BasicConsume(queue: queueName, autoAck: false, consumer: consumer); Console.WriteLine(" Press [enter] to exit."); Console.ReadLine(); } } } }} 先运行TopicCustomerA和TopicCustomerB保持订阅状态。然后执行TopicProduct发布消息。TopicCustomerA和TopicCustomerB收到的消息如下：