初看 Rabbit MQ

vmware 虚拟机centos 7

环境搭建

erlang跟rabbit mq 我都是用的最新的版本

参考这篇帖子：

https://www.cnblogs.com/dreasky/p/9146494.html

安装后执行下： ./rabbitmq-server

demo编写

laravel 框架。

composer 安装

composer require php-amqplib/php-amqplib

让我们来写两个脚本吧

生产者脚本 ProductMqCron

namespace App\Console\Commands;

use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
use Illuminate\Console\Command;

class ProductMqCron extends Command
{
    protected $signature = 'ProductMqCron {page_id?}';
    protected $description = 'ProductMqCron';
    protected $channel = null;
    public function __construct()
    {
        parent::__construct();
        $connection = new AMQPStreamConnection("192.168.95.130", "5672", "guest", "guest");
        $this->channel = $connection->channel();
        $this->channel->queue_declare("hello", false, false, false, false);
    }

    public function handle()
    {
        $this->productMq();
    }
    public function productMq()
    {
        $i = 1;
        while (true) {
            $msg = new AMQPMessage($i);
            $this->channel->basic_publish($msg, '', 'hello');
            $i++;
            if ($i / 1000 == 0) {
                sleep(1);
            }
        }
    }
}

消费者脚本

namespace App\Console\Commands;
 
use PhpAmqpLib\Connection\AMQPStreamConnection;
use Illuminate\Console\Command;

class ConsumeMqCron extends Command
{
    protected $signature = 'ConsumeMqCron';
    protected $description = 'ConsumeMqCron';
    protected $channel = null;
    public function __construct()
    {
        parent::__construct();
        $connection = new AMQPStreamConnection("192.168.95.130", "5672", "guest", "guest");
        $this->channel = $connection->channel();
        $this->channel->queue_declare("hello", false, false, false, false);
    }
    
    public function handle()
    {
        echo "开始消费了\n";
        while (true) {
            $this->consumeMq();
        }
    }
    public function consumeMq()
    {
        $callback = function ($msg) {
            echo "consume page_id: " . $msg->body, "\n";
        };
        $this->channel->basic_consume("hello", '', false, true, false, false, $callback);
        while (count($this->channel->callbacks)) {
            $this->channel->wait();
        }
    }
}