laravel 使用rabbitmq

composer.json加上："php-amqplib/php-amqplib": "^2.12",然后执行composer updatexxie

消费者：

$mqConfig = GlobalConfig::get('event.mqConfig');$exchange = 'exchange.demo';$queue = 'queue.demo';$consumerTag = 'demo.consumer' . getmypid();if (!isset($mqConfig['address']) || empty($mqConfig['address'])) {    return;}if (!isset($mqConfig['user']) || empty($mqConfig['user'])) {    return;}if (!isset($mqConfig['password']) || empty($mqConfig['password'])) {    return;}$mqAddress = $mqConfig['address'];$mqUser = $mqConfig['user'];$mqPassword = $mqConfig['password'];$mqAddressArr = explode(',', $mqAddress);$addressCount = count($mqAddressArr);if (!$addressCount) {    return;}$connection = null;try {    if ($addressCount > 1) {        $configArr = [];        foreach ($mqAddressArr as $row) {            $address = explode(':', $row);            $host = $address[0];            $port = $address[1];            $configArr[] = ['host' => $host, 'port' => $port, 'user' => $mqUser, 'password' => $mqPassword, 'vhost' => '/'];        }        $connection = AMQPStreamConnection::create_connection($configArr);    } else {        $mqAddress = explode(':', $mqAddress);        $host = $mqAddress[0];        $port = $mqAddress[1];        $connection = new AMQPStreamConnection($host, $port, $mqUser, $mqPassword, '/');    }    if (empty($connection)) {        return;    }    $channel = $connection->channel();    $channel->queue_declare($queue, false, true, false, false);    $channel->exchange_declare($exchange, 'direct', false, true, false);    $channel->queue_bind($queue, $exchange);    $callback = function ($message) {        try {            $record = json_decode($message->body, true);            $this->handleUserRunRecord($record);        } catch (\Exception $e) {            echo 'pulluserrunrecord1 error ,msg:' . $e->getMessage() . "\n";            Log::error('pulluserrunrecord1 error ,msg:' . $e->getMessage());        }        $message->delivery_info['channel']->basic_ack($message->delivery_info['delivery_tag']);        // Send a message with the string "quit" to cancel the consumer.        if ($message->body === 'quit') {            $message->delivery_info['channel']->basic_cancel($message->delivery_info['consumer_tag']);        }    };    $shutdown = function ($channel, $connection) {        $channel->close();        $connection->close();    };    $channel->basic_consume($queue, $consumerTag, false, false, false, false, $callback);    register_shutdown_function($shutdown, $channel, $connection);    // Loop as long as the channel has callbacks registered    while (count($channel->callbacks)) {        $channel->wait();    }} catch (\Exception $e) {   }

生产者：

$exchange = 'exchange.demo'; $queue = 'queue.demo';$consumerTag = 'demo.consumer' . getmypid();$connection = new AMQPStreamConnection('127.0.0.1', 5672, 'guest', 'guest', '/');$channel = $connection->channel();$channel->queue_declare($queue, false, true, false, false);$channel->exchange_declare($exchange, AMQPExchangeType::DIRECT, false, true, false);$channel->queue_bind($queue, $exchange);//$messageBody = implode(' ', array_slice($this->argument, 1));$messageBody = '{"name":"123","id":"1"}';$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 中文文档－PHP版 - RabbitMQ PHP版

RabbitMQ Tutorials — RabbitMQ

来源地址：https://blog.csdn.net/m0_37634311/article/details/127545310