RabbitMQ 实战记录

啊 好久没有写博客了, 我的博客就是总结自己的学习之路
加油 重新启程 keep calm and think more.

准备

composer require php-amqplib/php-amqplib

生产者

<?php

namespace app\api\controller;

use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
use PhpAmqpLib\Wire\AMQPTable;
use think\Request;

class Test
{
    /**
     * 消息队列
     * @throws \Exception
     */
    public function deliver()
    {
        $params = Request::instance()->param();
        unset($params['sign']);
        $params['send_time']   = time();
        $params['create_time'] = time();

        $this->deliverMessage(json_encode($params), 'test_exchange', 'api_test_queue', 'api_test');

        echo 'Done';exit();
    }

    /**
     * 消息队列 - 投递消息
     * @param $message
     * @param $exchangeName
     * @param $queueName
     * @param $routingKey
     * @throws \Exception
     */
    private function deliverMessage($message, $exchangeName, $queueName, $routingKey)
    {
        // 创建连接和信道
        $connection = new AMQPStreamConnection(
            RABBITMQ_CONFIG['host'],
            RABBITMQ_CONFIG['port'],
            RABBITMQ_CONFIG['TEST']['user'],
            RABBITMQ_CONFIG['TEST']['password'],
            RABBITMQ_CONFIG['TEST']['vhost']
        );
        if (!$connection->isConnected()) {
            var_dump('Connection failed');
        }

        $channel = $connection->channel();
        // 声明交换机
        // type参数: (direct:精准推送 fanout:广播 topic:组播)
        $channel->exchange_declare($exchangeName, 'direct', false, false, false);
        $table = new AMQPTable();
        $table->set('x-queue-mode', 'lazy');
        // 声明队列
        $channel->queue_declare($queueName, false, true, false, false, false, $table);
        // 将队列绑定到交换机
        $channel->queue_bind($queueName, $exchangeName, $routingKey);
        $msg = new AMQPMessage($message, [
            'delivery_mode' => AMQPMessage::DELIVERY_MODE_NON_PERSISTENT,
        ]);
        // 向队列发送消息
        $channel->basic_publish($msg, $exchangeName, $routingKey);
        // 关闭信道, 关闭连接
        $channel->close();
        $connection->close();
    }
}

http://tp5.test/api/test/deliver

消费者

<?php

namespace app\api\command;

require_once APP_PATH . 'api.php';

use think\console\{
    Command, Input, Output
};
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Wire\AMQPTable;

class Test extends Command
{
    protected function configure()
    {
        $this->setName('cmd_test')
             ->setDescription('RabbitMQ Consumer Test');
    }

    /**
     * 执行命令
     * @param Input $input
     * @param Output $output
     * @return int|void|null
     * @throws \ErrorException
     * @throws \Exception
     */
    protected function execute(Input $input, Output $output)
    {
        // 交换机
        $exchangeName = "test_exchange";
        // 队列
        $queueName    = "api_test_queue";
        // 路由键
        $routingKey   = "api_test";
        // 创建连接和信道
        $connection   = new AMQPStreamConnection(
            RABBITMQ_CONFIG['host'],
            RABBITMQ_CONFIG['port'],
            RABBITMQ_CONFIG['TEST']['user'],
            RABBITMQ_CONFIG['TEST']['password'],
            RABBITMQ_CONFIG['TEST']['vhost']
        );
        $channel = $connection->channel();
        // 声明交换机
        $channel->exchange_declare($exchangeName, 'direct', false, false, false);
        // 设置惰性队列
        $table = new AMQPTable();
        $table->set('x-queue-mode', "lazy");
        // 声明队列
        $channel->queue_declare($queueName, false, true, false, false, false, $table);
        // 将队列绑定到交换机
        $channel->queue_bind($queueName, $exchangeName, $routingKey);

        $callback = function ($msg) {
            // TODO 处理数据
            try {
                $msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
            } catch (\Exception $e) {
                var_dump($e->getMessage());
            }
        };

        // 设置consumer同时处理队列数量
        $channel->basic_qos(null, 1, null);
        // 消费队列
        $channel->basic_consume($queueName, '', false, false, false, false, $callback);
        while (count($channel->callbacks)) {
            $channel->wait();
        }
        // 关闭信道,关闭连接
        $channel->close();
        $connection->close();
    }
}

php think cmd_test

• api.php

// Rabbit配置信息
define('RABBITMQ_CONFIG', [
    'host' => '127.0.0.1',
    'port' => 5672,
    'TEST' => [  // 玩法虚拟机
        'user'     => 'test' ,
        'password' => 'test',
        'vhost'    => 'test'
    ],
]);