{
"require": {
"php-amqplib/php-amqplib": ">=2.6.1"
}
}
消费者:receive.php
<?php
require_once '../vendor/autoload.php';
$connection = new \PhpAmqpLib\Connection\AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
$channel = $connection->channel();
$channel->queue_declare($queue = 'queue_for', $passive = false, $durable = true, $exclusive = false, $auto_delete = false);
echo " [*] Waiting for messages. To exit press CTRL+C\n";
$callback = function ($msg) {
$msgArr = json_decode($msg->body, true);
echo "正在处理count为" . $msgArr['count'] . "的队列....\n";
for ($i = 0; $i < $msgArr['count']; $i++) {
$myfile = fopen("testfile.txt", "a");
$txt = $i . ' 读取到的时间:' . $msgArr['time'] . "写入文件的时间" . date('Y-m-d H:i:s');
fwrite($myfile, $txt . " \n");
fclose($myfile);
}
echo "count为" . $msgArr['count'] . "的队列处理完成!\n";
$msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
};
$channel->basic_consume('queue_for', '', false, $no_ack = false, false, false, $callback);
while (count($channel->callbacks)) {
$channel->wait();
}
$channel->close();
$connection->close();
生产者: send.php
<?php
require_once '../vendor/autoload.php';
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
class Send
{
public function main()
{
$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
$channel = $connection->channel();
$channel->queue_declare($queue = 'queue_for', $passive = false, $durable = true, $exclusive = false, $auto_delete = false);
$rand = rand(1044400, 2444000);
$sendData = [
'count' => $rand,
'time' => date('Y-m-d H:i:s')
];
$msg = new AMQPMessage(json_encode($sendData));
$channel->basic_publish($msg, '', 'queue_for');
echo " [x] 任务已发出.." . $rand . "\n";
$channel->close();
$connection->close();
}
}
$obj = new Send();
$obj->main();