c#消息队列实际应用 消息队列和消息中间件的关系
在消息队列模型中，如何向所有消费者广播消息称为“发布/订阅”。本文用一个简单的例子，简单描述如何通过fanout switch发布和订阅消息，仅供学习和分享。如有不足之处，请指正。扇
在消息队列模型中，如何向所有消费者广播消息称为“发布/订阅”。本文用一个简单的例子，简单描述如何通过fanout switch发布和订阅消息，仅供学习和分享。如有不足之处，请指正。
扇出开关型号使用广播模式的扇区交换机路由到与绑定交换机对应的所有队列。发送到交换机的每条消息都将被转发到绑定到交换机的所有队列。与子网广播非常相似，每个子网中的主机都会收到重复的消息。扇出交换机是转发消息最快的交换机。
RabbitMQ控制台操作
添加两个队列。在同一个虚拟主机下添加两个队列Q1和Q2，如下图所示:
绑定扇出开关
将两个队列绑定到系统默认扇出开关，如下所示:
示例渲染
生产者，使用扇出型交换机发布消息，如下图所示:
当生成器发布消息时，队列Q1和Q2都会收到该消息，如下图所示:
当消费者启动时，两个消费者都会订阅相关的消息，如下图所示:
核心代码
消息发布建立连接后，将通道声明为扇出类型的交换机，如下所示:
1 /// <summary> 2 /// fanout类型交换机，发送消息 3 /// </summary> 4 public class RabbitMqFanoutSendHelper : RabbitMqHelper { 5 /// <summary> 6 /// 发送消息 7 /// </summary> 8 /// <param name="msg"></param> 9 /// <returns></returns>10 public bool SendMsg(string msg)11 {12 try13 {14 using (var conn = GetConnection("/Alan.hsiang"))15 {16 using (var channel = conn.CreateModel())17 {18 channel.ExchangeDeclare(exchange: "amq.fanout", type: ExchangeType.Fanout,durable:true);19 20 var body = Encoding.UTF8.GetBytes(msg);21 22 channel.BasicPublish(exchange: "amq.fanout",23 routingKey: "",24 basicProperties: null,25 body: body);26 27 //Console.WriteLine(" [x] Sent {0}", message);28 };29 };30 return true;31 }32 catch (Exception ex)33 {34 throw ex;35 }36 }37 }
消息订阅1///<总结& gt2 ///扇出型开关，发送报文3//
连接建立后，通道用Fanout类型声明交换机，并将队列绑定到subscribe，如下所示:
1 /// <summary> 2 /// 扇形交换机接收消息 3 /// </summary> 4 public class RabbitMqFanoutReceiveHelper : RabbitMqHelper 5 { 6 public RabbitMqReceiveEventHandler OnReceiveEvent; 7 8 private IConnection conn; 9 10 private IModel channel;11 12 private EventingBasicConsumer consumer;13 14 public bool StartReceiveMsg(string queueName)15 {16 try17 {18 conn = GetConnection("/Alan.hsiang");19 20 channel = conn.CreateModel();21 channel.ExchangeDeclare(exchange: "amq.fanout", type: ExchangeType.Fanout,durable:true);22 //此处随机取出交换机下的队列23 //var queueName = channel.QueueDeclare().QueueName;24 channel.QueueBind(queue: queueName, exchange: "amq.fanout", routingKey: "");25 consumer = new EventingBasicConsumer(channel);26 consumer.Received += (model, ea) =>27 {28 var body = ea.Body.ToArray();29 var message = Encoding.UTF8.GetString(body);30 //Console.WriteLine(" [x] Received {0}", message);31 if (OnReceiveEvent != null)32 {33 OnReceiveEvent(queueName+"::"+message);34 }35 };36 channel.BasicConsume(queue: queueName,37 autoAck: true,38 consumer: consumer);39 return true;40 }41 catch (Exception ex)42 {43 throw ex;44 }45 }46 }
