在php中使用rabbitmq从概念到实战

快来打我* 2022-06-02 02:24 128阅读 0赞

首先要知道amqp协议是做什么的，amap是一个提供统一消息服务的应用层标准协议，基于此协议的客户端与消息中间件可传递消息,rabbitmq就是实现了这个协议的一个软件。  
消息者

<?php //创建连接--> //创建channel--> //创建交换机--> //创建队列--> //绑定交换机/队列/路由键--> //接收消息 $conn_args = array( 'host' => '127.0.0.1', 'port' => '5672', 'login' => 'guest', 'password' => 'guest', 'vhost'=>'/' ); //交换机名消息是不能直接发送到队列，它需要发送到交换器 $e_name='e_zcj';//交换机名 $q_name="q_zcj";//队列名 $k_route="key_1";//路由key //连接到borker和实例一个channel $conn=new AMQPConnection($conn_args); if(!$conn->connect()){ // AMQP服务器端可称为broker die("Cannot connect to the broker"); } //实例channel //大部分的业务操作是在rabbitmq中提供Channel这个接口中完 //成的，在php相应的扩展中的Amapchanenl需要连接的实例包 //括定义Queue、定义Exchange、绑定Queue与Exchange、发布 //消息 $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($e_name);//队列名 //持久化 这个方法做了什么去看c源码吧，， $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"); } function processMessage($envelope,$queue){ $msg=$envelope->getBody(); echo $msg."\n"; print_r($envelope)."\n"; //发送ack答应 $queue->ack($envelope->getDeliveryTag()); }

生产者

<?php // $conn_args = array( 'host' => '127.0.0.1', 'port' => '5672', 'login' => 'guest', 'password' => 'guest', 'vhost'=>'/' ); $e_name = 'e_zcj'; $k_route = 'key_1'; $conn = new AMQPConnection($conn_args); if (!$conn->connect()) { die("Cannot connect to the broker!\n"); } $channel = new AMQPChannel($conn); $message = "TEST MESSAGE! 郑程骏 is 天才"; $ex = new AMQPExchange($channel); $ex->setName($e_name); echo "Send Message:".$ex->publish($message, $k_route)."\n"; $conn->disconnect();