ball/app/common/command/AddAgreementPdf.php

<?php
namespace app\common\command;

use app\common\model\master_worker\MasterWorkerAgree;
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
use PhpAmqpLib\Wire\AMQPTable;
use think\console\Command;
use think\console\Input;
use think\console\Output;
use think\facade\Config;
use think\facade\Log;


/**
 * mq生成PDF协议队列
 * Class AddAgreementPdf
 * @package app\command
 */
class AddAgreementPdf extends Command
{
    protected function configure()
    {
        $this->setName('query_add_agreement')
            ->setDescription('mq生成PDF协议队列');
    }

    protected function execute(Input $input, Output $output)
    {
        $mq_config = Config::get('mq');
        // 连接到RabbitMQ服务
        $connection = new AMQPStreamConnection($mq_config['host'], $mq_config['port'], $mq_config['username'], $mq_config['password'], $mq_config['vhost']);
        $channel = $connection->channel();
        $channel->exchange_declare('add_agreement','topic',true,true,true,false,false);

        // 声明队列
        $queue = 'add_agreement';
        $channel->queue_declare($queue, true, true, false, true,false);

        //接收消息
        $callback = function($msg) {
            $param = json_decode($msg->body, true);
            self::addPdf($param);
        };
        $channel->basic_consume('add_agreement', '', false, true, false, false, $callback);
        while(count($channel->callbacks)) {
            $channel->wait();
        }
        $channel->close();
        $connection->close();
    }

    public static function notifyAgreement()
    {
//        $mq_config = Config::get('mq');
//        // 连接到RabbitMQ服务
//        $connection = new AMQPStreamConnection($mq_config['host'], $mq_config['port'], $mq_config['username'], $mq_config['password'], $mq_config['vhost']);
//        $channel = $connection->channel();
//        $channel->exchange_declare('add_agreement','topic',true,true,true,false,false);
//
//        // 声明队列
//        $queue = 'add_agreement';
//        $channel->queue_declare($queue, true, true, false, true,false);
//
//        //接收消息
//        $callback = function($msg) {
//            $param = json_decode($msg->body, true);
//            self::addPdf($param);
//        };
//        $channel->basic_consume('add_agreement', '', false, true, false, false, $callback);
//        while(count($channel->callbacks)) {
//            $channel->wait();
//        }
//        $channel->close();
//        $connection->close();
    }

    public static function addPdf($param)
    {
        $code = $param['code'];
        $url = $param['url'];
        $pdf = '/'.$code.'.pdf';
        $path = 'uploads/agreement_pdf/'.date('Ymd');
        if(!file_exists($path)){
            mkdir ($path,0777,true);
        }
        $shell_ = 'wkhtmltopdf --page-height 297mm '.$url.' '.$path.$pdf;
        $output = [];
        $return_var = 0;
        \exec($shell_ . ' > /dev/null 2>&1', $output, $return_var);
        if ($return_var === 0) {
            $agreement = MasterWorkerAgree::where('code', $code)->findOrEmpty();
            $agreement->pdf_url = $path.$pdf;
            $agreement->save();
        } else {
            // 处理错误
            echo "Command failed with return code: $return_var";
        }
    }


    public static function sendMq($code,$url):bool
    {
        $mq_config = Config::get('mq');
        // 连接到RabbitMQ服务
        $connection = new AMQPStreamConnection($mq_config['host'], $mq_config['port'], $mq_config['username'], $mq_config['password'], $mq_config['vhost']);
        $channel = $connection->channel();
        $channel->exchange_declare('add_agreement','topic',true,true,true,false,false);

        // 声明队列
        $queue = 'add_agreement';
        $channel->queue_declare($queue, true, true, false, true,false);

        // 发送消息
        $data = [
            'code'=>$code,
            'url'=>$url
        ];
        $messageBody = json_encode($data,JSON_UNESCAPED_UNICODE);
        $message = new AMQPMessage($messageBody);
        $channel->basic_publish($message, '', $queue);
        $channel->close();
        $connection->close();
        return true;
    }
}