TP5使用RabbitMQ实现消息队列的项目实践

composer require php-amqplib/php-amqplib

<?php
namespace app\api\command;
use think\console\Command;
use think\console\Input;
use think\console\Output;
/**
 * 自定义命令测试
 */
class Test extends Command
{
    /**
     * 配置
     */
    protected function configure()
    {
        // 设置命令的名称和描述
        $this->setName('test')->setDescription('这是一个测试命令');
    }
    /**
     * 执行
     */
    protected function execute(Input $input, Output $output)
    {
        $output->writeln("测试命令");
    }
}

<?php
return [
    'app\api\command\Test',
];

php think test

test command

<?php
namespace app\api\command;
use PhpAmqpLib\Connection\AMQPStreamConnection;
use think\console\Command;
use think\console\Input;
use think\console\Output;
/**
 * RabbitMq 启动命令
 */
class Ramq extends Command
{
    protected $consumerTag = 'customer';
    protected $exchange = 'xcuser';
    protected $queue = 'xcmsg';
    protected function configure()
    {
        $this->setName('ramq')->setDescription('rabbitmq');
    }
    protected function execute(Input $input, Output $output)
    {
        $output->writeln("消息队列开始");
        $this->start();
        // 指令输出
        $output->writeln('消费队列结束');
    }
    /**
     * 关闭
     */
    function shutdown($channel, $connection)
    {
        $channel->close();
        $connection->close();
    }
    /**
     * 回调处理信息
     */
    function process_message($message)
    {
        if ($message->body !== 'quit') {
            echo $message->body;
        }
        //手动应答
        $message->delivery_info['channel']->basic_ack($message->delivery_info['delivery_tag']);
        if ($message->body === 'quit') {
            $message->delivery_info['channel']->basic_cancel($message->delivery_info['consumer_tag']);
        }
    }
    /**
     * 启动 守护进程运行
     */
    public function start()
    {
        $host = '127.0.0.1';
        $port = 5672;
        $user = 'guest';
        $pwd = 'guest';
        $vhost = '/';
        $connection = new AMQPStreamConnection($host, $port, $user, $pwd, $vhost);
        $channel = $connection->channel();
        $channel->queue_declare($this->queue, false, true, false, false);
        $channel->exchange_declare($this->exchange, 'direct', false, true, false);
        $channel->queue_bind($this->queue, $this->exchange);
        $channel->basic_consume($this->queue, $this->consumerTag, false, false, false, false, array($this, 'process_message'));
        register_shutdown_function(array($this, 'shutdown'), $channel, $connection);
        while (count($channel->callbacks)) {
            $channel->wait();
        }
    }
}

return [
    'app\api\command\Ramq',// rabbitmq
];

<?php
namespace app\api\controller;
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
use think\Controller;
/**
 * 发送端
 */
class MessageQueue extends Controller
{
    const exchange = 'xcuser';
    const queue = 'xcmsg';
    /**
     * 发送消息
     */
    public function pushMessage($data)
    {
        $host = '127.0.0.1';
        $port = 5672;
        $user = 'guest';
        $pwd = 'guest';
        $vhost = '/';
        $connection = new AMQPStreamConnection($host, $port, $user, $pwd, $vhost);
        $channel = $connection->channel();
        $channel->exchange_declare(self::exchange, 'direct', false, true, false);
        $channel->queue_declare(self::queue, false, true, false, false);
        $channel->queue_bind(self::queue, self::exchange);
        $messageBody = $data;
        $message = new AMQPMessage($messageBody, array('content_type' => 'text/plain', 'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT));
        $channel->basic_publish($message, self::exchange);
        $channel->close();
        $connection->close();
        echo 'ok';
    }
    /**
     * 执行
     */
    public function index()
    {
        $data = json_encode(['msg' => '测试数据', 'id' => '15']);
        $this->pushMessage($data);
    }
}

php think ramq