下载安装
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);
}
