99国产高清,国产精品性夜天天拍拍2021,日本人啪啪

一般的隊列系統，是指linux中的crontab定時啟動腳本來處理任務:

首先下載一個rabbitmq的客戶端，他相當于一個容器，裝排隊數據的容器

http://www.rabbitmq.com/download.html

默認的端口是55672 訪問地址http://127.0.0.1:55672/

默認帳號密碼 guest guest

你可以看到rabbitmq 的管理界面

mq的任務是一個不浪費資源，的一個隊列系統！

php使用需要下載一個amqp擴展

或者直接點擊下面的地址找到適合自己的版本，下載

http://pecl.php.net/package/amqp/1.2.0/windows

rabbitmq.1.dll 放在C盤windows下

php_amqp.dll 放入php擴展中

開啟php_amqp.dll的引用

重啟服務器

用phpinfo();

查看是否引用成功，如果出現以下的amqp擴展，那就說明成功了

首先是rabbitmq的生產者：

創建第一個index文件：然后去mq中查看，如果添加一個test001的隊列名信息，就說明已經添加進去了，xx22的信息已經在mq中存儲！

接下來就需要跑數據了。

createQueue(array('xxx','2222'),'test001');

echo "ok";

function createQueue($message,$queueName,$exchangeName = '', $queueKey = '')

{

$queueName = self::getQueueName($queueName);

$conn_args = array('host' =>'localhost', 'port'=> '5672',

'login' =>'guest', //mq帳號

'password'=> '', //mq密碼

'vhost' => '/');

$conn = new AMQPConnection($conn_args);

$conn->connect();

$channel = new AMQPChannel($conn);

if (!$exchangeName) {

$exchangeName = $queueName;

}

$queueName = $queueName;

if (!$queueKey) {

$queueKey = $queueName;

}

$ex = new AMQPExchange($channel);

$ex->setName($exchangeName);

$ex->setType(AMQP_EX_TYPE_TOPIC);

$ex->setFlags(AMQP_DURABLE); //exchange持久化

$ex->declareExchange();

$q = new AMQPQueue($channel);

$q->setName($queueName);

$q->setFlags(AMQP_DURABLE); //queue持久化

$q->declareQueue();

$q->bind($exchangeName, $queueKey);

$channel->startTransaction();

/**

* 消息持久化,delivery_mode:2持久化、delivery_mode:1非持久化，其中priority是設置消息的優先級，測試中發現并未起作用。

* 消息還有其他屬性，請參考http://www.php.net/manual/zh/amqpexchange.publish.php

$result = $ex->publish(json_encode($message), $queueKey, AMQP_NOPARAM, array('delivery_mode'=>2, 'priority'=> 9));

$channel->commitTransaction();

$conn->disconnect();

}

有了生產者，那就有消費者。

腳本如果沒有其他的修改或問題，基本上都是常年啟動的：

消費者基類：

class WorkerCommand{

function qInit($q_name,$e_name='',$k_route=''){

$q_name = Utils::getQueueName($q_name);

$conn_args = array(

'host' => '127.0.0.1', //mq的配置

'port' => '5672',

'login' => 'guest',

'password' => 'huoxingxing',

'vhost' => '/'

);

//創建連接和channel

$conn = new AMQPConnection($conn_args);

if (!$conn->connect()) {

die("Cannot connect to the broker!\n");

}

$channel = new AMQPChannel($conn);

//創建交換機

$ex = new AMQPExchange($channel);

if (!$e_name) {

$e_name = $q_name;

}

$ex->setName($e_name);

$ex->setType(AMQP_EX_TYPE_DIRECT); //direct類型

$ex->setFlags(AMQP_DURABLE); //持久化

// echo "Exchange Status:" . $ex->declareExchange() . "\n";

//創建隊列

$q = new AMQPQueue($channel);

$q->setName($q_name);

$q->setFlags(AMQP_DURABLE); //持久化

// echo "Message Total:" . $q->declareExchange() . "\n";

if (!$k_route) {

$k_route = $q_name;

}

//綁定交換機與隊列，并指定路由鍵

// echo 'Queue Bind: ' . $q->declareQueue($e_name, $k_route) . "\n";

//阻塞模式接收消息

echo "Message:\n";

while (True) {

$q->consume(array($this,'processMessage'));

//$q->consume('processMessage', AMQP_AUTOACK); //自動ACK應答

}

$conn->disconnect();

}

消費者：

class WorkerWareSyncBackUpCommand extends WorkerCommand {

function actionIndex()

{

$this->qInit('SyncWareBackup');

}

function processMessage($envelope, $queue)

{

$msg = json_decode($envelope->getBody());

Utils::doBackUp('back',$msg,'');

$queue->ack($envelope->getDeliveryTag()); //手動發送ACK應答

}

RibbetMQ php擴展使用實現隊列生產消費

相關軟件

相關文章

擴展知識

相關評論

閱讀本文后您有什么感想? 已有 0 人給出評價!

文章推薦

最新文章

本類常用軟件

RibbetMQ php擴展使用 實現隊列生產消費