RabbitMQ Headers Exchange示例

时间:2023-03-08 20:44:20

(1)、发布者

             var connectionFactory = new ConnectionFactory() { HostName="192.168.205.128",UserName="guest",Password="guest"};  //创建连接工厂
var connection = connectionFactory.CreateConnection(); //创建connection
var channel = connection.CreateModel(); //创建channel
//声明交换机
//String exchange, 交换机名称
//String type, 交换机类型
//Boolean durable, 是否持久化
//Boolean autoDelete, 是否自动删除
//IDictionary< String, Object > arguments 交换机参数
channel.ExchangeDeclare("headersExchange", ExchangeType.Headers, true, false, null);
//声明队列
//String queue, 队列名称
//Boolean durable, 是否持久化
//Boolean exclusive, 是否专有的(排外)
//Boolean autoDelete, 是否自动删除
//IDictionary<String, Object> arguments 队列参数
channel.QueueDeclare("headersQueue", true, false, false, null);
//将队列绑定到交换机上
//String queue, 队列名称
//String exchange, 交换机名称
//String routingKey, routingKey
//IDictionary< String, Object > arguments 绑定参数
IDictionary<String, Object> arguments = new Dictionary<String, Object>()
{
{"x-match","any"}, //设置x-match属性为any,即头信息中只要有一个匹配就可以了
{ "username","fanqi"}
};
channel.QueueBind("headersQueue", "headersExchange", string.Empty, arguments); //设置消息属性
var properties = channel.CreateBasicProperties();
properties.Headers = new Dictionary<String, Object>();
properties.Headers.Add("username", "fanqi"); //发布消息
//String exchange, 交换机名称
//String routingKey, routingKey
//IBasicProperties basicProperties, 发布属性
//Byte[] body 消息内容
channel.BasicPublish("headersExchange", string.Empty, properties, Encoding.UTF8.GetBytes("来自.net的问候"));
Console.ReadKey();

(2)、接收者

             var connectionFactory = new ConnectionFactory() { HostName = "192.168.205.128", UserName = "guest", Password = "guest" };  //创建连接工厂
var connection = connectionFactory.CreateConnection(); //创建connection
var channel = connection.CreateModel(); //创建channel
//声明交换机
//String exchange, 交换机名称
//String type, 交换机类型
//Boolean durable, 是否持久化
//Boolean autoDelete, 是否自动删除
//IDictionary< String, Object > arguments 交换机参数
channel.ExchangeDeclare("headersExchange", ExchangeType.Headers, true, false, null);
//声明队列
//String queue, 队列名称
//Boolean durable, 是否持久化
//Boolean exclusive, 是否专有的(排外)
//Boolean autoDelete, 是否自动删除
//IDictionary<String, Object> arguments 队列参数
channel.QueueDeclare("headersQueue", true, false, false, null);
//将队列绑定到交换机上
//String queue, 队列名称
//String exchange, 交换机名称
//String routingKey, routingKey
//IDictionary< String, Object > arguments 绑定参数 IDictionary<String, Object> arguments = new Dictionary<String, Object>()
{
{"x-match","any"}, //设置x-match属性为any,即头信息中只要有一个匹配就可以了
{ "username","fanqi"}
};
channel.QueueBind("headersQueue", "headersExchange", string.Empty, arguments);
//1.直接获取消息
//var result = channel.BasicGet("headersQueue", true);
//Console.WriteLine(Encoding.UTF8.GetString(result.Body));
//2.使用事件机制获取消息
EventingBasicConsumer consumer = new EventingBasicConsumer(channel);
consumer.Received += (sender, e) =>
{
Console.WriteLine(Encoding.UTF8.GetString(e.Body));
};
channel.BasicConsume("headersQueue", true, consumer);
Console.ReadKey();