[Open Source] RabbitMQ 安装与使用

时间 2019-12-09

标签 open source rabbitmq 安装使用栏目 RabbitMQ 繁體版

原文原文链接

前言

吃多了拉就是队列，吃饱了吐就是栈html

使用场景
- 对操做的实时性要求不高，而须要执行的任务极为耗时；(发送短信，邮件提醒，更新文章阅读计数，记录用户操做日志)
- 存在异构系统间的整合；

安装

下载 Erlang
- 安装完肯定ERLANG_HOME环境变量是否添加,不然：Setx ERLANG_HOME “D:\Program Files\erl8.2″
下载安装包
- 安装完经过rabbitmqctl status肯定rabbitmq状态
管理服务
- 默认安装成功会自动启动服务
- 经过开始菜单能够启动,中止,卸载服务
占用端口
- 4369(集群、Erlang)
- 5671,5672(应用层标准高级消息队列协议)
- 25672(Erlang分发,CLI通讯)
- 15672(若是管理插件启用)
- 61613,61614(若是消息文本协议STOMP已启用)
- 1883,8883(若是erl实时通讯已启用)
支持的平台
- 基于Ubuntu和Debian的Linux发行版
- 基于Fedora，CentOS和RPM的Linux发行版
- Mac OS X
- Windows XP及更高版本

概念

Connections：客户端链接，建立该资源很是耗时，应尽可能避免屡次建立。
Channel：消息通道，在客户端的每一个链接里，可创建多个channel，每一个channel表明一个会话任务。
Exchange：消息交换机，它指定消息按什么规则，路由到哪一个队列。
Queue：消息队列载体，每一个消息都会被投入到一个或多个队列。c#
Broker：简单来讲就是消息队列服务器实体。
Binding：绑定，它的做用就是把exchange和queue按照路由规则绑定起来。
Routing Key：路由关键字，exchange根据这个关键字进行消息投递。
vhost：虚拟主机，一个broker里能够开设多个vhost，用做不一样用户的权限分离。windows
producer：消息生产者，就是投递消息的程序。
consumer：消息消费者，就是接收消息的程序。服务器

消息队列的发送过程大概以下：

客户端建立Connection,链接到消息队列服务器，打开一个channel。
客户端声明一个Exchange，并设置相关属性。
客户端声明一个Queue，并设置相关属性。
客户端使用routing key，在exchange和queue之间创建好绑定关系。
客户端发送消息首先到exchange
exchange根据type路由到对应的队列(能够是多个队列)中.

Exchange Type

direct(直连)
- routing key 与 binding key相同
fanout
- 给全部绑定队列发送消息
topic
- routing key:audit.irs.corporate => binding key:audit.#
- routing key:audit.irs => binding key:audit.*
default
- direct
- binding key为queue名称

经常使用命令

管理插件
- rabbitmq-plugins enable rabbitmq_management // 启用
- rabbitmq-plugins disable rabbitmq_management // 禁用
管理队列
- rabbitmqctl list_queues // 查看队列
管理用户及权限
- rabbitmqctl list_users // 查看全部用户
- rabbitmqctl add_user user_admin passwd_admin // 添加用户
- rabbitmqctl set_user_tags user_admin administrator // 添加权限
- rabbitmqctl delete_user guest // 删除用户
- rabbitmqctl change_password {username} {newpassowrd} // 修改密码
管理虚拟主机vhost
- rabbitmqctl add_vhost vhostpath // 建立虚拟主机
- rabbitmqctl delete_vhost vhostpath // 删除虚拟主机
- rabbitmqctl list_vhosts // 列出全部虚拟主机

使用

发送消息(以持久化代码为例)

var factory = new ConnectionFactory
{
    HostName = hostName,                // rabbit server
    UserName = "admin",
    Password = "admin",
    Port = 5672,                        // Broker端口
    VirtualHost = "/"                   // 虚拟Host,需提早配置
};

using (var connection = factory.CreateConnection()) // 建立与RabbitMQ服务器的链接
{
    using (var channel = connection.CreateModel())  // 建立1个Channel(大部分API在该Channel中)
    {
        // 定义1个队列,自动会和默认的exchange 作direct类型绑定
        channel.QueueDeclare(
            queue: "hello",                     // 队列名称
            durable: true,                      // 队列是否持久化
            exclusive: false,                   // 排他队列:若是一个队列被声明为排他队列，该队列仅对首次声明它的链接可见，并在链接断开时自动删除。(活动在一次链接内)
            autoDelete: false,                  // 自动删除:当最后一个消费者取消订阅时，队列自动删除。若是您须要仅由一个使用者使用的临时队列，请将自动删除与排除。当消费者断开链接时，队列将被删除。(至少消费者能连一次)
            arguments: null);                   // 配置参数

        var randomQueue = channel.QueueDeclare();                   // 定义随机的队列 该队列为临时队列(排他队列 + 自动删除)


        // 定义Exchange(通常而言,不须要定义exchange,rabbitmq默认建立了全部类型的exchange)
        //channel.ExchangeDeclare("direct-demo", ExchangeType.Direct);    // 定义direct exchange
        //channel.ExchangeDeclare("fannout-demo", ExchangeType.Fanout);   // 定义fanout exchange
        //channel.ExchangeDeclare("topic-demo", ExchangeType.Topic);      // 定义fanout exchange


        // 定义queue exchange key 关系(在某些业务场景下,会使用该关系作路由功能)
        //channel.QueueBind(queue: "hello", exchange: "amq.direct", routingKey: "hello"); // 默认绑定的关系和该行代码效果同样
        //channel.QueueBind("hello", "amq.fanout", "hello");                              // 该类型下的routingKey 实际不须要

        var properties = channel.CreateBasicProperties();
        properties.Persistent = true;

        while (true)
        {
            string message = "Hello World!" + DateTime.Now;
            var body = Encoding.UTF8.GetBytes(message);

            // 发送消息到队列中
            channel.BasicPublish(
                exchange: string.Empty,         // 传递为Empty的时候,经过   `(AMQP default)`传递
                routingKey: "hello",            // routing key 与 queuebind中的binding key对应
                basicProperties: properties,    // 消息header
                body: body);                    // 消息body:发送的是bytes 能够任意编码

            Console.WriteLine(" [x] Sent {0}", message);
        }
    }
}

接收消息(以消息响应为例)

var factory = new ConnectionFactory
{
    HostName = hostName,                // rabbit server
    UserName = "admin",
    Password = "admin",
    Port = 5672,                        // Broker端口
    VirtualHost = "/"                   // 虚拟Host,需提早配置
};
using (var connection = factory.CreateConnection())
{
    using (var channel = connection.CreateModel())
    {
        var consumer = new EventingBasicConsumer(channel);  // 建立Consumer
        consumer.Received += (model, ea) =>         // 经过回调函数异步推送咱们的消息
        {
            var body = ea.Body;
            var message = Encoding.UTF8.GetString(body);
            Thread.Sleep(1000);
            channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false); // 消息响应
            Console.WriteLine(" [x] Received {0}", message);
        };

        channel.BasicQos(0, 1, false);  // 设置perfetchCount=1 。这样就告诉RabbitMQ 不要在同一时间给一个工做者发送多于1个的消息
        channel.BasicConsume(queue: "hello",
                                noAck: false,      // 须要消息响应（Acknowledgments）机制
                                consumer: consumer);

        Console.WriteLine(" Press [enter] to exit.");
        Console.ReadLine();
    }
}

消息响应(acknowledgments)
- 为了防止消息丢失，RabbitMQ提供了消息响应（acknowledgments）机制。消费者会经过一个ack（响应），告诉RabbitMQ已经收到并处理了某条消息，而后RabbitMQ才会释放并删除这条消息。
持久化
- 新队列(没法修改队列)配置为可持久化
- 发送消息配置为持久化
- 消息何时刷到磁盘?
  - 写入文件前会有一个Buffer,大小为1M,数据在写入文件时，首先会写入到这个Buffer，若是Buffer已满，则会将Buffer写入到文件（未必刷到磁盘）。
  - 固定的刷盘时间：25ms,也就是无论Buffer满不满，每一个25ms，Buffer里的数据及未刷新到磁盘的文件内容一定会刷到磁盘。
  - 每次消息写入后，若是没有后续写入请求，则会直接将已写入的消息刷到磁盘：使用Erlang的receive x after 0实现，只要进程的信箱里没有消息，则产生一个timeout消息，而timeout会触发刷盘操做。

常见问题

RabbitMQ 管理插件启动报错
- 确认RabbitMQ服务是否启动
- C:\Windows目录下，将.erlang.cookie文件，拷贝到用户目录下 C:\Users{用户名}，这是Erlang的Cookie文件，容许与Erlang进行交互
- 从新安装erl 和 rabbit,尽可能不要带空格的路径
修改配置文件
- http://www.rabbitmq.com/configure.html