在消息队列模型中，如何向所有消费者广播消息称为“发布/订阅”。本文用一个简单的例子，简单描述如何通过fanout switch发布和订阅消息，仅供学习和分享。如有不足之处，请指正。

扇出开关型号

使用广播模式的扇区交换机路由到与绑定交换机对应的所有队列。发送到交换机的每条消息都将被转发到绑定到交换机的所有队列。与子网广播非常相似，每个子网中的主机都会收到重复的消息。扇出交换机是转发消息最快的交换机。

RabbitMQ控制台操作

添加两个队列。

在同一个虚拟主机下添加两个队列Q1和Q2，如下图所示:

绑定扇出开关

将两个队列绑定到系统默认扇出开关，如下所示:

示例渲染

生产者，使用扇出型交换机发布消息，如下图所示:

当生成器发布消息时，队列Q1和Q2都会收到该消息，如下图所示:

当消费者启动时，两个消费者都会订阅相关的消息，如下图所示:

核心代码

消息发布

建立连接后，将通道声明为扇出类型的交换机，如下所示:

1 /// <sum***ry> 2 /// fanout类型交换机，发送消息 3 /// </sum***ry> 4 public class RabbitMqFanoutSendHelper : RabbitMqHelper { 5 /// <sum***ry> 6 /// 发送消息 7 /// </sum***ry> 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 /// <sum***ry> 2 /// 扇形交换机接收消息 3 /// </sum***ry> 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 }