- 安装rabbitmq
- 使用composer安装laravel的php-amqplib/php-amqblib包
- 推送消息到队列和读取队列中的消息
安装rabbitmq
安装rabbit首先需要安装erlang环境，然后下载rabbitmq客户端进行安装，由于我实在windows环境下进行搭建的，所以过程很简单，这里就不直接介绍安装方法了，可以自行百度。安装完成之后记得使用命令行安装rabbitmq的浏览器插件，然后就可以在浏览器中看到rabbitmq的信息了。
由于我研究的时间不是很长，所以这里面很多我也不是很清楚。这个Queues下面的列表代表着这个服务器上存在几个队列。
使用composer安装扩展
扩展地址
可以直接在composer.json文件中添加，然后执行composer update，也可以在命令行中执行composer require
推送消息到rabbitmq中
Github上有详细的实例代码，我这里只是将其简化。
config.php
<?php
//创建配置文件
//使用自动加载类
require_once __DIR__ . '/vendor/autoload.php';
define('HOST', 'localhost');
define('PORT', 5672); // 注意这里端口号，在浏览器中的端口号是15672，但在这里确实5672
define('USER', 'guest');
define('PASS', 'guest');
define('VHOST', '/');
//If this is enabled you can see AMQP output on the CLI
//如果为true会显示一堆这种类型的调试信息。
define('AMQP_DEBUG', false);
amqp_publisher.php
<?php
// 使用配置文件 include(__DIR__ . '/config.php');
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
$exchange = 'router'; // 交换器，在我理解，如果两个队列使用一个交换器就代表着两个队列是同步的，这个队列里存在的消息，在另一个队列里也会存在 $queue = 'test'; // 队列名称
$connection = new AMQPStreamConnection(HOST, PORT, USER, PASS, VHOST); // 创建连接
$channel = $connection->channel(); $channel->queue_declare($queue, false, true, false, false);
$channel->exchange_declare($exchange, 'direct', false, true, false);
$channel->queue_bind($queue, $exchange); // 队列和交换器绑定
$messageBody = 'testinfo12'; // 要推送的消息
$message = new AMQPMessage($messageBody, array('content_type' => 'text/plain', 'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT));
$channel->basic_publish($message, $exchange); // 推送消息
$channel->close(); $connection->close();
执行完这段代码之后会发现rabbitmq的test队列会多出一个消息
从rabbitmq中取出消息
在这里我使用的是github中demo中的basic_get.php中的方法，还有一种amqp_consumer.php方法，大家可以试试看。
basic_get.php
<?php
include(__DIR__ . '/config.php');
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage; $exchange = 'router'; $queue = 'test';
$connection = new AMQPStreamConnection(HOST, PORT, USER, PASS, VHOST);
$channel = $connection->channel();
$message = $channel->basic_get($queue); //取出消息
print_r($message->body);
$channel->basic_ack($message->delivery_info['delivery_tag']); // 确认取出消息后会发送一个ack来确认取出来了，然后会从rabbitmq中将这个消息移除，如果删掉这段代码，会发现rabbitmq中的消息还是没有减少
$channel->close();
$connection->close();
在laravel中使用
因为我们需要实时监听这个消息队列中是否有消息，如果有就取出消息队列进行处理。
这里laravel提供了一个commad。
在项目目录下执行php artisan make:console GetMessage 这里的名字可以自己起
然后找到app/Console/Commands，会发现多出了一个GetRQMessage.php文件。
<?php
namespace Ckk\Console\Commands;
use Illuminate\Console\Command;
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
class UserCenter extends Command {
/**
* The name and signature of the console command.
*
* @var string
*/
protected $signature = 'getRQMessage';
// 这里是生成命令的名称
/**
* The console command description.
*
* @var string
*/
protected $description = '这里是这个命令的描述';
/**
* Create a new command instance.
*
* @return void
*/
public function __construct() {
parent::__construct();
}
/*
*
* Execute the console command.
*
* @return mixed
*/
public function handle() {
// 这里写具体代码,这里可以创建一个配置文件，然后使用配置文件控制消息队列的配置信息
$queue = 'test';
$connection = new AMQPStreamConnection('localhost', '5672', 'guest', 'guest');
$channel = $connection->channel();
$channel->queue_declare($queue, false, true, false, false);
$message = $channel->basic_get($queue);
$channel->basic_ack($message->delivery_info['delivery_tag']);
if ($message->body) {
print_r($message->body);
\Log::info('queueInfo:' . $message->body);
}
$channel->close();
$connection->close();
}
在Console目录下的Kernel.php中添加
protected $commands = [
GetRQMessage::class,
// Commands\Inspire::class,
];
然后在命令行中执行php artisan getMessage 会发现rabbitmq中队列里的消息减少了一个，然后打开日志文件，会发现已经写进去了。那么怎么实时进行检测呢，只要使用linux自带的crontab不停的执行php artisan getRQMessage 就好了。
