RabbitMQ與.net core(五) topic類型 與 headers類型 的Exchange
阿新 • • 發佈:2019-01-10
模糊 collect ref lar create etc ade del system 原文:RabbitMQ與.net core(五) topic類型 與 headers類型 的Exchange
1.topic類型的Exchange
我們之前說過Topic類型的Exchange是direct類型的模糊查詢模式,可以通過routkey來實現模糊消費message,topic的模糊匹配有兩種模式:
1. 使用*來匹配一個單詞
2.使用#來匹配0個或多個單詞
我們來看代碼
消費端
using RabbitMQ.Client; using RabbitMQ.Client.Events; using System; using System.Collections.Generic;using System.Text; using System.Threading; namespace RabbitMQClient { class Program { private static readonly ConnectionFactory rabbitMqFactory = new ConnectionFactory() { HostName = "39.**.**.**", Port = 5672, UserName = "root", Password = "root", VirtualHost = "/" }; static void Main(string[] args) { var exchangeAll = "changeAll"; var queueman = "queueman"; var quemankey = "man.#"; using (IConnection conn = rabbitMqFactory.CreateConnection())using (IModel channel = conn.CreateModel()) { channel.ExchangeDeclare(exchangeAll, type: "topic", durable: true, autoDelete: false); channel.QueueDeclare(queueman, durable: true, exclusive: false, autoDelete: false); channel.QueueBind(queueman, exchangeAll, quemankey); channel.BasicQos(prefetchSize: 0, prefetchCount: 50, global: false); EventingBasicConsumer consumer = new EventingBasicConsumer(channel); consumer.Received += (model, ea) => { Byte[] body = ea.Body; String message = Encoding.UTF8.GetString(body); Console.WriteLine( message); channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false); }; channel.BasicConsume(queue: queueman, autoAck: false, consumer: consumer); Console.ReadLine(); } } } }
生產者代碼
using RabbitMQ.Client; using System; using System.Collections.Generic; using System.Text; using System.Threading; using System.Threading.Tasks; namespace RabbitMQConsole { class Program { static void Main(string[] args) { ConnectionFactory factory = new ConnectionFactory(); factory.HostName = "39.**.**.**"; factory.Port = 5672; factory.VirtualHost = "/"; factory.UserName = "root"; factory.Password = "root"; var exchangeAll = "changeAll"; //性別.姓氏.頭發長度 var keymanA = "man.chen.long"; var keymanB = "man.liu.long"; var keymanC = "woman.liu.long"; var keymanD = "woman.chen.short"; using (var connection = factory.CreateConnection()) { using (var channel = connection.CreateModel()) { channel.ExchangeDeclare(exchangeAll, type: "topic", durable: true, autoDelete: false); var properties = channel.CreateBasicProperties(); properties.Persistent = true; //發布消息 channel.BasicPublish(exchange: exchangeAll, routingKey: keymanA, basicProperties: properties, body: Encoding.UTF8.GetBytes(keymanA)); channel.BasicPublish(exchange: exchangeAll, routingKey: keymanB, basicProperties: properties, body: Encoding.UTF8.GetBytes(keymanB)); channel.BasicPublish(exchange: exchangeAll, routingKey: keymanC, basicProperties: properties, body: Encoding.UTF8.GetBytes(keymanC)); channel.BasicPublish(exchange: exchangeAll, routingKey: keymanD, basicProperties: properties, body: Encoding.UTF8.GetBytes(keymanD)); } } } } }
我們先運行消費端再運行生產段,結果如下
消費端:
2.headers類型的exchange
生成者代碼
using RabbitMQ.Client; using System; using System.Collections.Generic; using System.Text; using System.Threading; using System.Threading.Tasks; namespace RabbitMQConsole { class Program { static void Main(string[] args) { ConnectionFactory factory = new ConnectionFactory(); factory.HostName = "39.**.**.**"; factory.Port = 5672; factory.VirtualHost = "/"; factory.UserName = "root"; factory.Password = "root"; var exchangeAll = "changeHeader"; using (var connection = factory.CreateConnection()) { using (var channel = connection.CreateModel()) { channel.ExchangeDeclare(exchangeAll, type: "headers", durable: true, autoDelete: false); var properties = channel.CreateBasicProperties(); properties.Persistent = true; properties.Headers = new Dictionary<string, object> { { "sex","man"} }; //發布消息 channel.BasicPublish(exchange: exchangeAll, routingKey: "", basicProperties: properties, body: Encoding.UTF8.GetBytes("hihihi")); } } } } }
消費端代碼
using RabbitMQ.Client; using RabbitMQ.Client.Events; using System; using System.Collections.Generic; using System.Text; using System.Threading; namespace RabbitMQClient { class Program { private static readonly ConnectionFactory rabbitMqFactory = new ConnectionFactory() { HostName = "39.**.**.**", Port = 5672, UserName = "root", Password = "root", VirtualHost = "/" }; static void Main(string[] args) { var exchangeAll = "changeHeader"; var queueman = "queueHeader"; using (IConnection conn = rabbitMqFactory.CreateConnection()) using (IModel channel = conn.CreateModel()) { channel.ExchangeDeclare(exchangeAll, type: "headers", durable: true, autoDelete: false); channel.QueueDeclare(queueman, durable: true, exclusive: false, autoDelete: false); channel.QueueBind(queueman, exchangeAll, "",new Dictionary<string, object> { { "sex","man" } }); channel.BasicQos(prefetchSize: 0, prefetchCount: 50, global: false); EventingBasicConsumer consumer = new EventingBasicConsumer(channel); consumer.Received += (model, ea) => { Byte[] body = ea.Body; String message = Encoding.UTF8.GetString(body); Console.WriteLine( message); channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false); }; channel.BasicConsume(queue: queueman, autoAck: false, consumer: consumer); Console.ReadLine(); } } } }
RabbitMQ與.net core(五) topic類型 與 headers類型 的Exchange