封装php的RabbitMq

<?php ???return $arr = [ ???????‘RabbitMq‘ => [ ???????????// Rabbitmq 服务地址 ???????????‘host‘ => ‘127.0.0.1‘, ???????????// Rabbitmq 服务端口 ???????????‘port‘ => ‘5672‘, ???????????// Rabbitmq 帐号 ???????????‘login‘ => ‘guest‘, ???????????// Rabbitmq 密码 ???????????‘password‘ => ‘guest‘, ???????????‘vhost‘=>‘/‘ ???????] ???];

<?phpinclude dirname(__FILE__).‘/object.php‘;include dirname(__FILE__).‘/config.php‘;class RabbitMq implements object{ ???//保存类实例的静态成员变量 ???static private $_instance; ???static private $_conn; ???static private $amp ; ???static private $route = ‘key_1‘; ???static private $q ; ???static private $ex ; ???static private $queue; ???public static function getInstance(){ ???????global $arr; ???????if (!(self::$_instance instanceof self)) { ???????????self::$_instance = new self($arr[‘RabbitMq‘]); ???????????return self::$_instance; ???????} ???????return self::$_instance; ???} ???private function __construct($conn) ???{ ???????//创建连接和channel ???????$conn = new AMQPConnection($conn); ???????if(!$conn->connect()) { ???????????die("Cannot connect to the broker!\n"); ???????} ???????self::$_conn = new AMQPChannel($conn); ???????self::$amp = $conn; ???} ???/* * ????* ????* parm 交换机名 ????* parm 队列名 ????* ????* */ ???public function listen($exchangeName,$queuename){ ???????self::$queue = $queuename; ???????return $this->setExchange($exchangeName,$queuename); ???} ???//连接交换机 ???public function setExchange($exchangeName,$queueName){ ???????//创建交换机 ???????$ex = new AMQPExchange(self::$_conn); ???????self::$ex = $ex; ???????$ex->setName($exchangeName); ???????$ex->setType(AMQP_EX_TYPE_DIRECT); //direct类型 ???????$ex->setFlags(AMQP_DURABLE); //持久化 ???????$ex->declare(); ???????return self::setQueue($queueName,$exchangeName); ???} ???//创建队列 ???private static function setQueue($queueName,$exchangeName){ ???// ?创建队列 ???????$q = new AMQPQueue(self::$_conn); ???????$q->setName($queueName); ???????$q->setFlags(AMQP_DURABLE); ???????$q->declareQueue(); ???// 用于绑定队列和交换机 ???????$routingKey = self::$route; ???????$q->bind($exchangeName, ?$routingKey); ???????self::$q = $q; ???????return(self::$_instance); ???} ???/* * 消费者 * $fun_name = array($classobj,$function) or function name string * $autoack 是否自动应答 * * function processMessage($envelope, $queue) { ???????$msg = $envelope->getBody(); ???????echo $msg."\n"; //处理消息 ???????$queue->ack($envelope->getDeliveryTag());//手动应答 ???} */ ???public function run($func, $autoack = True){ ???????if (!$func || !self::$q) return False; ???????while(True){ ???????????if ($autoack) { ???????????????if(!self::$q->consume($func, AMQP_AUTOACK)){// ???????????????????self::$q->ack($envelope->getDeliveryTag()); ???????????????????echo 123; ???????????????} ???????????} ????????????self::$q->consume($func); ???????} ???} ???private static function closeConn(){ ???????self::$amp->disconnect(); ???} ???public function pushlish($msg){ ???????while (1) { ???????????sleep(1); ???????????if (self::$ex->publish(date(‘H:i:s‘) . "用户" . "注册", self::$route)) { ???????????????//写入文件等操作 ???????????????echo $msg; ???????????} ???????} ???} ???//__clone方法防止对象被复制克隆 ???public function __clone() ???{ ???????trigger_error(‘Clone is not allow!‘, E_USER_ERROR); ???}}

<?phpinclude dirname(__FILE__).‘/base.php‘;class Add{ ???public static function run(){ ???????$dbms=‘mysql‘; ????//数据库类型 ???????$host=‘127.0.01‘; //数据库主机名 ???????$dbName=‘test‘; ???//使用的数据库 ???????$user=‘root‘; ?????//数据库连接用户名 ???????$pass=‘admin‘; ?????????//对应的密码 ???????$dsn="$dbms:host=$host;dbname=$dbName"; ???????sleep(1); ???????try { ???????????$dbh = new PDO($dsn, $user, $pass); //初始化一个PDO对象 ???????????/*你还可以进行一次搜索操作 ???????????foreach ($dbh->query(‘SELECT * from FOO‘) as $row) { ???????????????print_r($row); //你可以用 echo($GLOBAL); 来看到这些值 ???????????} ???????????*/ ???????????$dbh = null; ???????} catch (PDOException $e) { ???????????die ("Error!: " . $e->getMessage() . "<br/>"); ???????}//默认这个不是长连接，如果需要数据库长连接，需要最后加一个参数：array(PDO::ATTR_PERSISTENT => true) 变成这样： ???????$db = new PDO($dsn, $user, $pass, array(PDO::ATTR_PERSISTENT => true)); ???????$sql = ‘INSERT INTO `test`.`t_reg`(`names`) VALUES (9)‘; ???????$row = $db->query($sql); ???????if(!$row){ ??????????return false; ???????} ???????????echo ‘OK‘; ???}}$consume = new Add();//tudo//$s = RabbitMq::getInstance()->listen(‘jiaohuanji‘,‘queue1‘)->run(array($consume,‘run‘)); 将run函数带入到consume里面作为回调 在consume里面增加$funname ,增加代码粘性$s = RabbitMq::getInstance()->listen(‘jiaohuanji‘,‘queue1‘)->run(array($consume,‘run‘));

<?phpinclude "base.php";RabbitMq::getInstance()->listen(‘jiaohuanji‘,‘queue1‘)->pushlish(‘请求已发送‘);

<?php ???interface object ???{ ???????public static function getInstance(); ???}