RabbitMQ学习笔记
第一部分
1. RabbitMQ简介
RabbitMq 是由 erlang开发的AMQP(高级消息队列协议)的开源实现。RAbbit MQ作为一个消息代理，主要负责接收、存储和转发消息。它提供了可靠的消息机制和灵活的消息路由，并支持消息集群和分布式部署，常用于应用解耦，耗时任务队列，流量削峰等场景。
上图是RabbitMQ的一个基本结构，生产者Producer和消费者Consumer都是RabbitMQ的客户端，Producer负责发送消息，Consumer负责接收消息。
名词概念：
Broker(Server): RabbitMq 服务器，接受客户端连接，实现AMQP消息队列和路由功能的进程。
Virtual Host: 一个虚拟概念，一个Virtual Host 可以有多个Exchange和Queue ，主要用于权限控制，隔离应用。
Exchange: 接受生产者发送的消息，并根据Binding规则将消息路由给服务器中的队列。ExchangType决定了Exchange路由消息的行为，ExchangeType 有 Fanout,Direct,Topic,Header 四种。
Queue: 消息队列，用于存储还未被消费的消息，队列是先进先出，默认是先存储的消息被消费。
Message: 消息，由Header和Body组成，Header是生产者添加的各种属性的集合。包括Message是否被持久化，由哪个MessageQueue 接收，优先级等信息，Body是真正传输的数据，内容格式是Byte[].
Connection: 连接，就是位于客户端和Broker之间的一个TCP连接。
Channel: 信道，仅仅创建Connection 之后客户端是不能发送数据的。需要在Connnection基础上创建Channel,AMQP规定只有通过Channel才能执行AMQP的命令。一个Connection可以包含多个Channel,之所以需要Channel是因为TCP连接的建立和释放都是十分昂贵的。
2. RabbitMQ的安装
-
Windows 平台安装
-
下载 Erlang 和 RabbitMQ
-
将 Erlang 和 RabbitMQ 解压之后放在同一目录下 如：
- C:\Program Files\erl10.5
- C:\Program Files\erl_rabbitmq_server-3.8.3
-
配置环境变量
- ERLANG_HOME C:\Program Files\erl10.5
- RABBITMQ_SERVER C:\Program Files\erl_rabbitmq_server-3.8.3
- %ERLANG_HOME%\bin
- %RABBITMQ_SERVER%\sbin
-
查看环境变量
-
在执行配置环境变量之后服务器此时可能还没有加载path中新加入的路径
-
两种方法可供选择
-
重启电脑
-
查看Path
-
打开DOS
输入 set path
-
查看显示的内容是否有新加入的path
-
如果没有，在DOS 中输入
set ERLANG_HOME
-
此时会显示该节点下的路径，再次输入 set path,查看是否有新加入的路径
-
如果没有，重启服务器
-
-
-
-
Rabbit MQ 服务
-
安装服务
rabbitmq-service install
-
启用服务
rabbitmq-service enable
-
启动服务
rabbitmq-service start
-
-
拷贝文件
-
将 C:\Users\当前用户.erlang.cookie 文件拷贝到以下路径
C:\Program Files\erl_rabbitmq_server-3.8.3\sbin
C:\Windows\System32\config\systemprofile
-
执行 net stop RabbitMQ && net start RabbitMQ 重启服务
-
-
命令查看
- rabbitmqctl status 查询状态
- rabbitmqctl list_users 用户列表
- rabbitmqctl add_user ljs 123
- rabbitmqctl set_permissions ljs "." "." ".*" 赋予ljs 所有的读写队列权限
- rabbitmqctl set_user_tags ljs administrator 分配用户组
-
疑难杂症
-
打不开 localhost:15672
执行 rabbitmq-plugins enable rabbitmq_management
-
远程打开 15672
找到这个文件rabbit.app
C:\Program Files\RabbitMQ Server\rabbitmq_server-3.7.13\ebin\rabbit.app
将第39行：{loopback_users, [<<”guest”>>]}，
改为：{loopback_users, []}，然后命令行输入：net stop RabbitMQ && net start RabbitMQ重启服务
原因：rabbitmq从3.3.0开始禁止使用guest/guest权限通过除localhost外的访问
-
-
-
Liunx 下安装
敬请期待！！！
3. C# 驱动RabbitMQ
nuget RabbitMQ.Client
1. 代码示例
生产者代码
class Program
{
static void Main(string[] args)
{
var factory = new ConnectionFactory()
{
//rabbitmq-server所在设备ip，这里就是本机
HostName = "127.0.0.1",
UserName = "wyy",//用户名
Password = "123321"//密码
};
//第一步：创建连接connection
using (var connection = factory.CreateConnection())
{
//第二步：创建通道channel
using (var channel = connection.CreateModel())
{
//第三步：声明交换机exchang
channel.ExchangeDeclare(exchange: "myexchange",
type: ExchangeType.Direct,
durable: true,
autoDelete: false,
arguments: null);
//第四步：声明队列queue
channel.QueueDeclare(queue: "myqueue",
durable: true,
exclusive: false,
autoDelete: false,
arguments: null);
Console.WriteLine("生产者准备就绪....");
//第五步：绑定队列到交互机
channel.QueueBind(queue:"myqueue", exchange:"myexchange", routingKey:"mykey");
string message = "";
//第六步：发送消息
//在控制台输入消息，按enter键发送消息
while (!message.Equals("quit", StringComparison.CurrentCultureIgnoreCase))
{
message = Console.ReadLine();
var body = Encoding.UTF8.GetBytes(message);
//基本发布
channel.BasicPublish(exchange: "myexchange",
routingKey: "mykey",
basicProperties: null,
body: body);
Console.WriteLine($"消息【{message}】已发送到队列");
}
}
}
Console.ReadKey();
}
}
消费者代码
class Program
{
static void Main(string[] args)
{
var factory = new ConnectionFactory()
{
//rabbitmq-server所在设备ip，这里就是本机
HostName = "127.0.0.1",
UserName = "wyy",//用户名
Password = "123321"//密码
};
//第一步：创建连接connection
using (var connection = factory.CreateConnection())
{
//第二步：创建信道channel
using (var channel = connection.CreateModel())
{
//第三步：声明队列queue
channel.QueueDeclare(queue: "myqueue",
durable: true,
exclusive: false,
autoDelete: false,
arguments: null);
//第四步：定义消费者
var consumer = new EventingBasicConsumer(channel);
consumer.Received += (model, ea) =>
{
var body = ea.Body;
var message = Encoding.UTF8.GetString(body);
Console.WriteLine($"接受到消息【{message}】");
};
Console.WriteLine("消费者准备就绪....");
//第五步：处理消息
channel.BasicConsume(queue: "myqueue",
autoAck: true,
consumer: consumer);
Console.ReadLine();
}
}
}
}
依次运行Producer和Consumer两个应用程序
2. QueueDeclare方法详解
QueueDeclare(string queue, bool durable, bool exclusive, bool autoDelete, IDictionary<string, object> arguments)
//声明队列newsQueue
channel.QueueDeclare(queue: "myqueue",
durable: false,
exclusive: false,
autoDelete: false,
arguments: new Dictionary<string, object>() {
//队列中消息的过期时间是8s
{ "x-message-ttl",1000*8 },
//队列60s没有被使用，则删除该队列
{"x-expires",1000*60 },
//队列最多保存100条消息
{"x-max-length",100 },
//队列中ready类型消息总共不能超过1000字节
{"x-max-length-bytes",1000 },
//当队列消息满了时,丢弃传来后续消息
{"x-overflow","reject-publish" },
//丢弃的消息发送到deadExchange交换机
{"x-dead-letter-exchange","deadExchange" },
//丢弃的消息发送到deadExchange交换机时的RoutingKey
{"x-dead-letter-routing-key","deadKey" },
//队列中最大的优先级等级为10（在Publish消息时对每条消息设置优先级）
{"x-max-priority",10 },
//设置队列默认为lazy
{"x-queue-mode","lazy" }
});
QueueDeclare 方法的参数如下
queue:队列名
durable:是否持久化。设为true时，队列信息保存在rabbitmq的内置数据库中，服务器重启时队列也会恢复。
exclusive:是否排外。设置为true时只有首次声明该队列的Connection可以访问，其他的Connection不能访问该队列；且在Connection断开时，该队列会被删除(即使durable设置为true也会被删除)
autoDelete: 是否自动删除。设置为true时，表示在最后一条使用该队列的连接(Connection)断开时，将自动删除这个队列
arguments:设置队列的一些其他的属性，为Dictionary<string,object>类型，以下是arguments的属性总结
参数名 作用 示例 Message TTL 设置队列中小时的有效时间 {"x-message-ttl",1000*60}
设置队列中所有消息的有效时间为60sAuto expire 自动删除队列。一定的时间内队列没有被消费，则自动删除队列 {"x-expires",1000*60}
设置队列的过期时长60s,如果60s没有对列被访问，则删除队列Max Length 对列能保存消息的最大条数 {"x-max-length",100}
设置队列最多保存100条消息Max Length bytes 队列中ready类型的消息的总字节数 {"x-max-length-bytes",1000}
设置队列中ready类型消息总共不能超过1000字节Overflow behaviour 当队列消息满时，再接收消息时的处理方法。有两种处理方案：默认“drop-head” 模式，表示从队列头部丢弃消息；“reject-publish”表示不接收后续的消息 {"x-overflow","reject-publish"},设置当队列消息满了时，丢弃后续的消息 Dead letter exchange 用于存储被丢弃的消息的交换机名。Overflow behaviour 的两种处理方案中丢弃的消息都会发送到这个交换机 {"x-dead-letter-exchange","beiyongExchange"},设置丢弃的消息发送到名字为beiyongExchange的交换机 Dead letter Routing key 被丢弃的消息发送到 Dead letter exchange时使用的routing key {“x-dead-letter-routing-key”,"deadkey"}
设置丢弃的消息发送到beiyongExchange交换机时的RoutingKey 值是 deadkeyMaximum priority 设置队列中消息优先级的最大等级，在publish时可以设置单条消息的优先级等级 {"x-max-prioity",10}
设置队列中消息优先级的最大等级是10Lazy mode 设置队列的模式。如果设置为Lazy,表示队列中的消息尽可能放在磁盘中，以减少内存的占用；不设置时消息都存放在队列中，用以尽可能快的处理消息 {"x-queue-mode","lazy"}
3.6版本以后可用，设置队列中的消息尽可能存在磁盘中，减少内存的占用。在消息拥堵时和消息持久化配置使用可以减少内存的占用
3.ExchangeDeclare方法详解
void ExchangeDeclare(string exchange, string type, bool durable, bool autoDelete, IDictionary<string, object> arguments)
channel.ExchangeDeclare(exchange: "myexchange",
type: ExchangeType.Direct,
durable: true,
autoDelete: false,
arguments: new Dictionary<string, object> { {"alternate-exchange","BeiyongExchange" }//如果消息不能路由到该交换机，就把消息路由到备用交换机BeiyongExchange上 });
exchange：交换机名字。
type：交换机类型。exchange有direct、fanout、topic、header四种类型，在下一篇会详细介绍；
durable：是否持久化。设置为true时，交换机信息保存在rabbitmq的内置数据库中，服务器重启时交换机信息也会恢复；
autoDelete：是否自动删除。设置为true时，表示在最后一条使用该交换机的连接(Connection)断开时，自动删除这个exchange；
arguments：其他的一些参数，类型为Dictionary<string,object> 。
4. Virtual Host
1. 说明
RabbitMq的VirtualHost（虚拟消息服务器），每个VirtualHost相当于一个相对独立的RabbitMQ服务器；每个VirtualHost之间是相互隔离的，exchange、queue、message不能互通。
2. 创建Virtual Host
-
命令行
rabbitmqctl add_vhost 服务器名称
例如
rabbitmqctl add_vhost SX
-
创建用户并配置权限
创建用户：
rabbitmqctl add_user 用户名 密码
例如：
rabbitmqctl add_user sxadmin 123
为用户分配权限
rabbitmqctl set_permissions -p / 用户名 ".*" ".*" ".*"
例如：
rabbitmqctl set_permissions -p SX sxadmin ".*" ".*" ".*"
其中[/]可以访问所有的虚拟服务器（virtual host），如果改成虚拟服务器名称的话，表示该用户能访问这个虚拟服务器。
后面的3个[".*"]应该是配置，读，写的权限正则表达式（我暂时没理清这个）。
-
配置角色
rabbitmqctl set_user_tags 用户名 角色类型
例如
rabbitmqctl set_user_tags sxadmin policymaker
解释如下：none
不能访问 management pluginmanagement
用户可以通过AMQP做的任何事外加：
列出自己可以通过AMQP登入的virtual hosts
查看自己的virtual hosts中的queues, exchanges 和 bindings
查看和关闭自己的channels 和 connections
查看有关自己的virtual hosts的“全局”的统计信息，包含其他用户在这些virtual hosts中的活动。policymaker
management可以做的任何事外加：
查看、创建和删除自己的virtual hosts所属的policies和parametersmonitoring
management可以做的任何事外加：
列出所有virtual hosts，包括他们不能登录的virtual hosts
查看其他用户的connections和channels
查看节点级别的数据如clustering和memory使用情况
查看真正的关于所有virtual hosts的全局的统计信息administrator
policymaker和monitoring可以做的任何事外加:
创建和删除virtual hosts
查看、创建和删除users
查看创建和删除permissions
关闭其他用户的connections
第二部分
1. Direct
1.1 direct 路由规则
direct 类型的路由规则很简单
exchange 和 queue进行Binding时会设置routingKey(为了避免和下边的routingKey混淆，这里称为BindingKey)
channel.QueueBind(queue:"Q1", exchange:"myexchange", routingKey:"orange");
将消息发送到Broker时会设置对应的RoutingKey
channel.BasicPublish(exchange: "myexchange",routingKey: "orange", basicProperties: null, body: body);
| 只有RoutingKey 和BindingKey完全相同时，exchange才会把消息路由到绑定的queue中去 |
1.2 代码示例
我们知道了Direct类型的交换机只有routingKey和BindingKey相同时才会进行消息路由，根据这一点我们可以通过routingKey将消息路由到不同的queue中。如果在进行日志处理时，需求是所有的日志都保存到文本文件，出现错误日志时则还需要短信通知一遍及时处理。我们可以创建两个队列：只接收错误日志的log_error队列，和接收所有日志信息的log_all队列。消费者C1 处理log_error队列中的消息，将这些消息通过短信通知管理员，消费者C2 处理log_all队列，将这些信息写入文本文件。
生产者 发送日志消息
生产者发送
static void Main(string[] args)
{
var factory = new ConnectionFactory()
{
HostName = "127.0.0.1",
UserName = "wyy",//用户名
Password = "123321"//密码
};
//创建连接connection
using (var connection = factory.CreateConnection())
{
//创建通道channel
using (var channel = connection.CreateModel())
{
//声明交换机exchang
channel.ExchangeDeclare(exchange: "myexchange",
type: ExchangeType.Direct,
durable: true,
autoDelete: false,
arguments: null);
//声明两个队列，log_all保存所有日志，log_error保存error类型日志
channel.QueueDeclare(queue: "log_all",
durable: true,
exclusive: false,
autoDelete: false,
arguments: null);
channel.QueueDeclare(queue: "log_error",
durable: true,
exclusive: false,
autoDelete: false,
arguments: null);
//绑定所有日志类型到log_all队列
string[] logtypes = new string[] { "debug", "info", "warn", "error" };
foreach (string item in logtypes)
{
channel.QueueBind(queue: "log_all",
exchange: "myexchange",
routingKey: item);
}
//绑定错误日志到log_all队列
channel.QueueBind(queue: "log_error",
exchange: "myexchange",
routingKey: "error");
//准备100条测试日志信息
List msgList = new List();
for (int i = 1; i < 100; i++)
{
if (i%4==0)
{
msgList.Add(new LogMsg() { LogType = "info", Msg = Encoding.UTF8.GetBytes($"info第{i}条信息") });
}
if (i % 4 == 1)
{
msgList.Add(new LogMsg() { LogType = "debug", Msg = Encoding.UTF8.GetBytes($"debug第{i}条信息") });
}
if (i % 4 == 2)
{
msgList.Add(new LogMsg() { LogType = "warn", Msg = Encoding.UTF8.GetBytes($"warn第{i}条信息") });
}
if (i % 4 == 3)
{
msgList.Add(new LogMsg() { LogType = "error", Msg = Encoding.UTF8.GetBytes($"error第{i}条信息") });
}
}
Console.WriteLine("生产者发送100条日志信息");
//发送日志信息
foreach (var item in msgList)
{
channel.BasicPublish(exchange: "myexchange",
routingKey: item.LogType,
basicProperties: null,
body: item.Msg);
}
}
}
Console.ReadKey();
}
} 消费者C1 用于处理log_error队列中的消息，错误消息进行短信通知
消费者C1
static void Main(string[] args)
{
var factory = new ConnectionFactory()
{
//rabbitmq-server所在设备ip，这里就是本机
HostName = "127.0.0.1",
UserName = "wyy",//用户名
Password = "123321"//密码
};
//创建连接connection
using (var connection = factory.CreateConnection())
{
//创建通道channel
using (var channel = connection.CreateModel())
{
//声明交换机exchang
channel.ExchangeDeclare(exchange: "myexchange",
type: ExchangeType.Direct,
durable: true,
autoDelete: false,
arguments: null);
//声明队列queue
channel.QueueDeclare(queue: "log_all",
durable: true,
exclusive: false,
autoDelete: false,
arguments: null);
channel.QueueBind(queue:"log_error",exhcange:"myexchange",routingKey:"errot");
//定义消费者
var consumer = new EventingBasicConsumer(channel);
consumer.Received += (model, ea) =>
{
var body = ea.Body;
var message = Encoding.UTF8.GetString(body);
//只是为了演示，并没有存入文本文件
Console.WriteLine($"接收成功！【{message}】，发送短信通知");
};
Console.WriteLine("消费者C1【接收错误日志，发送短信通知】准备就绪....");
//处理消息
channel.BasicConsume(queue: "log_error",
autoAck: true,
consumer: consumer);
Console.ReadLine();
}
}
}
消费者C2 用于处理log_all队列中的消息，所有的消息记录到文本中。
消费者C2
static void Main(string[] args)
{
var factory = new ConnectionFactory()
{
//rabbitmq-server所在设备ip，这里就是本机
HostName = "127.0.0.1",
UserName = "wyy",//用户名
Password = "123321"//密码
};
//创建连接connection
using (var connection = factory.CreateConnection())
{
//创建通道channel
using (var channel = connection.CreateModel())
{
//声明交换机exchang
channel.ExchangeDeclare(exchange: "myexchange",
type: ExchangeType.Direct,
durable: true,
autoDelete: false,
arguments: null);
//声明队列queue
channel.QueueDeclare(queue: "log_all",
durable: true,
exclusive: false,
autoDelete: false,
arguments: null);
//绑定
string[] logtypes = new string[] { "debug", "info", "warn", "error" };
foreach (string item in logtypes)
{
channel.QueueBind(queue: "log_all",
exchange: "myexchange",
routingKey: item);
}
//定义消费者
var consumer = new EventingBasicConsumer(channel);
consumer.Received += (model, ea) =>
{
var body = ea.Body;
var message = Encoding.UTF8.GetString(body);
//只是为了演示，并没有存入文本文件
Console.WriteLine($"接收成功！【{message}】，存入文本文件");
};
Console.WriteLine("消费者C2【接收所有日志信息，存入文本文件】准备就绪....");
//处理消息
channel.BasicConsume(queue: "log_all",
autoAck: true,
consumer: consumer);
Console.ReadLine();
}
}
}
执行结果如下
2. Fanout
2.1 路由规则
Fanout类型的exchange路由规则是最简单的。交换机会把消息广播到与该exchange绑定的所有queue中，即所有和该exchange绑定的队列都会收到消息。fanout类型exchange和队列绑定时不需要指定routingkey，即使指定了routingkey也会被忽略掉。结构如图：
fanout类型主要用于发布订阅的一些场景。
2.2 代码示例
这里通过代码简单演示将消息同时使用短信和邮件两种方式通知用户的流程。首先声明一个fanout类型的exchange，然后声明两个队列SMSqueue和EmailQueue,这两个队列都和exchange绑定。消费者1处理EmailQueue,通过邮件方式发送；消费者2处理SMSqueue的消息通过短信方式发送通知。
生产者发送消息
static void Main(string[] args)
{
var factory = new ConnectionFactory()
{
//rabbitmq-server所在设备ip，这里就是本机
HostName = "127.0.0.1",
UserName = "wyy",//用户名
Password = "123321"//密码
};
//第一步：创建连接connection
using (var connection = factory.CreateConnection())
{
//创建通道channel
using (var channel = connection.CreateModel())
{
//声明交换机exchang
channel.ExchangeDeclare(exchange: "myfanoutexchange",
type: ExchangeType.Fanout,
durable: true,
autoDelete: false,
arguments: null);
//声明SMSqueue队列，用于短信通知
channel.QueueDeclare(queue: "SMSqueue",
durable: true,
exclusive: false,
autoDelete: false,
arguments: null);
//声明队列,Email队列，用于邮件通知
channel.QueueDeclare(queue: "EMAILqueue",
durable: true,
exclusive: false,
autoDelete: false,
arguments: null);
//绑定exchange和queue
channel.QueueBind(queue: "SMSqueue", exchange: "myfanoutexchange", routingKey: string.Empty,arguments:null);
channel.QueueBind(queue: "EMAILqueue", exchange: "myfanoutexchange", routingKey: string.Empty, arguments: null);
Console.WriteLine("生产者准备就绪....");
string message = "";
//第六步：发送消息
//在控制台输入消息，按enter键发送消息
while (!message.Equals("quit", StringComparison.CurrentCultureIgnoreCase))
{
message = Console.ReadLine();
var body = Encoding.UTF8.GetBytes(message);
//基本发布
channel.BasicPublish(exchange: "myfanoutexchange",
routingKey: string.Empty,
basicProperties: null,
body: body);
Console.WriteLine($"消息【{message}】已发送到队列");
}
}
}
Console.ReadKey();
}
消费者1 将Emailqueue的消息通过邮件的方式发送通知
static void Main(string[] args)
{
var factory = new ConnectionFactory()
{
//rabbitmq-server所在设备ip，这里就是本机
HostName = "127.0.0.1",
UserName = "wyy",//用户名
Password = "123321"//密码
};
//创建连接connection
using (var connection = factory.CreateConnection())
{
//创建通道channel
using (var channel = connection.CreateModel())
{
//声明交换机exchang
channel.ExchangeDeclare(exchange: "myfanoutexchange",
type: ExchangeType.Fanout,
durable: true,
autoDelete: false,
arguments: null);
//声明队列queue
channel.QueueDeclare(queue: "EMAILqueue",
durable: true,
exclusive: false,
autoDelete: false,
arguments: null);
//绑定exchange和queue
channel.QueueBind(queue: "EMAILqueue", exchange: "myfanoutexchange", routingKey: string.Empty, arguments: null);
//定义消费者
var consumer = new EventingBasicConsumer(channel);
consumer.Received += (model, ea) =>
{
var body = ea.Body;
var message = Encoding.UTF8.GetString(body);
//只是为了演示，并没有存入文本文件
Console.WriteLine($"接收成功！【{message}】，邮件通知");
};
Console.WriteLine("邮件通知服务准备就绪...");
//处理消息
channel.BasicConsume(queue: "EMAILqueue",
autoAck: true,
consumer: consumer);
Console.ReadLine();
}
}
}
消费者2 将SMSqueue的消息通过短信的方式发送通知
static void Main(string[] args)
{
var factory = new ConnectionFactory()
{
//rabbitmq-server所在设备ip，这里就是本机
HostName = "127.0.0.1",
UserName = "wyy",//用户名
Password = "123321"//密码
};
//创建连接connection
using (var connection = factory.CreateConnection())
{
//创建通道channel
using (var channel = connection.CreateModel())
{
//声明交换机exchang
channel.ExchangeDeclare(exchange: "myfanoutexchange",
type: ExchangeType.Fanout,
durable: true,
autoDelete: false,
arguments: null);
//声明队列queue
channel.QueueDeclare(queue: "SMSqueue",
durable: true,
exclusive: false,
autoDelete: false,
arguments: null);
//绑定exchange和queue
channel.QueueBind(queue: "SMSqueue", exchange: "myfanoutexchange", routingKey: string.Empty,arguments:null);
//定义消费者
var consumer = new EventingBasicConsumer(channel);
consumer.Received += (model, ea) =>
{
var body = ea.Body;
var message = Encoding.UTF8.GetString(body);
//只是为了演示，并没有存入文本文件
Console.WriteLine($"接收成功！【{message}】，短信通知");
};
Console.WriteLine("短信通知服务准备就绪...");
//处理消息
channel.BasicConsume(queue: "myfanoutqueue",
autoAck: true,
consumer: consumer);
Console.ReadLine();
}
}
}
执行结果
3. Topic
3.1 路由规则
topic 类型的路由规则也是基于routingkey和bindingkey的，其路由过程和direct类型基本一致。两者的区别在于direct类型的exchange要求routingkey 和 bindingkey必须相同才能能将消息路由到绑定的queue中，而topic类型的bindingkey是一个匹配规则，只要routingkey符合bingdingkey的规则就可以将消息路由到绑定的queue中。结构如图。注意routingkey和bindingkey的结构都是一系列由点号连接单词的字符串，例如【aaa.bb.cc】。
bindingkey的两个特殊的符号：*表示一个单词，#表示0或多个单词(注意是单词，不是字符)。如下图，usa.news和usa.weather都和usa.#匹配，而usa.news和europe.news都和#.news匹配
3.2 代码示例
这里使用代码实现上面的例子，为了方便我们定义两个队列：接收美国相关信息的usaQueue(bindingKey是usa.#)和接收新闻消息的newsQueue(bindingKey是#.news)。然后定义两个消费者，消费者1处理usaQueue的消息，消费者2处理newsQueue的消息。
生产者代码：
static void Main(string[] args)
{
var factory = new ConnectionFactory()
{
//rabbitmq-server所在设备ip，这里就是本机
HostName = "127.0.0.1",
UserName = "wyy",//用户名
Password = "123321"//密码
};
//创建连接connection
using (var connection = factory.CreateConnection())
{
//创建通道channel
using (var channel = connection.CreateModel())
{
//声明交换机exchang
channel.ExchangeDeclare(exchange: "mytopicExchange",
type: ExchangeType.Topic,
durable: true,
autoDelete: false,
arguments: null);
//声明队列usaQueue
channel.QueueDeclare(queue: "usaQueue",
durable: true,
exclusive: false,
autoDelete: false,
arguments: null);
//声明队列newsQueue
channel.QueueDeclare(queue: "newsQueue",
durable: true,
exclusive: false,
autoDelete: false,
arguments: null);
Console.WriteLine("生产者准备就绪....");
//绑定usaQueue队列到交互机,routingKey为usa.#
channel.QueueBind(queue: "usaQueue", exchange: "mytopicExchange", routingKey: "usa.#", arguments: null);
//绑定newsQueue队列到交互机,routingKey为#.news
channel.QueueBind(queue: "newsQueue", exchange: "mytopicExchange", routingKey: "#.news", arguments: null);
////--------------------开始发送消息
//1.发送美国新闻消息
string message1 = "美国新闻消息：内容balabala";
var body1 = Encoding.UTF8.GetBytes(message1);
channel.BasicPublish(exchange: "mytopicExchange",
routingKey: "usa.news",
basicProperties: null,
body: body1);
Console.WriteLine($"消息【{message1}】已发送到队列");
//2.发送美国天气消息
string message2 = "美国天气消息：内容balabala";
var body2 = Encoding.UTF8.GetBytes(message2);
channel.BasicPublish(exchange: "mytopicExchange",
routingKey: "usa.weather",
basicProperties: null,
body: body2);
Console.WriteLine($"消息【{message2}】已发送到队列");
//3.发送欧洲新闻消息
string message3 = "欧洲新闻消息：内容balabala";
var body3 = Encoding.UTF8.GetBytes(message3);
channel.BasicPublish(exchange: "mytopicExchange",
routingKey: "europe.news",
basicProperties: null,
body: body3);
Console.WriteLine($"消息【{message3}】已发送到队列");
//4.发送欧洲天气消息
string message4 = "欧洲天气消息：内容balabala";
var body4 = Encoding.UTF8.GetBytes(message4);
//基本发布
channel.BasicPublish(exchange: "mytopicExchange",
routingKey: "europe.weather",
basicProperties: null,
body: body4);
Console.WriteLine($"消息【{message4}】已发送到队列");
}
}
Console.ReadKey();
}
消费者1代码，只处理usaQueue中的消息
static void Main(string[] args)
{
var factory = new ConnectionFactory()
{
//rabbitmq-server所在设备ip，这里就是本机
HostName = "127.0.0.1",
UserName = "wyy",//用户名
Password = "123321"//密码
};
//创建连接connection
using (var connection = factory.CreateConnection())
{
//创建通道channel
using (var channel = connection.CreateModel())
{
//声明交换机exchang
channel.ExchangeDeclare(exchange: "mytopicExchange",
type: ExchangeType.Topic,
durable: true,
autoDelete: false,
arguments: null);
//声明队列queue
channel.QueueDeclare(queue: "usaQueue",
durable: true,
exclusive: false,
autoDelete: false,
arguments: null);
Console.WriteLine("usaQueue消费者准备就绪....");
//绑定usaQueue队列到交互机
channel.QueueBind(queue: "usaQueue", exchange: "mytopicExchange", routingKey: "usa.#", arguments: null);
//定义消费者
var consumer = new EventingBasicConsumer(channel);
consumer.Received += (model, ea) =>
{
var body = ea.Body;
var message = Encoding.UTF8.GetString(body);
Console.WriteLine($"接收成功！【{message}】");
};
//处理消息
channel.BasicConsume(queue: "usaQueue",
autoAck: true,
consumer: consumer);
Console.ReadLine();
}
}
}
消费者2接收newsQueue中的消息：
static void Main(string[] args)
{
var factory = new ConnectionFactory()
{
//rabbitmq-server所在设备ip，这里就是本机
HostName = "127.0.0.1",
UserName = "wyy",//用户名
Password = "123321"//密码
};
//创建连接connection
using (var connection = factory.CreateConnection())
{
//创建通道channel
using (var channel = connection.CreateModel())
{
//声明交换机exchang
channel.ExchangeDeclare(exchange: "mytopicExchange",
type: ExchangeType.Topic,
durable: true,
autoDelete: false,
arguments: null);
//声明队列queue
channel.QueueDeclare(queue: "newsQueue",
durable: true,
exclusive: false,
autoDelete: false,
arguments: null);
Console.WriteLine("newsQueue消费者准备就绪....");
//绑定usaQueue队列到交互机
channel.QueueBind(queue: "newsQueue", exchange: "mytopicExchange", routingKey: "#.news", arguments: null);
//定义消费者
var consumer = new EventingBasicConsumer(channel);
consumer.Received += (model, ea) =>
{
var body = ea.Body;
var message = Encoding.UTF8.GetString(body);
Console.WriteLine($"接收成功！【{message}】");
};
//处理消息
channel.BasicConsume(queue: "newsQueue",
autoAck: true,
consumer: consumer);
Console.ReadLine();
}
}
}
生产者发送的四条消息中，消息1的routingKey为usa.news,同时符合usaQueue的bindingKey和newsQueue的bindingKey，所以消息1同时路由到两个队列中；消息2的routingKey为usa.weather只符合usa.#,发送到usaQueue;消费的routkey为europe.news，只符合#.news，发送到newsQueue;消息四的routingKey为europe.weather,和两个队列的bindingKey都不符合，所以被丢弃。执行结果：
一点补充：topic 类型交换机十分灵活，可以轻松实现direct和fanout类型交换机的功能。如果绑定队列时所有的bingingKey都是#，则交换机和fanout类型交换机变现一致；如果所有的bindingkey都不包含*和#，则交换机和direct类型交换机表现一致。
4. Header
由于使用较少，暂未学习。
第三部分
1. 消息确认
在一些场合，如转账、付费时每一条消息都必须保证成功的被处理。AMQP是金融级的消息队列协议，有很高的可靠性，在这里介绍RabbitMQ是怎么保证消息被成功处理的。消息确认可以分为两种：一种是生产者发送消息到Broker时，Broker给生产者发送消息回执，用于告诉生产者消息已被成功发送到Broker；一种是消费者接收到Broker发送的消息时，消费者给Broker发送确认回执，用于通知消息已成功被消息者接收。
下面分别介绍生产端和消费端的消息确认方法。准备条件：使用Web管理工具添加exchange，queue,并绑定，bindingKey为myKey,如下所示：
1.1 生产端消息确认(tx机制和Confirm模式)
生产端的消息确认：当生产者将消息发送给broker时，broker接收到消息给生产者发送消息回执。生产端的消息确认方式有两种：tx机制和Confirm模式。
tx机制
tx机制可以叫做事务机制，RabbitMq中有三个与tx机制的方法：txSelect(),txCommit(),txRollback()。channel.txSelect()用于将当前的channel设置成transaction模式；channel.txCommit()提交事务channel.txRollback()回滚事务。使用tx机制，我们首先要通过txSelect方法开启事务，然后发布消息给broker服务器，如果txCommit提交成功，则说明消息成功被Broker接收了；如果txCommit执行之前，broker异常崩溃或者由于其他原因抛出异常，这个时候我们可以捕获异常，通过txRaollback回滚事务。看一个tx机制简单的实现：
var factory = new ConnectionFactory()
{
//rabbitmq-server所在设备ip，这里就是本机
HostName = "127.0.0.1",
UserName = "wyy",//用户名
Password = "123321"//密码
};
//创建连接connection
using (var connection = factory.CreateConnection())
{
//创建通道channel
using (var channel = connection.CreateModel())
{
Console.WriteLine("生产者准备就绪....");
string message = "";
//发送消息
//在控制台输入消息，按enter键发送消息
while (!message.Equals("quit", StringComparison.CurrentCultureIgnoreCase))
{
message = Console.ReadLine();
var body = Encoding.UTF8.GetBytes(message);
try
{
//开启事务机制
channel.TxSelect();
//发送消息
channel.BasicPublish(exchange: "myexchange",
routingKey: "mykey",
basicProperties: null,
body: body);
//事务提交
channel.TxCommit();
Console.WriteLine($"【{message}】发送到Broke成功！");
}
catch (Exception)
{
Console.WriteLine($"【{message}】发送到Broker失败！");
channel.TxRollback();
}
}
}
}
Console.ReadKey();
}
运行结果
Confirm模式
C# 的RabbitMQ API中有三个方法跟Confirm相关的方法：ConfirmSelect(),WaitForConfirms()和WaitForConfirmOrDie()。channel.ConfirmSelect()表示开启Confirm模式；channel.WaitForConfirms等待所有消息确认，如果所有的消息都被服务端成功接收则返回true,只要有一条没有被成功接收就返回false。channel.WaitForConfirmsOrDie()和WaitForConfirm作用类似，也是等待所有消息确认，区别在于该方法没有返回值（Void）,如果有任意一条消息没有被成功接收，该方法会立刻抛出一个OperationInterrupedException类型异常。看一个Confirm模式简单实现：
static void Main(string[] args)
{
var factory = new ConnectionFactory()
{
//rabbitmq-server所在设备ip，这里就是本机
HostName = "127.0.0.1",
UserName = "wyy",//用户名
Password = "123321"//密码
};
//创建连接connection
using (var connection = factory.CreateConnection())
{
//创建通道channel
using (var channel = connection.CreateModel())
{
Console.WriteLine("生产者准备就绪....");
string message = "";
//在控制台输入消息，按enter键发送消息
while (!message.Equals("quit", StringComparison.CurrentCultureIgnoreCase))
{
message = Console.ReadLine();
var body = Encoding.UTF8.GetBytes(message);
//开启Confirm模式
channel.ConfirmSelect();
//发送消息
channel.BasicPublish(exchange: "myexchange",
routingKey: "mykey",
basicProperties: null,
body: body);
//WaitForConfirms确认消息(可以同时确认多条消息)是否发送成功
if (channel.WaitForConfirms())
{
Console.WriteLine($"【{message}】发送到Broke成功！");
}
}
}
}
Console.ReadKey();
}
运行结果
1.2 消费者端消息确认(自动确认和显示确认)
从Broker发送到消费者时有两种确认方法：自动确认和显示确认。
自动确认
自动确认：当RabbitMQ将消息发送给消费者后，消费者端接收到消息后，不等待消息处理结束，立即自动回送一个确认回执。自动确认的方法十分简单，设置消费方法的参数autoAck=true即可，我们前面的例子都是使用的自动确认。
注： Broker会在接收到确认回执时删除信息，如果消费者收到消息并返回了确认回执，然后消费者在处理消息时挂了，那么这条消息再也找不回来了。
显示确认
我们知道自动确认可能导致消息丢失的问题，我们不免想到：Broker收到消息回执时删除消息，如果可以让消费者在接收到消息时不立即返回确认回执，等到消息处理完成后(或者完成一部分的逻辑)再确认返回回执，这样就保证消费端不会丢失消息了。这正是显示确认的思路。使用显示确认也比较简单，首先将Resume的方法参数AutoAck设置为false,然后在消费端使用代码：channel.BasicAck()/BaseicReject()等方法来确认和拒绝消息。
生产者代码：
static void Main(string[] args)
{
var factory = new ConnectionFactory()
{
//rabbitmq-server所在设备ip，这里就是本机
HostName = "127.0.0.1",
UserName = "wyy",//用户名
Password = "123321"//密码
};
//创建连接connection
using (var connection = factory.CreateConnection())
{
//创建通道channel
using (var channel = connection.CreateModel())
{
Console.WriteLine("生产者准备就绪....");
string message = "";
//发送消息
//在控制台输入消息，按enter键发送消息
while (!message.Equals("quit", StringComparison.CurrentCultureIgnoreCase))
{
message = Console.ReadLine();
var body = Encoding.UTF8.GetBytes(message);
//基本发布
channel.BasicPublish(exchange: "myexchange",
routingKey: "mykey",
basicProperties: null,
body: body);
Console.WriteLine($"消息【{message}】已发送到队列");
}
}
}
Console.ReadKey();
}
消费者代码：
static void Main(string[] args)
{
var factory = new ConnectionFactory()
{
//rabbitmq-server所在设备ip，这里就是本机
HostName = "127.0.0.1",
UserName = "wyy",//用户名
Password = "123321"//密码
};
using (var connection = factory.CreateConnection())
{
using (var channel = connection.CreateModel())
{
//定义消费者
var consumer = new EventingBasicConsumer(channel);
consumer.Received += (model, ea) =>
{
string message = Encoding.UTF8.GetString(ea.Body);
Console.WriteLine($"接受到消息【{message}】");
//以news开头表示是新闻类型，处理完成后确认消息
if (message.StartsWith("news"))
{
//这里处理消息balabala
Console.WriteLine($"【{message}】是新闻消息，处理消息并确认");
channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);
}
//不以news开头表示不是新闻类型，不进行处理，把消息退回到queue中
else
{
Console.WriteLine($"【{message}】不是新闻类型，拒绝处理");
channel.BasicReject(deliveryTag: ea.DeliveryTag, requeue: false);
}
};
Console.WriteLine("消费者准备就绪....");
//第五步：处理消息
channel.BasicConsume(queue: "myqueue",
autoAck: false,
consumer: consumer);
Console.ReadKey();
}
}
}
channel.BasicAck(deliveryTag:ea.DeliveryTag,multiple:false)方法用于确认消息，deliveryTag是分发的标记，multiple表示是否确认多条。channel.BasicReject(deliveryTag:ea.DeliveryTag,requeue:false)用于拒绝消息，deliveryTag用于分发标记，requeue表示消息被拒绝后是否重新放回queue中，true表示放回queue中，false表示直接放弃。运行结果：
意外情况：使用显示确认时，如果消费者处理完消息不发送确认回执，那么消息不会被删除，消息的状态一直是Unacked，这条消息也不会在发送给其他的消费者。如果一个消费者在处理消息时尚未发送确认回执的情况下挂掉了，那么消息会被重新放回队列（状态从Unacked编程Ready），有其他消费者存在时发给其他消费者。
2. 消息持久化/优先级
2.1 消息持久化(Persistent)
前面接收exchange和queue的持久化，把exchange和queue的durable属性设置为true,重启rabbitmq服务时，exchange和queue也会回复，但是队列内的消息却是丢失。那么怎么实现消息持久化呢？实现的方法很简单，将exchange和queue的durable设置为true外，在发布消息的时候设置DeliveryMode = 2。
static void Main(string[] args)
{
var factory = new ConnectionFactory()
{
//rabbitmq-server所在设备ip，这里就是本机
HostName = "127.0.0.1",
UserName = "wyy",//用户名
Password = "123321"//密码
};
//创建连接connection
using (var connection = factory.CreateConnection())
{
//创建通道channel
using (var channel = connection.CreateModel())
{
Console.WriteLine("生产者准备就绪....");
string message = "";
//在控制台输入消息，按enter键发送消息
while (!message.Equals("quit", StringComparison.CurrentCultureIgnoreCase))
{
message = Console.ReadLine();
var body = Encoding.UTF8.GetBytes(message);
//设置消息持久化
var props = channel.CreateBasicProperties();
props.DeliveryMode = 2;
channel.BasicPublish(exchange: "myexchange",
routingKey: "mykey",
basicProperties: props,
body: body);
Console.WriteLine($"【{message}】发送到Broke成功！");
}
}
}
Console.ReadKey();
}
2.2 优先级
我们知道queue是先进先出的，即先发送的消息先被消费。但在具体业务中可能遇到需要提前被消费的需求，入一个常见的需求：普通客户的消费按照先进先出的顺序，Vip客户的消息要提前处理。消息实现优先级控制的方式是：首先在声明queue时设置队列的x-max-priority属性，然后在publish时设置消息的优先等级即可。
生产者：
static void Main(string[] args)
{
var factory = new ConnectionFactory()
{
//rabbitmq-server所在设备ip，这里就是本机
HostName = "127.0.0.1",
UserName = "wyy",//用户名
Password = "123321"//密码
};
//创建连接connection
using (var connection = factory.CreateConnection())
{
//创建通道channel
using (var channel = connection.CreateModel())
{
//声明交换机exchang
channel.ExchangeDeclare(exchange: "myexchange",
type: ExchangeType.Direct,
durable: true,
autoDelete: false,
arguments: null);
//声明队列queue
channel.QueueDeclare(queue: "myqueue",
durable: true,
exclusive: false,
autoDelete: false,
arguments: new Dictionary<string, object>() {
//队列优先级最高为10，不加x-max-priority的话，计算发布时设置了消息的优先级也不会生效
{"x-max-priority",10 }
});
//绑定exchange和queue
channel.QueueBind(queue: "myqueue", exchange: "myexchange", routingKey: "mykey");
Console.WriteLine("生产者准备就绪....");
//一些待发送的消息
string[] msgs = { "vip1", "hello2", "world3","common4", "vip5" };
//设置消息优先级
var props = channel.CreateBasicProperties();
foreach (string msg in msgs)
{
//vip开头的消息，优先级设置为9
if (msg.StartsWith("vip"))
{
props.Priority = 9;
channel.BasicPublish(exchange: "myexchange",
routingKey: "mykey",
basicProperties: props,
body: Encoding.UTF8.GetBytes(msg));
}
//其他消息的优先级为1
else
{
props.Priority = 1;
channel.BasicPublish(exchange: "myexchange",
routingKey: "mykey",
basicProperties: props,
body: Encoding.UTF8.GetBytes(msg));
}
}
}
}
Console.ReadKey();
}
不需要对消费者做额外的配置。
第四部分
1.1 消费模式
EventingBasicConsumer
EventingBasicConsumer是发布/订阅模式的消费者，即只要订阅了queue中的消息，Broker会立即把消息推送给消费者。EventingBasicConsumer是长连接模式，只要创建一个Connection，然后在Connection的基础上创建通道channel,消息的发送都是通过channel执行的，这样可以减少Connection的创建。前面使用了很多EventingBasicConsumer接收数据。
BasicGet
Get方式是短链接，消费者每次想获取消息，首先建立一个Connection，发送一个请求，Broker接收请求之后，响应一条消息给消费者，然后断开。RabbitMQ中的Get方式和HTTP中的请求响应流程基本一致，Get实时性较差，比较耗费资源。
1.2 QOS
通过显式确认的方式可以解决消息丢失的问题，但这种方式也存在一些问题：①当消息有十万，百万条时，一股脑的把消息发送给消费者，可能会造成消费者内存爆满；②当消息处理比较慢的时，单一的消费者处理这些消息可能很长时间，我们自然想到再添加一个消费者加快消息的处理速度，但是这些消息都被原来的消费者接收了，状态为Unacked，所以这些消息不会再发送给新添加的消费者。针对这些问题怎么去解决呢？
RabbitMQ提供的Qos(服务质量)可以完美解决上边的问题，使用Qos时，Broker不会再把消息一股脑的发送给消费者，我们可以设置每次传输给消费者的消息条数n，消费者把这n条消息处理完成后，再获取n条数据进行处理，这样就不用担心消息丢失、服务端内存爆满的问题了，因为没有发送的消息状态都是Ready，所以当我们新增一个消费者时，消息也可以立即发送给新增的消费者。注意Qos只有在消费端使用显示确认时才有效，使用Qos的方式十分简单，在消费端调用 channel.BasicQos() 方法即可，修改服务端代码如下：
static void Main(string[] args)
{
var factory = new ConnectionFactory()
{
//rabbitmq-server所在设备ip，这里就是本机
HostName = "127.0.0.1",
UserName = "wyy",//用户名
Password = "123321"//密码
};
using (var connection = factory.CreateConnection())
{
using (var channel = connection.CreateModel())
{
channel.BasicQos(prefetchSize: 0, prefetchCount: 2, global: false);
#region EventingBasicConsumer
//定义消费者
var consumer = new EventingBasicConsumer(channel);
//接收到消息时执行的任务
consumer.Received += (model, ea) =>
{
Thread.Sleep(1000 * 5);
//处理完成，手动确认
channel.BasicAck(ea.DeliveryTag, false);
Console.WriteLine($"处理消息【{Encoding.UTF8.GetString(ea.Body)}】完成");
};
Console.WriteLine("消费者准备就绪....");
//处理消息
channel.BasicConsume(queue: "myqueue",
autoAck: false,
consumer: consumer);
Console.ReadKey();
#endregion
}
}
}
channel.BasicQos(prefetchSize: 0, prefetchCount: 2, global: false) 方法中参数prefetchSize为预取的长度，一般设置为0即可，表示长度不限；prefetchCount表示预取的条数，即发送的最大消息条数；global表示是否在Connection中全局设置，true表示Connetion下的所有channel都设置为这个配置。
第五部分
1.1 搭建RabbitMQ集群
研究中……
1.2 Ngixn+RabbitMQ 负载均衡
研究中……