生产者：
复制代码
1 //direct类型 路由模式 1对1匹配
2 //生产者发送消息时需要指定一个路由键(routingKey),交换机只会把消息转发给包含该路由键的队列
3 //string exchange = "TestMq_Exchange"; //交换机
4 //string routingKey = "TestMq_RoutingKey"; //路由键
5
6 string queueName = "TestMq"; //队列名
7 for (int i = 0; i < 10; i++)
8 {
9 string message = "Hello World:" + i;
10 RabbitMqConfig rabbitMqConfig = RabbitMqConfig.Init(); //初始化配置文件
11 ConnectionFactory connFactory = new ConnectionFactory
12 {
13 Uri = new Uri(url),
14 RequestedConnectionTimeout = rabbitMqConfig.RequestedConnectionTimeout,
15 RequestedChannelMax = rabbitMqConfig.RequestedChannelMax,
16 RequestedHeartbeat = rabbitMqConfig.RequestedHeartbeat,
17 AutomaticRecoveryEnabled = false
18 };
19 string exchange = queueName + "_Exchange"; //交换机
20 string routingKey = queueName + "_RoutingKey"; //路由键
21 string exchangeType = ExchangeType.Direct; //类型 direct
22
23 IModel model = this._conn.CreateModel();
24 model.ExchangeDeclare(exchange, exchangeType, true, false, null); //声明交换机
25 model.QueueDeclare(queueName, true, false, false, null); //声明队列
26 model.QueueBind(queueName, exchange, routingKey, null); //绑定
27
28 IBasicProperties basicProperties = model.CreateBasicProperties();
29 basicProperties.Persistent = true; //消息持久化
30 basicProperties.DeliveryMode = 2; //消息持久化, 默认为1 非持久化
31 byte[] bytes = Encoding.UTF8.GetBytes(message);
32 model.BasicPublish(exchange, routingKey, basicProperties, bytes);
33
34 Console.WriteLine(DateTime.Now);
35 }
复制代码
初始化配置文件，在config当中配置
复制代码
1 ConfigurationManager.GetSection("RabbitMqConfig"); //读取配置文件
2
3 //config配置文件
4 <configSections>
5 <section name="RabbitMqConfig" type="Rabbit.Common.RabbitMqConfig,Rabbit.Common" />
6 </configSections>
7 <RabbitMqConfig RequestedHeartbeat="60" RequestedConnectionTimeout="300" RequestedChannelMax="500" Uri="amqp://admin:123456@127.0.0.1:5672//" />
复制代码
消费者1：
复制代码
1 //消费者1
2 string queueName = "TestMq"; //队列名
3 string exchange = queueName + "_Exchange"; //交换器
4 string routingKey = queueName + "_RoutingKey"; //路由关键字
5
6 var rabbitMq = RabbitMqConfig.Init(); //获取Rabbit队列配置
7 var rm = new ConnectionFactory()
8 {
9 Uri = new Uri(rabbitMq.Uri),
10 RequestedConnectionTimeout = rabbitMq.RequestedConnectionTimeout,
11 RequestedChannelMax = rabbitMq.RequestedChannelMax,
12 RequestedHeartbeat = rabbitMq.RequestedHeartbeat,
13 };
14
15 var connection = rm.CreateConnection(); //创建连接
16 var channel = connection.CreateModel(); //创建通道
17 //开启队列持久化(durable = true),不自动删除(autoDelete = false),是否专属(exclusive: false)
18 channel.ExchangeDeclare(exchange, ExchangeType.Direct, true, false, null); //声明一个交换器
19 channel.QueueDeclare(queueName, true, false, false, null); //声明一个队列
20 channel.QueueBind(queueName, exchange, routingKey, null); //绑定交换器和路由
21 channel.BasicQos(0, 1, false); //每次只接收1个，处理完后再接收下一个
22 var consumer = new EventingBasicConsumer(channel);
23
24 consumer.Received += (model, ea) =>
25 {
26 var message = Encoding.UTF8.GetString(ea.Body); //消息主体
27
28 //处理消息逻辑,可以使用异步处理
29 Console.WriteLine(DateTime.Now + " 收到消息：" + message);
30 Thread.Sleep(3000); //模拟消耗延时
31
32 channel.BasicAck(ea.DeliveryTag, false); //返回确认状态 该条消息将会从队列当中移除
33
34 };
35 //监听队列，手动返回完成 第二个参数值为false代表关闭RabbitMQ的自动应答机制，改为手动应答。
36 channel.BasicConsume(queueName, false, consumer);
复制代码
消费者2：
复制代码
//消费者2
string queueName = "TestMq2"; //队列名2
string exchange = "TestMq_Exchange"; //交换器 TestMq_Exchange
string routingKey = "TestMq_RoutingKey"; //路由关键字 TestMq_RoutingKey
。。。
。。。
复制代码
下图所示： 消费者1 完全匹配， 消费者2的队列名称为：TestMq2，与生产者不匹配
2，广播模式 / 分发模式 ( fanout )
这种模式下,消息会被所有消费者消费.也就是说,只要是"绑定"到某个交换机的队列,都会收到生产者发送到该交换机的消息.
fanout 类型的发送规则非常简单，它会把所有发送到该交换机Exchange的消息路由到所有与它绑定的Queue中；
也就是说：在fanout模式下，只跟 交换机Exchange有关系，跟路由key无关。
复制代码
1 string queueName = "TestMq";
2 string exchange = "TestMq_Exchange"; //只需要声明交换机 即可
3 string routingKey = ""; //路由键
4 string exchangeType = ExchangeType.Fanout; //类型 Fanout
5 for (int i = 0; i < 10; i++)
6 {
7 string message = "Hello World:" + i;
8 RabbitMqConfig rabbitMqConfig = RabbitMqConfig.Init(); //初始化配置文件
9 ConnectionFactory connFactory = new ConnectionFactory
10 {
11 Uri = new Uri(url),
12 RequestedConnectionTimeout = rabbitMqConfig.RequestedConnectionTimeout,
13 RequestedChannelMax = rabbitMqConfig.RequestedChannelMax,
14 RequestedHeartbeat = rabbitMqConfig.RequestedHeartbeat,
15 AutomaticRecoveryEnabled = false
16 };
17
18 IModel model = this._conn.CreateModel();
19 model.ExchangeDeclare(exchange, exchangeType, true, false, null); //声明交换机
20 model.QueueDeclare(queueName, true, false, false, null); //声明队列
21 model.QueueBind(queueName, exchange, routingKey, null); //绑定
22
23 IBasicProperties basicProperties = model.CreateBasicProperties();
24 basicProperties.Persistent = true; //消息持久化
25 basicProperties.DeliveryMode = 2; //消息持久化, 默认为1 非持久化
26 byte[] bytes = Encoding.UTF8.GetBytes(message);
27 model.BasicPublish(exchange, routingKey, basicProperties, bytes);
28
29 Console.WriteLine(DateTime.Now + ":" + message);
30 }
复制代码
消费者1： 这里设置： queueName = TestMq
复制代码
string queueName = "TestMq"; //队列名
string exchange = "TestMq_Exchange"; //交换机
string routingKey = ""; //路由键
....
....
channel.ExchangeDeclare(exchange, ExchangeType.Fanout, true, false, null); //声明一个交换器
复制代码
消费者2： 这里设置： queueName = TestMq
复制代码
1 string queueName = "TestMq"; //队列名
2 string exchange = "TestMq_Exchange"; //交换机
3 string routingKey = ""; //路由键
4
5 ....
6 ....
7
8 channel.ExchangeDeclare(exchange, ExchangeType.Fanout, true, false, null); //声明一个交换器
复制代码
上边示例当中使用相同的 queueName = "TestMq"; 会出现如下图所示的情况，各消费一半的情况，因为系统认为是一个队列
下边我们修改一下队列名称：
1 //消费者1
2 string queueName = "TestMq1"; //队列名1
3 string exchange = "TestMq_Exchange"; //交换机
4 string routingKey = ""; //路由键
1 //消费者2
2 string queueName = "TestMq2"; //队列名2
3 string exchange = "TestMq_Exchange"; //交换机
4 string routingKey = ""; //路由键
亚马逊测评 www.yisuping.com
