|
1、点对点模式（一对一）：主要就是一个生产者对应一个消费者；(服务端定义的队列)
2、工作队列模式：即一对多，一个生产者产生的数据可以被多个消费者消费，但各个消费者获取的数据是不一致的，应该消息都是经过一个队列中获取的；(服务端定义的队列)
3、发布订阅模式：引入了交换机exchange，生产者产生的数据发布到exchange中，exchange分发到不同的队列中，每一个队列里面的数据都是一模一样的，给多个消费者消费；
????????对应exchange交换机中的fanout(扇形交换机)
? ? ? ? 服务端发送消息时自动生成队列(千万别自己定义队列,否者客户端接受的消息不是广播的)
? ? ? ? 客户端根据订阅的交换机(队列)接受消息
4、routing模式：在pub/sub的基础上引入了routingkey的概念，即exchange会按照routingkey把消息分发到不同的队列中，给不同的消费者使用；
????????对应exchange交换机中的direct(直连交换机)????????
????????服务端定义路由
????????客户端根据自己绑定的路由接受消息
5、topics模式：在routing模式的基础上对routingkey 加入了??*??和?#??的概念，*? 代表 单字符串匹配，#? 代表 多字符串匹配，如 key为 beijing.chaoyang.20200312 =>? *.*.*.20200312 或者 #.20200312 ；(对应exchange交换机中的topic(主题交换机) 客户端定义队列,客户端根据自己绑定的路由接受消息)
?????????对应exchange交换机中的direct(直连交换机)??
????????服务端定义路由
????????客户端根据自己绑定的路由(通配符)接受消息
6、RPC模式：针对生产者产生的数据，消费者消费后会给生产者一个反馈，即远程过程调用。(这个在代码中没有体现,请自行编码体验)
using RabbitMQ.Client;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
namespace RabbitMqCommon
{
public class ConnectionUtils
{
public static IConnection GetConnection()
{
var factory = new ConnectionFactory();
//连接主机名
factory.HostName = "localhost";
//mqtt用户需要在控制台和web服务器中配置
//连接用户名
factory.UserName = "admin";
//连接密码
factory.Password = "123456";
//连接端口
factory.Port = 5672;
return factory.CreateConnection();
}
}
}
using RabbitMQ.Client;
using RabbitMqCommon;
using System;
using System.Collections.Generic;
using System.Collections.Specialized;
using System.ComponentModel;
using System.Data;
using System.Drawing;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using System.Windows.Forms;
namespace RabbitMqProduct
{
public partial class FrmProduct : Form
{
public FrmProduct()
{
InitializeComponent();
}
/// <summary>
/// rabbitmq工作队列
/// </summary>
/// <param name="sender"></param>
/// <param name="e"></param>
private void btnSendPoint_Click(object sender, EventArgs e)
{
if (!string.IsNullOrEmpty(rtbsend.Text))
{
try
{
//需要创建一个ConnectionFactory，设置目标，由于是在本机，
//所以设置为localhost，如果RabbitMQ不在本机，
//只需要设置目标机器的IP地址或者机器名称即可
// ，然后设置前面创建的用户名和密码。
//1.声明MessageQueue
//在Rabbit MQ中，无论是生产者发送消息还是消费者接受消息，都首先需要声明一个MessageQueue。这就存在一个问题，是生产者声明还是消费者声明呢？要解决这个问题，首先需要明确：
//a)消费者是无法订阅或者获取不存在的MessageQueue中信息。
//b)消息被Exchange接受以后，如果没有匹配的Queue，则会被丢弃。
var connection = ConnectionUtils.GetConnection();
#region 简单队列模式 一个生产者、一个消费者，不需要设置交换机（使用默认的交换机，一个direct类型的交换机,routing_key为queue名称）
//创建一个发布消息的的通道
using (var channel = connection.CreateModel())
{
//创建一个名称为hello的消息队列
channel.QueueDeclare("hello", true, false, false, null);
//我目前没有看出来有什么区别 加没加这个 客户端接收消息也是一个接一个地接收数据
channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: false);
//发送的数据内容
string message = rtbsend.Text;
//将发送的数据转成字节数组
var body = Encoding.UTF8.GetBytes(message);
//自定义一个消息头header
IBasicProperties basicProperties = channel.CreateBasicProperties();
IDictionary<string, object> dic = new Dictionary<string, object>();
dic.Add("param_code", "1234");
dic.Add("param_pass", "1234");
basicProperties.Headers = dic;
//basicProperties.DeliveryMode = 2;
//basicProperties.Persistent = false; 这个属性就是用来设置上面DeliveryMode的
//我们可以确认即使RabbitMQ重启，我们的task_queue队列也不会丢失。
//接下来，我们需要将IBasicProperties.SetPersistent设置为true，用来将我们的消息标示成持久化的。
//basicProperties.Persistent = true;
//开始传递
channel.BasicPublish("", "hello", basicProperties, body);
}
#endregion
}
catch (Exception ex)
{
throw ex;
}
}
}
/// <summary>
///基于exchange的fanout的工作队列 扇形交换机(fanout)
///总结:no routing 根据绑定的全路径发布消息
/// </summary>
/// <param name="sender"></param>
/// <param name="e"></param>
private void btnFanout_Click(object sender, EventArgs e)
{
if (!string.IsNullOrEmpty(rtbsend.Text))
{
try
{
#region Publish/Subscribe 发布/订阅模式 Fanout
var connection = ConnectionUtils.GetConnection();
//创建一个发布消息的的通道
using (var channel = connection.CreateModel())
{
var exchangestr = "exchange_fanout";
//创建一个名称为hello的消息队列
channel.ExchangeDeclare(exchangestr, ExchangeType.Fanout, true, false, null);
//发送的数据内容
string message = rtbsend.Text;
//将发送的数据转成字节数组
var body = Encoding.UTF8.GetBytes(message);
//自定义一个消息头header
IBasicProperties basicProperties = channel.CreateBasicProperties();
IDictionary<string, object> dic = new Dictionary<string, object>();
dic.Add("param_code", "1234");
dic.Add("param_pass", "1234");
basicProperties.Headers = dic;
//开始传递
channel.BasicPublish(exchangestr, "", basicProperties, body);
}
#endregion
}
catch (Exception ex)
{
throw ex;
}
}
}
/// <summary>
///基于exchange的topic的工作队列 主题交换机(topic)
///总结:根据绑定通配符routing的路径来发布消息
/// </summary>
/// <param name="sender"></param>
/// <param name="e"></param>
private void btnExChange_Click(object sender, EventArgs e)
{
if (!string.IsNullOrEmpty(rtbsend.Text))
{
var connection = ConnectionUtils.GetConnection();
#region 涉及交换机的发送和接收 Topic类型 处理routingKey和bindingKey，支持通配符。# 匹配0或多个单词，* 匹配单个单词。 Topic主题模式可以实现 Publish/Subscribe发布订阅模式 和 Routing路由模式 的双重功能
//创建一个发布消息的的通道
using (var channel = connection.CreateModel())
{
//声明交换机- channel.exchangeDeclare(交换机名字,交换机类型)
var exchangestr = "exchange_topic";
channel.ExchangeDeclare(exchangestr, ExchangeType.Topic);
IBasicProperties basicProperties = channel.CreateBasicProperties();
IDictionary<string, object> dic = new Dictionary<string, object>();
dic.Add("param_code", "1234topic");
dic.Add("param_pass", "1234topic");
basicProperties.Headers = dic;
basicProperties.DeliveryMode = 1;
for (int i = 0; i < 4; i++)
{
//发送消息的时候根据相关逻辑指定相应的routing key。
string routekey = string.Empty;
switch (i)
{
case 0:
routekey = "log.success";
break;
case 1:
routekey = "log.error";
break;
case 2:
routekey = "log.info";
break;
case 3:
routekey = "log.waring";
break;
}
//内容
string message = rtbsend.Text + i;
var body = Encoding.UTF8.GetBytes(message);
//开始传递
channel.BasicPublish(exchangestr, routekey, basicProperties, body);
}
}
#endregion
}
}
/// <summary>
/// rabbitmq主题模式 对应exchange的direct(直连交换机)
/// </summary>
/// <param name="sender"></param>
/// <param name="e"></param>
private void btnDirect_Click(object sender, EventArgs e)
{
if (!string.IsNullOrEmpty(rtbsend.Text))
{
var connection = ConnectionUtils.GetConnection();
#region 涉及交换机的发送和接收 Topic类型 处理routingKey和bindingKey，支持通配符。# 匹配0或多个单词，* 匹配单个单词。 Topic主题模式可以实现 Publish/Subscribe发布订阅模式 和 Routing路由模式 的双重功能
//创建一个发布消息的的通道
using (var channel = connection.CreateModel())
{
//声明交换机- channel.exchangeDeclare(交换机名字,交换机类型)
var exchangestr = "exchange_direct";
channel.ExchangeDeclare(exchangestr, ExchangeType.Direct, true, false, null);
IBasicProperties basicProperties = channel.CreateBasicProperties();
IDictionary<string, object> dic = new Dictionary<string, object>();
dic.Add("param_code", "1234direct");
dic.Add("param_pass", "1234direct");
basicProperties.Headers = dic;
basicProperties.DeliveryMode = 1;
for (int i = 0; i < 4; i++)
{
//发送消息的时候根据相关逻辑指定相应的routing key。
string routekey = string.Empty;
switch (i)
{
case 0:
routekey = "success";
break;
case 1:
routekey = "error";
break;
case 2:
routekey = "info";
break;
case 3:
routekey = "waring";
break;
}
//内容
string message = rtbsend.Text + i;
var body = Encoding.UTF8.GetBytes(message);
//开始传递
channel.BasicPublish(exchangestr, routekey, basicProperties, body);
}
}
#endregion
}
}
}
}
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using RabbitMqCommon;
using System;
using System.Collections.Generic;
using System.ComponentModel;
using System.Data;
using System.Drawing;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using System.Windows.Forms;
namespace RabbitMqConsume
{
public partial class FrmConsume : Form
{
public FrmConsume()
{
InitializeComponent();
CheckForIllegalCrossThreadCalls = false;
}
/*
在通配符模式下，RabbitMQ使用模糊匹配来决定把消息推送给哪个生产者。通配符有两个符号来匹配routingKey
*匹配一个字符 如:*.qq.com 可匹配 1.qq.com
#匹配一个或者多个字符。 如：*.qq.com 可匹配 1.qq.com或者1111.qq.com
其他的操作基本和routing模式一样。
header模式
header模式是把routingkey放到header中.取消掉了routingKey。并使用一个字典传递 K、V的方式来匹配。
比如同时要给用户发送邮件和短信，可直接通过header的键值对来匹配绑定的值，把消息传递给发短信和邮件的生产者.
*/
private void Form1_Load(object sender, EventArgs e)
{
}
/// <summary>
/// 工作队列模式
/// </summary>
/// <param name="sender"></param>
/// <param name="e"></param>
private void btnPointReceive_Click(object sender, EventArgs e)
{
try
{
//创建rabbitmq一个连接
var connection = ConnectionUtils.GetConnection();
#region 点对点模式接受到的消息
//创建一个接受消息的通道
var channel = connection.CreateModel();
//创建一个名称为hello的消息队列 如果存在则不会创建
channel.QueueDeclare("hello", true, false, false, null);
var consumer = new EventingBasicConsumer(channel);
channel.BasicConsume("hello", false, consumer);
consumer.Received += (model, ea) =>
{
var headers = ea.BasicProperties.Headers;
if (headers != null)
{
foreach (var item in headers)
{
this.rtbmsg.AppendText($"key:{item.Key},value:{Encoding.UTF8.GetString(item.Value as byte[])}\n");
}
}
var body = ea.Body.ToArray();
//内容
var message = Encoding.UTF8.GetString(body);
this.rtbmsg.AppendText(message + "\n");
//手动确认消息
channel.BasicAck(ea.DeliveryTag, false);
};
#endregion
}
catch (Exception ex)
{
throw new Exception(ex.Message + ex.StackTrace.ToString());
}
}
/// <summary>
///基于exchange的fanout的工作队列 扇形交换机(fanout)
///总结:no routing 只要绑定该队列就接受服务端发布的消息
///注意:在扇形交换机中不需要设置自定义的队列 否则会造成和工作组
/// </summary>
/// <param name="sender"></param>
/// <param name="e"></param>
private void btnFanoutReceive_Click(object sender, EventArgs e)
{
#region fanout多播模式，也称为Publish/Scribe模式 就是把交换机（Exchange）里的消息发送给所有绑定该交换机的队列，忽略routingKey。
//type: ExchangeType.Fanout 需要先启动消费者 不然会有问题
//Publish/Subscribe 发布/订阅模式 使用的exchange 交换机的模式 只有一个exchange 在接收消息的时候使用的是临时创建的队列
//注意:不要这样使用 using(var connection = ConnectionUtils.GetConnection()) 这样跑完之后连连接对象都释放掉了 当然接受不到消息的
//并不是所有优雅的代码都是这么写的 有的时候还是需要考虑下逻辑问题 该不该这样写 不要一根筋
//有的时候可能点的快 突然一下子能接受到一条 或几条消息 不要这样写
var connection = ConnectionUtils.GetConnection();
//创建一个发布消息的的通道 通道不能释放 如果通道释放了 连临时队列都没有了
var channel = connection.CreateModel();
//扇形交换机 存放数据的队列 可以使用临时队列 但是那样的话就有可能导致消息没有了 不是持久化的消息
var exchangestr = "exchange_fanout";
var queueName = channel.QueueDeclare().QueueName;
//声明交换机与交换机类型
channel.ExchangeDeclare(exchange: exchangestr, type: ExchangeType.Fanout, true, false, null);
//队列绑定交换机与路由key
//队列绑定交换机-channel.queueBind(队列名, 交换机名, 路由key[广播消息设置为空串])
//可以绑定多个 绑定完了之后 只要是想这个名称的交换机的队列中发送消息
//消费者只要是绑定多这个交换机的路由,那么这个队列里面就能接受到服务端发布的多个消息
channel.QueueBind(queueName, exchangestr, "");
//9、声明队列-channel.queueDeclare(名称，是否持久化，是否独占本连接,是否自动删除,附加参数)
var consumer = new EventingBasicConsumer(channel);
//消费方法，这些方法是由服务端主动PUSH消息过来，方法接收到消息后进行处理
//我将下面的处理方法设置成 不自动回复处理 可以使用channel.BasicAck 手动确认消息是否已经接收
channel.BasicConsume(queueName, false, consumer);
consumer.Received += (model, ea) =>
{
var headers = ea.BasicProperties.Headers;
if (headers != null)
{
foreach (var item in headers)
{
this.rtbmsg.AppendText($"key:{item.Key},value:{Encoding.UTF8.GetString(item.Value as byte[])}\n");
}
}
var body = ea.Body.ToArray();
//内容
var message = Encoding.UTF8.GetString(body);
this.rtbmsg.AppendText($"Exchange:{ea.Exchange},DeliveryTag:{ea.DeliveryTag },RoutingKey:{ ea.RoutingKey},message:{ message}\n");
//手动确认消息
channel.BasicAck(ea.DeliveryTag, true);
};
#endregion
}
/// <summary>
///基于exchange的topic的工作队列 主题交换机(topic)
///总结:routing 根据绑定的routing的通配符匹配路径接受服务端发布的消息
/// </summary>
/// <param name="sender"></param>
/// <param name="e"></param>
private void btnExchangeReceive_Click(object sender, EventArgs e)
{
#region 涉及交换机的接收 Topic
//创建rabbitmq一个连接
var connection = ConnectionUtils.GetConnection();
//创建一个发布消息的的通道
var channel = connection.CreateModel();
//这个是存放消息的一个队列名称
var exchangestr = "exchange_topic";
var queueName = "queue_topic";
//声明交换机与交换机类型
channel.ExchangeDeclare(exchange: exchangestr, type: ExchangeType.Topic);
//9、声明队列-channel.queueDeclare(名称，是否持久化，是否独占本连接,是否自动删除,附加参数)
channel.QueueDeclare(queueName, true, false, false, null);
//队列绑定交换机与路由key
//队列绑定交换机-channel.queueBind(队列名, 交换机名, 路由key[广播消息设置为空串])
//可以绑定多个 绑定完了之后 只要是想这个名称的交换机的队列中发送消息
//消费者只要是绑定多这个交换机的路由,那么这个队列里面就能接受到服务端发布的消息
channel.QueueBind(queueName, exchangestr, "log.*");
//要改变这种行为的话，我们可以在BasicQos方法中设置prefetchCount = 1。
//这样会告诉RabbitMQ一次不要给同一个worker提供多于一条的信息。
//话句话说，在一个工作者还没有处理完消息，并且返回确认标志之前，不要再给它调度新的消息。
//取而代之，它会将消息调度给下一个不再繁忙的工作者。
channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: false);
var consumer = new EventingBasicConsumer(channel);
channel.BasicConsume(queueName, false, consumer);
consumer.Received += (model, ea) =>
{
var headers = ea.BasicProperties.Headers;
if (headers != null)
{
try
{
foreach (var item in headers)
{
this.rtbmsg.AppendText($"key:{item.Key},value:{Encoding.UTF8.GetString(item.Value as byte[])}\n");
}
var body = ea.Body.ToArray();
//内容
var message = Encoding.UTF8.GetString(body);
this.rtbmsg.AppendText($"Exchange:{ea.Exchange},DeliveryTag:{ea.DeliveryTag },RoutingKey:{ ea.RoutingKey},message:{ message}\n");
//手动确认消息
channel.BasicAck(ea.DeliveryTag, true);
}
catch (Exception ex)
{
throw new Exception(ex.Message + ex.StackTrace.ToString());
}
};
};
#endregion
}
/// <summary>
/// 基于exchange的direct的工作队列 直连交换机(direct)
/// 总结:routing 根据绑定的routing的名称(全路径)接受服务端发布的消息
/// </summary>
/// <param name="sender"></param>
/// <param name="e"></param>
private void btnDirect_Click(object sender, EventArgs e)
{
#region 涉及交换机的接收 Direct
//创建rabbitmq一个连接
var connection = ConnectionUtils.GetConnection();
//创建一个发布消息的的通道
var channel = connection.CreateModel();
//这个是存放消息的一个队列名称
var exchangestr = "exchange_direct";
var queueName = "queue_direct";
//声明交换机与交换机类型
channel.ExchangeDeclare(exchange: exchangestr, type: ExchangeType.Direct, true, false, null);
//9、声明队列-channel.queueDeclare(名称，是否持久化，是否独占本连接,是否自动删除,附加参数)
channel.QueueDeclare(queueName, true, false, false, null);
//队列绑定交换机与路由key
//队列绑定交换机-channel.queueBind(队列名, 交换机名, 路由key[广播消息设置为空串])
//可以绑定多个 绑定完了之后 只要是想这个名称的交换机的队列中发送消息
//消费者只要是绑定多这个交换机的路由,那么这个队列里面就能接受到服务端发布的消息
channel.QueueBind(queueName, exchangestr, "success");
channel.QueueBind(queueName, exchangestr, "error");
channel.QueueBind(queueName, exchangestr, "info");
//channel.QueueBind(queueName, exchangestr, "waring");
//要改变这种行为的话，我们可以在BasicQos方法中设置prefetchCount = 1。
//这样会告诉RabbitMQ一次不要给同一个worker提供多于一条的信息。
//话句话说，在一个工作者还没有处理完消息，并且返回确认标志之前，不要再给它调度新的消息。
//取而代之，它会将消息调度给下一个不再繁忙的工作者。
channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: false);
var consumer = new EventingBasicConsumer(channel);
channel.BasicConsume(queueName, false, consumer);
consumer.Received += (model, ea) =>
{
var headers = ea.BasicProperties.Headers;
if (headers != null)
{
try
{
foreach (var item in headers)
{
this.rtbmsg.AppendText($"key:{item.Key},value:{Encoding.UTF8.GetString(item.Value as byte[])}\n");
}
var body = ea.Body.ToArray();
//内容
var message = Encoding.UTF8.GetString(body);
this.rtbmsg.AppendText($"Exchange:{ea.Exchange},DeliveryTag:{ea.DeliveryTag },RoutingKey:{ ea.RoutingKey},message:{ message}\n");
//手动确认消息
channel.BasicAck(ea.DeliveryTag, true);
}
catch (Exception ex)
{
throw new Exception(ex.Message + ex.StackTrace.ToString());
}
};
};
#endregion
}
}
}
代码仓库地址:https://gitee.com/javascripts/rabbitmq.git
rabbitmq中文文档直达车:http://rabbitmq.mr-ping.com
|