淘先锋技术网

首页 1 2 3 4 5 6 7

.net使用RabbitMQ小记

使用RabbitMQ的优点

        1.性能全面，rabbitmq性能比较全面，是消息中间件的首选

        2.高并发，rabbitmq实现语言是天生就具备高并发高可用的erlang语言

        3.任务异步处理，将不需要同步处理的并且耗时长的操作由消息队列通知消息接受方进行异步处理，提高了应用程序的响应时间

        4.应用解耦，MQ相当于一个中介，生产方通过MQ与消费方交互，它将应用程序进行解耦

核心API接口和类

IModel：表示一个AMQP 通道，并提供大多数的操作

IConnection：表示一个AMQP连接

ConnectionFactory:：构造一个IConnection实例

IBasicConsumer：表示一个消息消费者

DefaultBasicConsumer：消费者常用的基类

限制

客户端并不支持无符号64位整型（ulong）,但支持有符号64位整型

使用简介

命名空间引用

using RabbitMQ.Client;

连接到RabbitMQ

ConnectionFactory factory = new ConnectionFactory();
factory.UserName = user;
factory.Password = pass;
factory.VirtualHost = vhost;
factory.HostName = hostName;

IConnection conn = factory.CreateConnection();

打开一个通道

IModel channel = conn.CreateModel();

断开连接  

channel.close();

conn.close();

关闭连接后，通道也会自动释放掉

connection 是一个长连接，底层协议针对长连接做了优化，这意味着打开一个新的连接将会有巨大的开销。channel 也是长连接，新建通道的开销会小很多，一般情况下，建议通道也复用，不要重复新建。

设置客户端名称（方便识别多个客户端，强烈建议设置）

factory.ClientProvidedName = "app:audit component:event-consumer";

定义交换机和队列，并绑定

channel.ExchangeDeclare(exchangeName, ExchangeType.Direct);
channel.QueueDeclare(queueName, false, false, false, null);
channel.QueueBind(queueName, exchangeName, routingKey, null);

直接删除

channel.QueueDelete("queue-name", false, false);

删除队列，如果为空

channel.QueueDelete("queue-name", false, true);

删除队列，如果没有使用（没有任何消费者）

channel.QueueDelete("queue-name", false, true);

删除队列并清除消息

channel.QueueDelete("queue-name");

发布消息

byte[] messageBodyBytes = System.Text.Encoding.UTF8.GetBytes("Hello, world!");
channel.BasicPublish(exchangeName, routingKey, null, messageBodyBytes);

对发布消息进行精细控制

byte[] messageBodyBytes = System.Text.Encoding.UTF8.GetBytes("Hello, world!");
IBasicProperties props = channel.CreateBasicProperties();
props.ContentType = "text/plain";
props.DeliveryMode = 2;
channel.BasicPublish(exchangeName, routingKey, props, messageBodyBytes);

发布消息并带有header 参数

byte[] messageBodyBytes = System.Text.Encoding.UTF8.GetBytes("Hello, world!");

IBasicProperties props = channel.CreateBasicProperties();
props.ContentType = "text/plain";
props.DeliveryMode = 2;
props.Headers = new Dictionary<string, object>();
props.Headers.Add("latitude",  51.5252949);
props.Headers.Add("longitude", -0.0905493);

channel.BasicPublish(exchangeName, routingKey, props, messageBodyBytes);

发布消息带有截至时间

byte[] messageBodyBytes = System.Text.Encoding.UTF8.GetBytes("Hello, world!");

IBasicProperties props = channel.CreateBasicProperties();
props.ContentType = "text/plain";
props.DeliveryMode = 2;
props.Expiration = "36000000";

channel.BasicPublish(exchangeName, routingKey, props, messageBodyBytes);

订阅消息

var consumer = new EventingBasicConsumer(channel);
consumer.Received += (ch, ea) =>
                {
                    var body = ea.Body.ToArray();
                    // copy or deserialise the payload
                    // and process the message
                    // ...
                    channel.BasicAck(ea.DeliveryTag, false);
                };
string consumerTag = channel.BasicConsume(queueName, false, consumer);

启动自动重连

factory.AutomaticRecoveryEnabled = true;

//默认重连时间5秒，这里设置为10s

factory.NetworkRecoveryInterval = TimeSpan.FromSeconds(10);