我要评论一、rabbitmq介绍
1、概念
rabbitmq是一个开源的消息中间件，用于在分布式系统中传递和存储消息。它实现了高级消息队列协议（amqp），提供了可靠的消息传递机制。
rabbitmq基于生产者-消费者模型，其中生产者将消息发送到队列中，而消费者从队列中接收并处理消息。它支持多种消息传递模式，包括点对点、发布/订阅和请求/响应模式。
2、特点
rabbitmq具有以下特点：
-
可靠性：rabbitmq使用持久化机制来确保消息不会丢失，并提供了消息确认机制，确保消息被正确地传递和处理。
-
灵活性：它支持多种消息传递模式和交换机类型，可以根据需求灵活配置和定制。
-
可扩展性：rabbitmq可以通过添加更多的节点来实现水平扩展，以处理更大的消息负载。
-
高性能：它采用异步消息传递机制，能够处理大量的消息并保持低延迟。
-
多语言支持：rabbitmq提供了多种编程语言的客户端库，使得开发者可以使用自己熟悉的语言进行消息传递。
-
可管理性：rabbitmq提供了一个易于使用的管理界面，可以监控和管理消息队列、交换机和绑定等。
rabbitmq广泛应用于各种场景，包括微服务架构、异步任务处理、日志收集和分析、实时数据处理等。它是一个可靠、灵活和高性能的消息中间件解决方案。
3、优势
rabbitmq与其他消息中间件相比，各有优缺点。下面是一些常见的消息中间件与rabbitmq的对比：
-
apache kafka：
- 优点：kafka适用于高吞吐量和低延迟的场景，特别擅长处理大规模的实时数据流。它具有高可扩展性和持久化能力，并支持分布式订阅和发布。
- 缺点：相对于rabbitmq而言，kafka的配置和部署较为复杂。此外，kafka不提供消息确认机制，因此在某些场景下可能会出现消息丢失的风险。
-
activemq：
- 优点：activemq是一个功能丰富的消息中间件，支持多种消息传递模式和协议。它易于使用和部署，并提供了可靠的消息传递机制。
- 缺点：相对于rabbitmq而言，activemq的性能稍逊一筹。在高负载情况下，activemq可能会出现性能瓶颈。
-
redis pub/sub：
- 优点：redis pub/sub是一个简单而高效的消息中间件，适用于快速发布和订阅消息。它具有低延迟和高并发性能。
- 缺点：redis pub/sub不支持持久化消息，当消费者离线时，无法保证消息的可靠传递。此外，它也不支持复杂的消息路由和过滤机制。
总体而言，rabbitmq在可靠性、灵活性和可管理性方面表现出色。它适用于大多数场景，并且具有广泛的语言支持和成熟的社区生态系统。然而，在需要高吞吐量和低延迟的大规模实时数据处理场景下，kafka可能更适合。选择合适的消息中间件取决于具体的需求和系统架构。
二、rabbitmq模式与应用场景介绍
1、订阅模式（publish/subscribe）
一次向许多消费者发送消息，一个生产者发送的消息会被多个消费者获取，也就是将消息将广播到所有的消费者中。
应用场景： 更新商品库存后需要通知多个缓存和多个数据库，这里的结构应该是：
一个fanout类型交换机扇出两个个消息队列，分别为缓存消息队列、数据库消息队列
一个缓存消息队列对应着多个缓存消费者
一个数据库消息队列对应着多个数据库消费者
发布者：
/// <summary>
/// 发布订阅模式--生产者
/// </summary>
public void fabudingyuesender()
{
string exchange_name = "logs";
_model.exchangedeclare("logs", "fanout");
string message = "publish subscribe message";
var body = encoding.utf8.getbytes(message);
_model.basicpublish(exchange: exchange_name,
routingkey: "",
basicproperties: null,
body: body
);
console.writeline(" [x] sent '" + message + "'");
}订阅者：
/// <summary>
/// 发布订阅模式--消费者1
/// </summary>
public void fabudingyuereceiver1()
{
string exchange_name = "logs";
_model.exchangedeclare(exchange: exchange_name, "fanout");
// 声明临时队列
var queuename = _model.queuedeclare().queuename;
// 将队列绑定到交换机
_model.queuebind(queue: queuename, exchange: exchange_name, routingkey: "");
// 创建消费者，绑定回调函数，只执行一次
var consumer = new eventingbasicconsumer(_model);
consumer.received += (model, ea) =>
{
var body = ea.body.toarray();
var message = encoding.utf8.getstring(body);
console.writeline("receiver1收到消息：{0}", message);
};
// 启动消费者
_model.basicconsume(queue: queuename, autoack: true, consumer: consumer);
}2、路由模式（routing）
有选择地（routing key）接收消息，发送消息到交换机并且要指定路由key ，消费者将队列绑定到交换机时需要指定路由key，仅消费指定路由key的消息
应用场景： 如在超市打折促销活动中增加了香蕉，香蕉促销活动消费者指定routing key为香蕉，只有香蕉促销活动会接收到消息，其它促销活动不关心也不会消费此routing key的消息
发布者：
/// <summary>
/// 路由模式 -- 生产者
/// </summary>
public void routingsender()
{
string exchange_name = "exchange_direct";
string exchange_type = "direct";
_model.exchangedeclare(exchange_name, exchange_type);
string key1 = "this is key1 message";
string key2 = "this is key2 message";
var body1 = encoding.utf8.getbytes(key1);
var body2 = encoding.utf8.getbytes(key2);
_model.basicpublish(exchange: exchange_name,
routingkey: "key1",
basicproperties: null,
body: body1);
_model.basicpublish(exchange: exchange_name,
routingkey: "key2",
basicproperties: null,
body: body2);
console.writeline($"push suessce:{key1}&{key2}");
}消费者1：只接收指定路由key1的
/// <summary>
/// 路由模式 -- 消费者1
/// </summary>
public void routingreceiver1() {
string exchange_name = "exchange_direct";
string exchange_type = "direct";
//说明交换机
_model.exchangedeclare(exchange_name, exchange_type);
// 声明临时队列
var queuename = _model.queuedeclare().queuename;
//绑定队列、交换机与路由（仅接收路由key1的）
_model.queuebind(queuename,exchange_name,"key1");
//设置消息qps
_model.basicqos(0,1,false);
// 创建消费者，绑定回调函数，只执行一次
var consumer = new eventingbasicconsumer(_model);
consumer.received += (model, ea) =>
{
var body = ea.body.toarray();
var message = encoding.utf8.getstring(body);
console.writeline("routingreceiver1收到消息：{0}", message);
};
// 启动消费者
_model.basicconsume(queue: queuename, autoack: true, consumer: consumer);
}消费者2：同时接收路由key1和key2的
/// <summary>
/// 路由模式 -- 消费者2
/// </summary>
public void routingreceiver2() {
string exchange_name = "exchange_direct";
// 声明临时队列
var queuename = _model.queuedeclare().queuename;
//绑定队列、交换机与路由（接收路由key1和key2）
_model.queuebind(queuename, exchange_name, "key1");
_model.queuebind(queuename, exchange_name, "key2");
//设置消息qps
//_model.basicqos(0, 1, false);
// 创建消费者，绑定回调函数，只执行一次
var consumer = new eventingbasicconsumer(_model);
consumer.received += (model, ea) =>
{
var body = ea.body.toarray();
var message = encoding.utf8.getstring(body);
console.writeline("routingreceiver2收到消息：{0}", message);
};
// 启动消费者
_model.basicconsume(queue: queuename, autoack: true, consumer: consumer);
}3、主题模式（topics）
根据主题（topics）来接收消息，将路由key和某模式进行匹配，此时队列需要绑定在一个模式上，#匹配一个词或多个词，*只匹配一个词。
应用场景： 同上，水果促销活动可以接收主题为火龙果的消息，如白肉火龙果、红肉火龙果等
3、远程过程调用（rpc）
如果我们需要在远程计算机上运行功能并等待结果就可以使用rpc，具体流程可以看图。应用场景：需要等待接口返回数据，如订单支付
发布者确认（publisher confirms）
与发布者进行可靠的发布确认，发布者确认是rabbitmq扩展，可以实现可靠的发布。在通道上启用发布者确认后，rabbitmq将异步确认发送者发布的消息，这意味着它们已在服务器端处理。
应用场景： 对于消息可靠性要求较高，比如钱包扣款
（后面会针对使用rabbitmq实现rpcclient端与server远程端通信的专题博客文章，敬请期待）
三、rabbitmq的交换机介绍
rabbitmq中的交换机（exchange）是消息的分发中心，它接收生产者发送的消息，并根据一定的规则将消息路由到一个或多个队列中。下面是对rabbitmq交换机的几种常见类型的说明：
-
直连交换机（direct exchange）：
- 直连交换机是最简单的交换机类型，它根据消息的路由键（routing key）将消息发送到与之匹配的队列。
- 生产者在发送消息时指定了一个路由键，交换机会将消息发送到与该路由键完全匹配的队列中。
-
主题交换机（topic exchange）：
- 主题交换机根据消息的路由键和绑定键（binding key）的模式匹配将消息发送到一个或多个队列中。
- 绑定键可以使用通配符进行模糊匹配，其中“*”表示匹配一个单词，“#”表示匹配零个或多个单词。
- 生产者在发送消息时指定了一个路由键，交换机会将消息发送到与该路由键匹配的队列中。
-
扇形交换机（fanout exchange）：
- 扇形交换机将消息广播到所有与之绑定的队列中，忽略消息的路由键。
- 当生产者发送消息到扇形交换机时，交换机会将消息复制并发送到所有与之绑定的队列中。
-
头交换机（headers exchange）：
- 头交换机根据消息的头部属性（headers）进行匹配，并将消息发送到与之匹配的队列中。
- 生产者在发送消息时可以设置自定义的头部属性，交换机会根据这些属性进行匹配。
交换机通过绑定（binding）与队列进行关联，一个交换机可以绑定多个队列。绑定时需要指定一个绑定键，交换机根据绑定键和消息的路由键来确定消息应该被发送到哪个队列。
不同类型的交换机适用于不同的消息路由需求，开发者可以根据具体的场景选择合适的交换机类型来实现灵活的消息路由和分发
四、实战项目演练
新增.net 6 web api 项目，项目结构如下图：
1、配置本地rabbitmq服务信息
修改appsettings.json文件，新增以下配置
"rabbitmq": {
"hostname": "localhost",
"port": "5672",
"username": "", //用户名
"password": "" //用户密码
}2、新增rabbitmqconfigs.cs
rabbitmqconfigs.cs是一个抽象基类，主要实现通过iconfiguration获取本地rabbitmq服务信息，创建rabbitmq连接工厂。
public abstract class rabbitmqconfigs
{
protected connectionfactory greateconnectionfactory(iconfiguration configuration)
{
// 在这里设置connectionfactory的属性
var factory = new connectionfactory
{
// 设置连接属性
hostname = configuration["rabbitmq:hostname"],
port = int.parse(configuration["rabbitmq:port"]),
username = configuration["rabbitmq:username"],
password = configuration["rabbitmq:password"]
};
return factory;
}
}3、新增irabbitmqservices接口
定义需要实现的rabbitmq的各种模式的接口，本文的该项目演示只针对发布订阅和路由模式去实践演示
public interface irabbitmqservices
{
/// <summary>
/// 发布订阅模式--生产者
/// </summary>
void fabudingyuesender();
/// <summary>
/// 发布订阅模式--消费者1
/// </summary>
void fabudingyuereceiver1();
/// <summary>
/// 发布订阅模式--消费者2
/// </summary>
void fabudingyuereceiver2();
/// <summary>
/// 路由模式--生产者
/// </summary>
void routingsender();
/// <summary>
/// 路由模式--消费者1
/// </summary>
void routingreceiver1();
/// <summary>
/// 路由模式--消费者2
/// </summary>
void routingreceiver2();
}4、新建rabbitmqservices
rabbitmqservices是基于rabbitmqconfigs抽象基类并继承irabbitmqservices的实现
public class rabbitmqservices : rabbitmqconfigs, irabbitmqservices
{
private readonly iconnection _connection;
private readonly imodel _model;
static concurrentdictionary<string, taskcompletionsource<string>> pendingrequests = new();
public rabbitmqservices(iconfiguration configuration)
{
var factory = greateconnectionfactory(configuration);
_connection = factory.createconnection();
_model = _connection.createmodel();
}
/// <summary>
/// 发布订阅模式--生产者
/// </summary>
public void fabudingyuesender()
{
string exchange_name = "logs";
_model.exchangedeclare("logs", "fanout");
string message = "publish subscribe message";
var body = encoding.utf8.getbytes(message);
_model.basicpublish(exchange: exchange_name,
routingkey: "",
basicproperties: null,
body: body
);
console.writeline(" [x] sent '" + message + "'");
}
/// <summary>
/// 发布订阅模式--消费者1
/// </summary>
public void fabudingyuereceiver1()
{
string exchange_name = "logs";
_model.exchangedeclare(exchange: exchange_name, "fanout");
// 声明临时队列
var queuename = _model.queuedeclare().queuename;
// 将队列绑定到交换机
_model.queuebind(queue: queuename, exchange: exchange_name, routingkey: "");
// 创建消费者，绑定回调函数，只执行一次
var consumer = new eventingbasicconsumer(_model);
consumer.received += (model, ea) =>
{
var body = ea.body.toarray();
var message = encoding.utf8.getstring(body);
console.writeline("receiver1收到消息：{0}", message);
};
// 启动消费者
_model.basicconsume(queue: queuename, autoack: true, consumer: consumer);
}
/// <summary>
/// 发布订阅模式--消费者2
/// </summary>
public void fabudingyuereceiver2()
{
string exchange_name = "logs";
//声明交换机
_model.exchangedeclare(exchange: exchange_name, "fanout");
// 声明临时队列
var queuename = _model.queuedeclare().queuename;
// 将队列绑定到交换机
_model.queuebind(queue: queuename, exchange: exchange_name, routingkey: "");
// 创建消费者，绑定回调函数，只执行一次
var consumer = new eventingbasicconsumer(_model);
consumer.received += (model, ea) =>
{
var body = ea.body.toarray();
var message = encoding.utf8.getstring(body);
console.writeline("receiver2收到消息：{0}", message);
};
// 启动消费者
_model.basicconsume(queue: queuename, autoack: true, consumer: consumer);
}
/// <summary>
/// 路由模式 -- 生产者
/// </summary>
public void routingsender()
{
string exchange_name = "exchange_direct";
string exchange_type = "direct";
_model.exchangedeclare(exchange_name, exchange_type);
string key1 = "this is key1 message";
string key2 = "this is key2 message";
var body1 = encoding.utf8.getbytes(key1);
var body2 = encoding.utf8.getbytes(key2);
_model.basicpublish(exchange: exchange_name,
routingkey: "key1",
basicproperties: null,
body: body1);
_model.basicpublish(exchange: exchange_name,
routingkey: "key2",
basicproperties: null,
body: body2);
console.writeline($"push suessce:{key1}&{key2}");
}
/// <summary>
/// 路由模式 -- 消费者1
/// </summary>
public void routingreceiver1() {
string exchange_name = "exchange_direct";
string exchange_type = "direct";
//说明交换机
_model.exchangedeclare(exchange_name, exchange_type);
// 声明临时队列
var queuename = _model.queuedeclare().queuename;
//绑定队列、交换机与路由（仅接收路由key1的）
_model.queuebind(queuename,exchange_name,"key1");
//设置消息qps
_model.basicqos(0,1,false);
// 创建消费者，绑定回调函数，只执行一次
var consumer = new eventingbasicconsumer(_model);
consumer.received += (model, ea) =>
{
var body = ea.body.toarray();
var message = encoding.utf8.getstring(body);
console.writeline("routingreceiver1收到消息：{0}", message);
};
// 启动消费者
_model.basicconsume(queue: queuename, autoack: true, consumer: consumer);
}
/// <summary>
/// 路由模式 -- 消费者2
/// </summary>
public void routingreceiver2() {
string exchange_name = "exchange_direct";
// 声明临时队列
var queuename = _model.queuedeclare().queuename;
//绑定队列、交换机与路由（接收路由key1和key2）
_model.queuebind(queuename, exchange_name, "key1");
_model.queuebind(queuename, exchange_name, "key2");
//设置消息qps
//_model.basicqos(0, 1, false);
// 创建消费者，绑定回调函数，只执行一次
var consumer = new eventingbasicconsumer(_model);
consumer.received += (model, ea) =>
{
var body = ea.body.toarray();
var message = encoding.utf8.getstring(body);
console.writeline("routingreceiver2收到消息：{0}", message);
};
// 启动消费者
_model.basicconsume(queue: queuename, autoack: true, consumer: consumer);
}
}5、apicontroller.cs 实现api请求，模拟发布消息
// post api/<apicontroller>
[httppost]
public iactionresult post(string value)
{
// 发布消息到 rabbitmq
_rabbitmqservices.fabudingyuesender();
_rabbitmqservices.routingsender();
return ok();
}五、测试
新建一个控制台应用程序，修改program.cs文件，实现对mq消息的监听消费的逻辑。如下
using rabbirmqtestreciver;
using rabbitmq.client;
using rabbitmq.client.events;
using system.text;
iconnection _connection = null;
imodel _model = null;
var factory = new connectionfactory
{
// 设置连接属性
hostname = "localhost",
port = 5672,
username = "",//用户名
password = "" //用户密码
};
_connection = factory.createconnection();
_model = _connection.createmodel();
//发布订阅模式的消费者
fabudingyuereceiver1();
fabudingyuereceiver2();
//路由模式的消费者（指定接收路由key1）
routingreceiver1();
//路由模式的消费者（指定接收路由key1和key2）
routingreceiver2();
console.writeline("按任意键退出...");
console.readkey();
void fabudingyuereceiver1()
{
string exchange_name = "logs";
_model.exchangedeclare(exchange: exchange_name, "fanout");
// 声明临时队列
var queuename = _model.queuedeclare().queuename;
// 将队列绑定到交换机
_model.queuebind(queue: queuename, exchange: exchange_name, routingkey: "");
// 创建消费者，绑定回调函数，只执行一次
var consumer = new eventingbasicconsumer(_model);
consumer.received += (model, ea) =>
{
var body = ea.body.toarray();
var message = encoding.utf8.getstring(body);
console.writeline("fabudingyuereceiver1收到消息：{0}", message);
};
// 启动消费者
_model.basicconsume(queue: queuename, autoack: true, consumer: consumer);
}
/// <summary>
/// 发布订阅模式--消费者2
/// </summary>
void fabudingyuereceiver2()
{
string exchange_name = "logs";
//声明交换机
_model.exchangedeclare(exchange: exchange_name, "fanout");
// 声明临时队列
var queuename = _model.queuedeclare().queuename;
// 将队列绑定到交换机
_model.queuebind(queue: queuename, exchange: exchange_name, routingkey: "");
// 创建消费者，绑定回调函数，只执行一次
var consumer = new eventingbasicconsumer(_model);
consumer.received += (model, ea) =>
{
var body = ea.body.toarray();
var message = encoding.utf8.getstring(body);
console.writeline("fabudingyuereceiver2收到消息：{0}", message);
};
// 启动消费者
_model.basicconsume(queue: queuename, autoack: true, consumer: consumer);
}
/// <summary>
/// 路由模式 -- 消费者1
/// </summary>
void routingreceiver1()
{
string exchange_name = "exchange_direct";
string exchange_type = "direct";
//说明交换机
_model.exchangedeclare(exchange_name, exchange_type);
// 声明临时队列
var queuename = _model.queuedeclare().queuename;
//绑定队列、交换机与路由（仅接收路由key1的）
_model.queuebind(queuename, exchange_name, "key1");
//设置消息qps
_model.basicqos(0, 1, false);
// 创建消费者，绑定回调函数，只执行一次
var consumer = new eventingbasicconsumer(_model);
consumer.received += (model, ea) =>
{
var body = ea.body.toarray();
var message = encoding.utf8.getstring(body);
console.writeline("routingreceiver1收到消息：{0}", message);
};
// 启动消费者
_model.basicconsume(queue: queuename, autoack: true, consumer: consumer);
}
void routingreceiver2()
{
string exchange_name = "exchange_direct";
// 声明临时队列
var queuename = _model.queuedeclare().queuename;
//绑定队列、交换机与路由（接收路由key1和key2）
_model.queuebind(queuename, exchange_name, "key1");
_model.queuebind(queuename, exchange_name, "key2");
//设置消息qps
//_model.basicqos(0, 1, false);
// 创建消费者，绑定回调函数，只执行一次
var consumer = new eventingbasicconsumer(_model);
consumer.received += (model, ea) =>
{
var body = ea.body.toarray();
var message = encoding.utf8.getstring(body);
console.writeline("routingreceiver2收到消息：{0}", message);
};
// 启动消费者
_model.basicconsume(queue: queuename, autoack: true, consumer: consumer);
}
启动两个项目，调用发布项目的api接口模拟发布消息,查看控制台输出结果。
结果：
发表评论