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

这篇好文章是转载于：学新通技术网