AMQP

时间：2014-06-20 13:42:54 阅读：227 评论：0 收藏：0 [点我收藏+]

标签：php amqp amqpconnection amqpchannel amqpexchange amqpqueue

<?php
/*
* 消息队列
* @access Queue.php
* @author liuhaipeng <liuhaipeng@dhgate.com>
* @create 2013-10-17 上午10:42:11
* @version 1.0
* @copyright Copyright © 2013, DHgate.com
*/

class Scrm_Queue {
protected static $connect = NULL;
protected static $channel = NULL;
protected static $exchangeName = NULL;
protected static $queue = NULL;
protected static $queueName = NULL;

public function __construct($config, $queueName = NULL, $exchangeName = NULL) {
self::$queueName = $queueName;
self::$exchangeName = $exchangeName;

// 创建amqpconnection实例
$connect = new AMQPConnection($config);
if (! $connect->connect()) {
die("Cannot connect to the broker!\n");
}
$channel = new AMQPChannel($connect);

self::$connect = &$connect;
self::$channel = &$channel;
}

public function __destruct() {
self::$connect->disconnect();
}

/**
* 写入消息队列
* @param unknown $message
*/
public function setQueue($message = array(), $routing_key = ‘key_1‘) {
// 创建交换机
$exchange = new AMQPExchange(self::$channel);
$exchange->setName(self::$exchangeName);
$exchange->setType(AMQP_EX_TYPE_DIRECT);
$exchange->setFlags(AMQP_DURABLE | AMQP_AUTODELETE); // 交换器进行持久化，即 RabbitMQ 重启后会自动重建
$exchange->declare();

// 创建队列
$queue = new AMQPQueue(self::$channel);
$queue->setName(self::$queueName);
$queue->setFlags(AMQP_DURABLE);
// $queue->declare(); // AMQP 1.2.0 由 declare() 改为 declareQueue()

// 绑定交换机与队列，并指定路由键
$queue->bind($exchange->getName(), $routing_key);

$message = json_encode($message);
return $exchange->publish($message, $routing_key);
}

/**
* 读取消息队列中未处理的消息
* @return multitype:mixed
*/
public function getQueue() {
$queue = new AMQPQueue(self::$channel);
$queue->setName(self::$queueName);
$queue->setFlags(AMQP_DURABLE); // 持久化
// $messageCount = $queue->declare();

$result = array();
$i = 0;
// AMQP_NOPARAM / AMQP_AUTOACK
while($message = $queue->get(AMQP_NOPARAM)) {
$getBody = $message->getBody();
$result[$i] = json_decode($getBody, true);
$queue->ack($message->getDeliveryTag());

$i++;
if ($i >= 500) break;
}
return $result;
}
}

AMQP,布布扣,bubuko.com

AMQP

标签：php amqp amqpconnection amqpchannel amqpexchange amqpqueue

原文地址：http://blog.csdn.net/peng905/article/details/28425699

踩

(0)

评论一句话评论（0）

分享档案

更多>

2021年07月29日 (22)
2021年07月28日 (40)
2021年07月27日 (32)
2021年07月26日 (79)
2021年07月23日 (29)
2021年07月22日 (30)
2021年07月21日 (42)
2021年07月20日 (16)
2021年07月19日 (90)
2021年07月16日 (35)

周排行