RabbitMQ学习之（四）_PHP操作RabbitMQ简单Demo

RabbitMQ学习之（四）_PHP操作RabbitMQ简单Demo

原理流程

生产者主要做的是：创建连接-->创建channel-->创建交换机对象-->发送消息消费者主要做的是：创建连接-->创建channel-->创建交换机-->创建队列-->绑定交换机/队列/路由键-->接收消息

案例一：

生产者

'127.0.0.1', 'port' => '5672', 'login' => 'admin', 'password' => 'admin', 'vhost'=>'/');$content = isset($_GET['content']) ? htmlspecialchars($_GET['content']) : '请输入消息';//创建连接和channel $conn = new AMQPConnection($conn_args); if (!$conn->connect()) { die("Cannot connect to the broker!\n"); } $channel = new AMQPChannel($conn); //创建交换机 $e_name = 'e_linvo'; //交换机名 $ex = new AMQPExchange($channel); $ex->setName($e_name); $ex->setType(AMQP_EX_TYPE_DIRECT); //direct类型 $ex->setFlags(AMQP_DURABLE); //持久化 $ex->declare(); //echo "Send Message:".$ex->publish("TEST MESSAGE，key_1 by xust" . date('H:i:s', time()), 'key_1')."\n"; //echo "Send Message:".$ex->publish("TEST MESSAGE，key_2 by xust" . date('H:i:s', time()), 'key_2')."\n"; $ex->publish($content, 'key_1'); echo "finish\n";

消费者

'127.0.0.1', 'port' => '5672', 'login' => 'admin', 'password' => 'admin', 'vhost'=>'/' ); $e_name = 'e_linvo'; //交换机名 $q_name = 'q_linvo'; //队列名 $k_route = 'key_2'; //路由key //创建连接和channel $conn = new AMQPConnection($conn_args); if (!$conn->connect()) { die("Cannot connect to the broker!\n"); } $channel = new AMQPChannel($conn); //创建交换机 $ex = new AMQPExchange($channel); $ex->setName($e_name); $ex->setType(AMQP_EX_TYPE_DIRECT); //direct类型 $ex->setFlags(AMQP_DURABLE); //持久化 $ex->declare(); //创建队列 $q = new AMQPQueue($channel); $q->setName($q_name); $q->setFlags(AMQP_DURABLE); //持久化 $q->declare(); //最好队列object在这里declare()下，否则如果是新的queue会报错 //绑定交换机与队列，并指定路由键，可以多个路由键 $q->bind($e_name, 'key_1'); //$q->bind($e_name, 'key_33'); //阻塞模式接收消息 echo "Message:\n"; $q->consume('processMessage', AMQP_AUTOACK); //自动ACK应答 $conn->disconnect(); /** * 消费回调函数 * 处理消息 */ function processMessage($envelope, $queue) { var_dump($envelope->getRoutingKey()); $msg = $envelope->getBody(); mysql_insert($msg); } //消息入库 function mysql_insert($json){ $pdo = new PDO("mysql:host=localhost;dbname=rabbitmq","root","000000",array(PDO::ATTR_PERSISTENT => true) ); $sql = "insert into rabbitmq_table(sJson) values ('{$json}')"; echo $sql."\n"; if ($pdo->exec($sql)){ echo "insert success\n"; } else { echo "error\n"; } }

表结构

DROP TABLE IF EXISTS `rabbitmq_table`;CREATE TABLE `rabbitmq_table` ( `id` int(11) NOT NULL AUTO_INCREMENT, `sJson` varchar(255) NOT NULL DEFAULT '', PRIMARY KEY (`id`)) ENGINE=MyISAM DEFAULT CHARSET=utf8;

案例二：

生产者

'localhost', 'port' => '5672', 'login' => 'admin', 'password' => 'admin');$conn = new AMQPConnection($conn_args);if ($conn->connect()) { echo "链接成功\n";}else { echo "链接失败 \n ";}//你的消息$message = json_encode(array('消息测试','My RabbitMQ-PHP','消息队列'));//创建channel$channel = new AMQPChannel($conn);//创建exchange$ex = new AMQPExchange($channel);$ex->setName('exchange');//创建名字$ex->setType(AMQP_EX_TYPE_DIRECT);$ex->setFlags(AMQP_DURABLE);//$ex->setFlags(AMQP_AUTODELETE);//echo "exchange status:".$ex->declare();echo "exchange status:".$ex->declareExchange();echo "\n";for($i=0;$i<10;$i++){ if($routingkey=='key2'){ $routingkey='key'; }else{ $routingkey='key2'; } $ex->publish($message,$routingkey);}

消费者

'127.0.0.1' , 'port'=> '5672', 'login'=>'admin' , 'password'=> 'admin','vhost' =>'/');$conn = new AMQPConnection($conn_args);$conn->connect();//设置queue名称，使用exchange，绑定routingkey$channel = new AMQPChannel($conn);$q = new AMQPQueue($channel);$q->setName('queue2');$q->setFlags(AMQP_DURABLE);$q->declare();$q->bind('exchange',$bindingkey);//消息获取$messages = $q->get(AMQP_AUTOACK) ;if ($messages){var_dump(json_decode($messages->getBody(), true ));}$conn->disconnect();

案例三：

生产者

'127.0.0.1', 'port' => '5672', 'login' => 'admin', 'password' => 'admin', 'vhost'=>'/');$e_name = 'e_linvo'; //交换机名//$q_name = 'q_linvo'; //无需队列名$k_route = 'key_1'; //路由key//创建连接和channel$conn = new AMQPConnection($conn_args);if (!$conn->connect()) { die("Cannot connect to the broker!\n");}$channel = new AMQPChannel($conn);//消息内容$message = "我的RabbitMQ走通了~~";//创建交换机对象$ex = new AMQPExchange($channel);$ex->setName($e_name);//发送消息//$channel->startTransaction(); //开始事务for($i=0; $i<5; ++$i){ echo "Send Message:".$ex->publish($message, $k_route)."\n";}//$channel->commitTransaction(); //提交事务$conn->disconnect();

消费者

'127.0.0.1', 'port' => '5672', 'login' => 'admin', 'password' => 'admin', 'vhost'=>'/');$e_name = 'e_linvo'; //交换机名$q_name = 'q_linvo'; //队列名$k_route = 'key_1'; //路由key//创建连接和channel$conn = new AMQPConnection($conn_args);if (!$conn->connect()) { die("Cannot connect to the broker!\n");}$channel = new AMQPChannel($conn);//创建交换机$ex = new AMQPExchange($channel);$ex->setName($e_name);$ex->setType(AMQP_EX_TYPE_DIRECT); //direct类型$ex->setFlags(AMQP_DURABLE); //持久化echo "Exchange Status:".$ex->declare()."\n";//创建队列$q = new AMQPQueue($channel);$q->setName($q_name);$q->setFlags(AMQP_DURABLE); //持久化echo "Message Total:".$q->declare()."\n";//绑定交换机与队列，并指定路由键echo 'Queue Bind: '.$q->bind($e_name, $k_route)."\n";//阻塞模式接收消息echo "Message:\n";while(True){ $q->consume('processMessage'); //$q->consume('processMessage', AMQP_AUTOACK); //自动ACK应答}$conn->disconnect();/*** 消费回调函数* 处理消息*/function processMessage($envelope, $queue) { $msg = $envelope->getBody(); echo $msg."\n"; //处理消息 $queue->ack($envelope->getDeliveryTag()); //手动发送ACK应答}

然后分别命令行执行两个程序，消费者程序处于监听状态，多次执行生产者测试，观察效果

php /data/ //执行生产者php /data/ //执行消费者

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。