rabbitmq学习和部署记

2018-07-05 本文已影响0人恐龙打酱油

网上rabbitmq的学习日志非常丰富，官网文档也很完美，这里主要记录学习和部署过程中的一些记录。会按以下菜单进行记录。后面会贴上引用文章的链接。

1、rabbitmq的介绍

2、rabbitmq和redis的区别

3、rabbitmq的安装

4、rabbitmq的部署配置

5、简单队列模式的实现

6、工作队列模式的实现

7、发布订阅模式的实现

8、路由模式的实现

9、通配符模式的实现

10、RPC模式的实现

1、rabbitmq的介绍

RabbitMQ

RabbitMQ是实现AMQP（高级消息队列协议）的消息中间件的一种，最初起源于金融系统，用于在分布式系统中存储转发消息，在易用性、扩展性、高可用性等方面表现不俗。消息中间件主要用于组件之间的解耦，消息的发送者无需知道消息使用者的存在，反之亦然。

Redis

是一个Key-Value的NoSQL数据库，开发维护很活跃，虽然它是一个Key-Value数据库存储系统，但它本身支持MQ功能，所以完全可以当做一个轻量级的队列服务来使用。

2、rabbitmq和redis的区别

可靠消费

Redis：没有相应的机制保证消息的消费，当消费者消费失败的时候，消息体丢失，需要手动处理

RabbitMQ：具有消息消费确认，即使消费者消费失败，也会自动使消息体返回原队列，同时可全程持久化，保证消息体被正确消费

可靠发布

Reids：不提供，需自行实现

RabbitMQ：具有发布确认功能，保证消息被发布到服务器

高可用

Redis：采用主从模式，读写分离，但是故障转移还没有非常完善的官方解决方案

RabbitMQ：集群采用磁盘、内存节点，任意单点故障都不会影响整个队列的操作

持久化

Redis：将整个Redis实例持久化到磁盘

RabbitMQ：队列，消息，都可以选择是否持久化

消费者负载均衡

Redis：不提供，需自行实现

RabbitMQ：根据消费者情况，进行消息的均衡分发

队列监控

Redis：不提供，需自行实现

RabbitMQ：后台可以监控某个队列的所有信息，（内存，磁盘，消费者，生产者，速率等）

流量控制

Redis：不提供，需自行实现

RabbitMQ：服务器过载的情况，对生产者速率会进行限制，保证服务可靠性

应用场景分析

Redis：

轻量级，高并发，延迟敏感

即时数据分析、秒杀计数器、缓存等

RabbitMQ：

重量级，高并发，异步

批量数据异步处理、并行任务串行化，高负载任务的负载均衡等

3、rabbitmq的安装

rabbitmq需要erlang支持，所以需要先下载安装erlang。我基本上都是先用迅雷下载后再拖到虚机中进行安装。

# wget http://erlang.org/download/otp_src_20.3.tar.gz

# wget https://dl.bintray.com/rabbitmq/all/rabbitmq-server/3.7.6/rabbitmq-server-3.7.6-1.el7.noarch.rpm

安装erlang需要先通过yum安装必要的依赖。

# yum install build-essential openssl openssl-devel unixODBC unixODBC-devel make gcc gcc-c++ kernel-devel m4 ncurses-devel openssl-devel java-1.7.0-openjdk-devel.x86_64 fop fakefop -y

然后安装erlang

# ./configure --prefix=/usr/erlang

# make

# make install

很多人会在profile中加入PATH=$PATH:/usr/erlang/bin，然而我并没有加，遇到问题后再说。

安装成功后运行

# erl

输出对应的版本号表示安装成功。

版本号

接下来安装rabbitmq，安装rabbitmq这里我遇到比较多的依赖问题，尤其是下面这个：

依赖报错

首先我已经在官网上确认过3.7.6的rabbitmq依赖的erlang是19.3-20.3的版本。所以依赖是没有问题，但是rabbitmq却识别不到我安装的erlang的版本。

官网介绍的版本对应表如下：http://www.rabbitmq.com/which-erlang.html

没办法我只能在安装rabbitmq的时候使用 --nodeps，忽略这个依赖问题。

# rpm -ivh --nodeps rabbitmq-server-3.7.6-1.el7.noarch.rpm

结果很快就安装成功了，我认为是环境变量没有添加导致的，但是我现在并不想解决这个问题。如果后面使用生产和消费的时候再出现类似的问题再解决。

4、rabbitmq的部署配置

首先开启后台

# service rabbitmq-server start

# rabbitmq-plugins enable rabbitmq_management

然后添加所需的用户以及分配他们的权限，以下的命令可以在后台中做操作。

# rabbitmqctl add_user rabbitadmin 123456

# rabbitmqctl add_user rabbitmonitor 123456

# rabbitmqctl add_user rabbitpolicymaker 123456

# rabbitmqctl add_user rabbitmanager 123456

# rabbitmqctl add_user rabbitproduct 123456

# rabbitmqctl add_user rabbitcustomer 123456

# rabbitmqctl set_user_tags rabbitadmin administrator

# rabbitmqctl set_user_tags rabbitmonitor monitoring

# rabbitmqctl set_user_tags rabbitpolicymaker policymaker

# rabbitmqctl set_user_tags rabbitmanager management

# rabbitmqctl list_users

用户列表

用户创建完成，可以登录后台http://127.0.0.1:15672，可视式地看到用户列表。

用户列表

5、简单队列模式的实现

接下来使用PHP来做测试，首先按官网建议的方式安装php-amqplib包：

# composer require php-amqplib/php-amqplib

接下来配置相关的参数：

define('HOST', '192.168.1.182');

define('PORT', 5672);

define('USER', 'rabbitcustomer');

define('PASS', '123456');

define('VHOST', '/');

define('AMQP_DEBUG', true);

然后调用

$connection = newAMQPStreamConnection(HOST, PORT, USER, PASS, VHOST);

$channel = $connection->channel();

//第1个参数name 队列名称

//第2个参数passive 是否被动，默认为false

//第3个参数durable 表示队列将在服务器重启后继续存在，默认为true

//第4个参数exclusive是否独占，默认为false 表示可以在其他通道中访问队列

//第5个参数auto_delete自动删除，默认为false 表示关闭频道后不会删除队列

var_dump($channel->queue_declare('rabbit', false, true, false, false)); ##这里表示创建一个rabbit队列

提示报错：

Fatal error: Uncaught PhpAmqpLibExceptionAMQPProtocolConnectionException: NOT_ALLOWED - access to vhost '/' refused for user 'rabbitcustomer'

rabbitcustomer这个用户没有访问 '/'的权限。设置rabbitcustomer用户访问名为“/ ”的虚拟主机，并对名称以“rabbit”开头的所有资源具有配置权限，并对所有资源执行写入和读取权限资源。

# rabbitmqctl set_permissions -p / rabbitcustomer "^rabbit.*" ".*" ".*"

设置好后再调用PHP程序，还有报错：

Fatal error: Uncaught PhpAmqpLibExceptionAMQPProtocolChannelException: ACCESS_REFUSED - access to queue 'rabbitcustomer' in vhost '/' refused for user 'rabbitcustomer'

说明rabbitcustomer用户并没有权限创建队列。

再次检查权限问题时发现并没有给rabbitadmin 赋值任何权限。

rabbitmqadmin命令Access refused

rabbitmqadmin用户没有创建队列的权限，需要先赋权限。

# rabbitmqctl set_permissions -p / rabbitadmin ".*" ".*" ".*"

再次调用成功返回。表示rabbitcustomer能够创建自己的队列了。

队列列表

接着rabbitcustomer发送一个消息到队列中。

$channel->basic_publish(new AMQPMessage('Hello World for RabbitMQ!'), '', 'hello');

die(" [x] Sent 'Hello World!'");

执行后表示已经将消息Hello World for RabbitMQ!发送到了队列。

接着调用获取basic_consume函数，获取队列消息。

$callback = function ($msg) {

echo ' [x] Received ', $msg->body, "\n";

};

//第1个参数name 队列名称

//第2个参数consumer_tag 表示消费者标识符

//第3个参数no_local 表示不接收此消费者发布的消息

//第4个参数no_ack 表示告诉服务器消费者是否会确认消息

//第5个参数exclusive 是否请求独占的使用者访问权限，这意味着只有此使用者才能访问队列

//第6个参数nowait

//第7个参数callback，指定回调函数

$channel->basic_consume($queue, '', false, true, false, false, $callback);

while (count($channel->callbacks)) {

$channel->wait();

}

先调用发送消息，再调用获取消息，同样可以获取到消息。获取后以后再调用上面的程序就无法再获取信息了。这就完成了一个最简单的生产消费模式的例子。

[x] Received Hello World for RabbitMQ!

6、工作队列模式的实现

工作队列模式其实就是在简单模式的基础上添加多个worker来获取列表数据，达到负载均衡的目的。同时可以控制队列发送和接收的确认。当消息弹出在处理过程中，worker断线导致消息读取失败，则会由消息队列任务再次发送。可以控制消息在成功接收后是否删除消息等等。

在创建worker的程序中添加1个以上的程序。当发送消息的时候，消息会平均分摊到worker去读取和执行。

queue_declare和basic_consume函数中的参数都有其特定的意思，我在这里用表格具体表述。

两个函数的参数说明

首先是queue_declare函数中的第二个参数passive，当true时会先判断队列是否存在，如果存在则返回队列名称，如果不存在则报错。当false时则正常去创建队列。这个参数可以用于worker中判断队列是否存在。

第三个参数durable表示队列是否持久化，如果是长期存储在服务器中，即使重启服务也会存在。经过测试发现，重启后队列虽然还存在，但是消息丢失了，这是因为建立消息时也需要进行配置，下面是消息非持久化和持久化的代码。官方说明中，这个持久化并不能确认一定成功，如果要求比较高，则需要使用到发布订阅模式中的发布确认。

$msg = new AMQPMessage($data); //非持久化

$msg =newAMQPMessage( $data,array('delivery_mode'=> AMQPMessage::DELIVERY_MODE_PERSISTENT)); //持久化

第四个参数是exclusive表示是否独占，置了独占为true的队列只可以在本次的连接中被访问，也就是说在当前连接创建多少个channel访问都没有关系，但是当连接断开或者有新的连接进来时，该队列就无法访问了。

第五个参数是自动删除，queue会自动删除自己，当消息已经被全部读取后，并且没有新的worker连接去读取消息时，队列将删除自己。

然后是basic_consume函数中的第二个参数consumer_tag，用于标识当前的消费者。

第三个参数no_local，表示不接收此消费者发布的消息，什么意思，consumer_tag标识的消费者会发布消息吗？在发布消息的函数中可以由谁来发布消息吗？

第四个参数no_ack，表示告诉服务器消费者是否会确认消息，当确认该消息已接收，则消息会被清除。这个时候有可能会丢失操作，当消息接收后如果没有执行完成导致操作失败，消息就有可能丢失。这就需要手动来确认，代码片段如下：

$msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);

在执行这段代码的时候出现报错：

Fatal error: Uncaught PhpAmqpLib\Exception\AMQPProtocolChannelException: PRECONDITION_FAILED - unknown delivery tag 1

表示no_ack自动确认和上面这段代码不能同时存在。设置no_ack参数为false后，再次调试可以看到未确认执行完成的消息会在另一个worker

未执行完成的第11条消息会被另一个worker执行

如果没有可用的worker，那么消息会继续存储在队列中等待读取，可以通过下面这个命令输出未读取的消息。

#rabbitmqctl list_queues -n rabbit messages_ready messages_unacknowledged

在多工作队列读取消息的时候，会因为队列未获取消息确认前始终有worker闲置的情况，所以需要在worker.php代码中basic_consume调用前加入下面这个代码。

$channel->basic_qos(null, 1, null);

$channel->basic_consume($queue, $consumerTag, false, false, false, false, $callback);

但是要避免所有worker被占满的问题。

7、发布订阅模式的实现

RabbitMQ中消息传递模型的核心思想是生产者永远不会将任何消息直接发送到队列。实际上，生产者通常甚至不知道消息是否会被传递到队列。

在发布订阅模式中需要使用exchange_declare函数。参数

//第1个参数exchange 名称

//第2个参数类型，枚举值有 direct、topic、headers 、fanout

direct 直接入队，不需要与队列绑定，而且需要通过RouteKey做队列标识

topic 通配符入队，不需要与队列绑定，而且需要通过通配符的方式做为队列标识

fanout 输出端，需要与队列做绑定，只输出至已绑定的队列中

//第3个参数passive 是否被动，默认为false

//第4个参数durable表示是否持久化

//第5个参数auto_delete自动删除，默认为false 表示关闭频道后不会删除队列

$channel->exchange_declare('rabbit-ex', 'direct ', false, true, false);

调用创建成功，通过下面的命令查看已存在的队列，或者在rabbitmq后台查看exchanges列表。列表中有许多默认存在的exchange，可以理解为所有的消息其实都是通过exchanges来转换到队列中的，如果没有指定exchanges，消息则会通过默认的exchanges来转换。

# rabbitmqctl list_exchanges

交换器列表

创建queue以及exchange后，开始绑定。

$channel->queue_bind($queue, $exchange);

$channel->basic_publish($msg, $exchange, $queue); //指定交换器发送到指定队列

获取消息时可以正常获取消息。

exchange可以在不指定队列时创建临时队列，并且在获取消息的自动选择exchange创建好的队列来获取消息。这种模式主要用于对消息存储不敏感，只需要获取最新消息。

在发布消息时不创建队列也不绑定队列和交换器，代码片段如下：

$channel->exchange_declare($exchange, 'fanout', false, false, false);

$channel->basic_publish($msg, $exchange); //消息发布不指定队列，由交换器自行选择队列，不需要显式绑定队列

在获取代码时的代码片段如下：

list($queue, ,) = $channel->queue_declare("", false, false, true, false); //以独占的方式创建一个随机队列,这个连接关闭后即自动删除队列，并返回临时队列名

$channel->queue_bind($queue, $exchange); //绑定队列和交换器

$channel->basic_consume($queue, $consumerTag, false, false, false, false, $callback);

在获取消息时创建两个worker来获取消息，可以查看到有两个bingings

# rabbitmqctl list_bindings

绑定的两个bindings

临时的队列则不会产生负载均衡，会在多个worker同时获取到相同的消息。

两个worker有相同的消息

8、路由模式的实现

路由模式主要使用direct类型的exchange来实现。通过不同的routing_key把不同级别的消息存放到不同的队列中。

发布者的代码片段如下：

$severity = 'error'; //可以是info 、noting、warning、error

$channel->exchange_declare($exchange, 'direct', false, false, false);

$channel->basic_publish($msg, $exchange, $severity);

将消息发送到对应的交换器，并绑定对应的routing_key。

发送对应的routing_key消息

消费者的代码片段如下：

list($queue, ,) = $channel->queue_declare("", false, false, true, true); //以独占的方式创建一个随机队列,这个连接关闭后即自动删除队列

$channel->queue_bind($queue, $exchange, $severity); //绑定对应的routing_key

$channel->basic_consume($queue, ‘’, false, true, false, false, $callback);

分别输出noting和error的日志消息

9、通配符模式的实现

路由模式主要使用topic类型的exchange来实现。通过不同的binding_key实现存放多个不同级别的消息存放到队列中。

发布者代码片段如下：

$channel->exchange_declare($exchange, 'topic', false, false, false);

$channel->basic_publish($msg, $exchange, $binding_key); //消息发布不指定队列，由交换器自行选择默认队列，绑定对应的binding_key

发布消息

消费者代码片段如下：

$channel->queue_bind($queue, $exchange, $binding_key); //绑定对应的binding_key

$channel->basic_consume($queue, $consumerTag, false, true, false, false, $callback);

分种不同的匹配方式获取日志消息

10、RPC模式的实现

通过Rabbitmq来实现RPC调用是非常简单的。Server端等待接收消息，当接收到消息完成处理后，又作为一个消息发布者将执行结果回返给Client。Client发送消息并等待Server端的返回结果，从而达到远程调用的目的，官方提供了非常直观的图例。

基于rabbitmq的RPC

执行的顺序如下：

1、客户端启动时，会创建一个匿名的独占回调队列。

2、对于RPC请求，客户端发送带有两个属性的消息： reply_to（设置为回调队列）和correlation_id（设置为每个请求的唯一值）。

3、请求被发送到rpc_queue队列。

4、RPC worker（aka：server）正在等待该队列上的请求。当出现请求时，它会执行该作业，并使用reply_to字段中的队列将带有结果的消息发送回客户端。

5、客户端等待回调队列上的数据。出现消息时，它会检查correlation_id属性。如果它与请求中的值匹配，则返回对应用程序的响应。

要完成这样的调用需要使用到两个重要的参数，一个是reply_to，一个是correlation_id。发送消息的代码片段如下：

$msg = new AMQPMessage(

(string) $content,

array(

'correlation_id' => $this->corr_id,

'reply_to' => $this->callback_queue

)

);

reply_to 表示回调的队列，当消息处理完成后，将回调结果返回给相应的队列中，每一个客户端可以通过回调队列来达到区分客户端的目的。

correlation_id 主要用来标识消息的唯一值，并根据该属性，能够将响应与请求进行唯一匹配。

服务端的代码片段如下：

//创建一个队列，准备接收客户端的消息

$channel->queue_declare($queue, false, false, false, false);

echo " [x] Awaiting RPC requests\n";

//服务端的回调，通过回调队列将消息返回给对应的客户端

$callback = function ($req) {

$n = strlen($req->body);

echo " [Messages] $n\n";

$msg = new AMQPMessage(

'[message had execute] string length'.$n.'. return true.',

array('correlation_id' => $req->get('correlation_id'))

);

sleep(1);

$req->delivery_info['channel']->basic_publish(

$msg,

'',

$req->get('reply_to')

);

$req->delivery_info['channel']->basic_ack(

$req->delivery_info['delivery_tag']

);

};

//负载均衡，可以创建多个服务端来接收消息

$channel->basic_qos(null, 1, null);

$channel->basic_consume($queue, '', false, false, false, false, $callback);

客户端的代码片段如下：

class FibonacciRpcClient

{

private $connection;

private $channel;

private $callback_queue;

private $response;

private $corr_id;

private $queue;

public function __construct()

{

$this->queue = 'rpc_queue';

$this->connection = new AMQPStreamConnection(HOST, PORT, USER, PASS, VHOST);

$this->channel = $this->connection->channel();

list($this->callback_queue, ,) = $this->channel->queue_declare("", false, false, true, false);

$this->channel->basic_consume($this->callback_queue, '', false, false, false, false, array($this,'onResponse'));

}

public function onResponse($rep)

{

if ($rep->get('correlation_id') == $this->corr_id) {$this->response = $rep->body;}

}

public function call($n)

{

$this->response = null;

$this->corr_id = uniqid();

$msg = new AMQPMessage((string) $n, array('correlation_id' => $this->corr_id, 'reply_to' => $this->callback_queue));

$this->channel->basic_publish($msg, '', $this->queue);

while (!$this->response) {

$this->channel->wait();

}

return $this->response;

}

}

$param = isset($argv[1]) && !empty($argv[1]) ? $argv[1] : ''; //获取当前日志消息的级别

$fibonacci_rpc = new FibonacciRpcClient();

$response = $fibonacci_rpc->call($param);

echo ' [.] Got ', $response, "\n";