1. 云图网首页
  2. 技术专区
  3. 智能运维

RabbitMQ学习之（四）_PHP操作RabbitMQ简单Demo详解程序员

• 智能运维

原理流程

生产者主要做的是：创建连接–>创建channel–>创建交换机对象–>发送消息

消费者主要做的是：创建连接–>创建channel–>创建交换机–>创建队列–>绑定交换机/队列/路由键–>接收消息

案例一：

生产者 

<?php   
$conn_args = array( 
    'host' => '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";

消费者

<?php   
$conn_args = array(   
    'host' => '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;

启动消费者程序，同时不断请求生产者程序，查看数据变化

RabbitMQ学习之（四）_PHP操作RabbitMQ简单Demo详解程序员

 案例二：

生产者

<?php   
$routingkey='key2'; 
//设置你的连接 
$conn_args = array('host' => '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); 
}

消费者

<?php 
$bindingkey='key2'; 
//连接RabbitMQ 
$conn_args = array( 'host'=>'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();

案例三：

生产者

<?php 
$conn_args = array( 
    'host' => '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();

消费者

<?php   
//配置信息 
$conn_args = array( 
    'host' => '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/wwwroot/default/publisher.php  //执行生产者 
php /data/wwwroot/default/consumer.php  //执行消费者

RabbitMQ学习之（四）_PHP操作RabbitMQ简单Demo详解程序员

附加删除队列：

RabbitMQ学习之（四）_PHP操作RabbitMQ简单Demo详解程序员RabbitMQ学习之（四）_PHP操作RabbitMQ简单Demo详解程序员

 

 

 

 

原创文章，作者：Maggie-Hunter，如若转载，请注明出处：https://blog.ytso.com/2353.html

赞 (0)
0
上一篇 2021年7月16日
下一篇 2021年7月16日

相关推荐

发表回复

登录后才能评论