下载安装
Erlang
RabbitMQ
启动RabbitMQ管理平台插件
DOS下进入到安装目录\sbin,执行以下命令
rabbitmq-plugins enable rabbitmq_management
当出现以下结果时,重启RabbitMQ服务
set 3 plugins.
Offline change; changes will take effect at broker restart.
访问http://localhost:15672(账号密码:guest)
注意:以下为C#代码,请引用NuGet包:RabbitMQ.Client
参考文章
RabbitMQ快速入门
名词解析
P(Publisher):生产者
C(Consumer):消费者
Channel:信道
Queue:队列
Exchange:信息交换机
简单演示
信息发送端
static void Send()
{
//1. 实例化连接工厂
var factory = new ConnectionFactory() { HostName = "localhost" };
//2. 建立连接
using (var connection = factory.CreateConnection())
{
//3. 创建信道
using (var channel = connection.CreateModel())
{
//4. 声明队列
channel.QueueDeclare(queue: "rabbitmq",
durable: false,
exclusive: false,
autoDelete: false,
arguments: null);
//5. 构建字节数据包
var message = "Hello RabbitMQ!";
var body = Encoding.UTF8.GetBytes(message);
//6. 发送数据包
channel.BasicPublish(exchange: "",
routingKey: "rabbitmq",
basicProperties: null,
body: body);
Console.WriteLine(" [x] Sent {0}", message);
}
}
}
信息接收端
static void Receive()
{
//1. 实例化连接工厂
var factory = new ConnectionFactory() { HostName = "localhost" };
//2. 建立连接
using (var connection = factory.CreateConnection())
{
//3. 创建信道
using (var channel = connection.CreateModel())
{
//4. 声明队列
channel.QueueDeclare(queue: "rabbitmq",
durable: false,
exclusive: false,
autoDelete: false,
arguments: null);
//5. 构造消费者实例
var consumer = new EventingBasicConsumer(channel);
//6. 绑定消息接收后的事件委托
consumer.Received += (model, ea) =>
{
var message = Encoding.UTF8.GetString(ea.Body);
Console.WriteLine(" [x] Received {0}", message);
};
//7. 启动消费者
channel.BasicConsume(queue: "rabbitmq",
autoAck: true,
consumer: consumer);
Console.WriteLine(" Press [enter] to exit.");
Console.ReadLine();
}
}
}
轮询调度
P生产的多个任务进入到队列中,多个C间可以并行处理任务。默认情况下,RabbitMQ把信息按顺序发送给每一个C。平均每个C将获得同等数量的信息。
信息确认
按照最简单的演示来说,信息一旦发送到C中,则该信息就会从队列中移除。一旦中间信息处理异常/失败,C端程序退出等,都将会导致信息未处理完成,而此时队列中已将信息移除了,那么就会导致一系列的问题。我们可以在C端设置手动确认信息,从而解决上述问题的发生。
Receive代码块
//6. 绑定消息接收后的事件委托
consumer.Received += (model, ea) =>
{
var message = Encoding.UTF8.GetString(ea.Body);
Console.WriteLine(" [x] Received {0}", message);
Thread.Sleep(5000);//模拟耗时
Console.WriteLine(" [x] Done");
// 发送信息确认信号(手动信息确认)
channel.BasicAck(ea.DeliveryTag, false);
};
//7. 启动消费者
/*
autoAck参数属性
true:自动信息确认,当C端接收到信息后,自动发送ack信号,不管信息是否处理完毕
false:关闭自动信息确认,通过调用BasicAck方法手动进行信息确认
*/
channel.BasicConsume(queue: "rabbitmq",
autoAck: false,
consumer: consumer);
信息持久化
当RabbitMQ退出或死机时会清空队列和信息。通过将队列和信息标记为持久的,来告知RabbitMQ将信息持久化。
Send代码块
//4. 声明队列
//durable设置为true,表示此队列为持久的。
//注意:RabbitMQ不允许你使用不同的参数重新定义一个已经存在的队列,所以你需要重启服务/更改队列名称
channel.QueueDeclare(queue: "rabbitmq",
durable: true, //标记队列持久
exclusive: false,
autoDelete: false,
arguments: null);
//设置IbasicProperties.SetPersistent属性值为true来标记我们的消息持久化
var properties = channel.CreateBasicProperties();
properties.Persistent = true;
//5. 构建字节数据包
var message = "Hello RabbitMQ!";
var body = Encoding.UTF8.GetBytes(message);
//6. 发送数据包
channel.BasicPublish(exchange: "",
routingKey: "rabbitmq",
basicProperties: properties, //指定BasicProperties
body: body);
公平调度
上述演示中,如果队列中存在多个信息,在开启多个C的情况下,只有一个C忙个不停,另外的却一直处于空闲状态。通过调用BasicQos,告知RabbitMQ在某个C信息处理完毕,并且已经收到信息确认之后,才可以继续发送信息到这个C。否则,将会把信息分发到另外空闲的C。
Receive代码块
//4. 声明队列
channel.QueueDeclare(queue: "rabbitmq",
durable: true,
exclusive: false,
autoDelete: false,
arguments: null);
////设置prefetchCount为1来告知RabbitMQ在未收到消费端的消息确认时,不再分发消息
channel.BasicQos(prefetchSize: 0,
prefetchCount: 1,
global: false);
发布/订阅
上述中的演示,P推送信息至队列中,C从队列中处理信息。但是如果需要将P推送的信息至每个订阅的C中处理信息,那么我们就可以使用Exchange。
fanout(将信息分发到exchange上绑定的所有队列上)
Send代码块
//1. 实例化连接工厂
var factory = new ConnectionFactory() { HostName = "localhost" };
//2. 建立连接
using (var connection = factory.CreateConnection())
{
//3. 创建信道
using (var channel = connection.CreateModel())
{
//4. 声明信息交换机
channel.ExchangeDeclare(exchange: "fanoutDemo",
type: "fanout");
//5. 构建字节数据包
var message = "Hello RabbitMQ!";
var body = Encoding.UTF8.GetBytes(message);
//6. 发布到指定exchange,fanout类型的会忽视routingKey的值,所以无需填写
channel.BasicPublish(exchange: "fanoutDemo",
routingKey: "",
basicProperties: null,
body: body);
Console.WriteLine(" [x] Sent {0}", message);
}
}
Receive代码块
//1. 实例化连接工厂
var factory = new ConnectionFactory() { HostName = "localhost" };
//2. 建立连接
using (var connection = factory.CreateConnection())
{
//3. 创建信道
using (var channel = connection.CreateModel())
{
//4. 声明信息交换机
channel.ExchangeDeclare(exchange: "fanoutDemo",
type: "fanout");
//生成随机队列名称
var queueName = channel.QueueDeclare().QueueName;
//绑定队列到指定fanout类型exchange
channel.QueueBind(queue: queueName,
exchange: "fanoutDemo",
routingKey: "");
//5. 构造消费者实例
var consumer = new EventingBasicConsumer(channel);
//6. 绑定消息接收后的事件委托
consumer.Received += (model, ea) =>
{
var message = Encoding.UTF8.GetString(ea.Body);
Console.WriteLine(" [x] Received {0}", message);
};
channel.BasicConsume(queue: queueName,
autoAck: true,
consumer: consumer);
Console.WriteLine(" Press [enter] to exit.");
Console.ReadLine();
}
}
direct(C绑定的队列名称须和P发布指定的路由名称一致)
Send代码块
//4. 声明信息交换机
channel.ExchangeDeclare(exchange: "directDemo",
type: "direct");
//5. 构建字节数据包
var message = "Hello RabbitMQ!";
var body = Encoding.UTF8.GetBytes(message);
//6. 发布到指定exchange
channel.BasicPublish(exchange: "directDemo",
routingKey: "a",
basicProperties: null,
body: body);
Receive代码块
//4. 声明信息交换机
channel.ExchangeDeclare(exchange: "directDemo",
type: "direct");
//生成随机队列名称
var queueName = channel.QueueDeclare().QueueName;
//绑定队列到指定direct类型exchange
channel.QueueBind(queue: queueName,
exchange: "directDemo",
routingKey: "b");
topic(支持通配符的路由规则)
通配字符:
- *:匹配一个单词
- #:匹配0个或多个单词
- .:仅作为分隔符
Send代码块
//4. 声明信息交换机
channel.ExchangeDeclare(exchange: "topicDemo",
type: "topic");
//5. 构建字节数据包
var message = "Hello RabbitMQ!";
var body = Encoding.UTF8.GetBytes(message);
//6. 发布到指定exchange
channel.BasicPublish(exchange: "topicDemo",
routingKey: "admin.user.error", //模拟后台用户错误
basicProperties: null,
body: body);
Receive代码块
//4. 声明信息交换机
channel.ExchangeDeclare(exchange: "topicDemo",
type: "topic");
//生成随机队列名称
var queueName = channel.QueueDeclare().QueueName;
//绑定队列到指定topic类型exchange
channel.QueueBind(queue: queueName,
exchange: "topicDemo",
routingKey: "admin.*.#"); //订阅所有后台异常错误
RPC(远程过程调用)
- 进行远程调用的客户端需要指定接收远程回调的队列,并申明消费者监听此队列。
- 远程调用的服务端除了要申明消费端接收远程调用请求外,还要将结果发送到客户端用来监听的结果的队列中去。
客户端代码块
var factory = new ConnectionFactory() { HostName = "localhost" };
using (var connection = factory.CreateConnection())
{
using (var channel = connection.CreateModel())
{
var correlationId = Guid.NewGuid().ToString();
var replyQueue = channel.QueueDeclare().QueueName;
var properties = channel.CreateBasicProperties();
properties.ReplyTo = replyQueue;
properties.CorrelationId = correlationId;
string number = args.Length > 0 ? args[0] : "30";
var body = Encoding.UTF8.GetBytes(number);
//发布消息
channel.BasicPublish(exchange: "", routingKey: "rpc_queue", basicProperties: properties, body: body);
Console.WriteLine($"[*] Request fib({number})");
// //创建消费者用于消息回调
var callbackConsumer = new EventingBasicConsumer(channel);
channel.BasicConsume(queue: replyQueue, autoAck: true, consumer: callbackConsumer);
callbackConsumer.Received += (model, ea) =>
{
if (ea.BasicProperties.CorrelationId == correlationId)
{
var responseMsg = $"Get Response: {Encoding.UTF8.GetString(ea.Body)}";
Console.WriteLine($"[x]: {responseMsg}");
}
};
Console.ReadLine();
}
}
服务端代码块
static void Main(string[] args)
{
var factory = new ConnectionFactory() { HostName = "localhost" };
using (var conection = factory.CreateConnection())
{
using (var channel = conection.CreateModel())
{
channel.QueueDeclare(queue: "rpc_queue", durable: false,
exclusive: false, autoDelete: false, arguments: null);
var consumer = new EventingBasicConsumer(channel);
Console.WriteLine("[*] Waiting for message.");
consumer.Received += (model, ea) =>
{
var message = Encoding.UTF8.GetString(ea.Body);
int n = int.Parse(message);
Console.WriteLine($"Receive request of Fib({n})");
int result = Fib(n);
var properties = ea.BasicProperties;
var replyProerties = channel.CreateBasicProperties();
replyProerties.CorrelationId = properties.CorrelationId;
channel.BasicPublish(exchange: "", routingKey: properties.ReplyTo,
basicProperties: replyProerties, body: Encoding.UTF8.GetBytes(result.ToString()));
channel.BasicAck(ea.DeliveryTag, false);
Console.WriteLine($"Return result: Fib({n})= {result}");
};
channel.BasicConsume(queue: "rpc_queue", autoAck: false, consumer: consumer);
Console.ReadLine();
}
}
}
private static int Fib(int n)
{
if (n == 0 || n == 1)
{
return n;
}
return Fib(n - 1) + Fib(n - 2);
}