PHP中的AMQP类

一定要注意php安装AMQP的版本,版本不同使用的方法不一样。在官方网站就有2个版本的AMQP
第一版本:xxx,详细的url找不到了
第二版本:http://docs.php.net/manual/da/book.amqp.php
千万不要出现这种情况,找到一个官方的版本,然后按照example,怎么调试都不通….按照PHP安装 AMQP扩展 安装的AMQP扩展是最新的,现在和PHP官方给出的第二版本,也有一些区别。主要体现在exchange和queue中有个declare的方法,分别更改成declarExchange()和declarQueue().

一、AMQPConnection

::__construct ([ array $credentials = array() ] )

这方法比较简单,credentials的英文含义是“凭证”,但我喜欢把它理解为config。这些配置包括如下:
host、vhost、port、login、password、read_timeout、write_timeout、connect_timeout,如果config省略,php会使用默认的这5个值,初始化,强烈建议带上config初始化,其中read_timeout、write_timeout、connect_timeout这三个值是最新版本才有的,PHP官方现在还没有,单位都是(秒)

::connect ( void )  、  ::pconnect ( void )
::disconnect ( void )   、::pdisconnect ( void )
::reconnect ( void ) 、::preconnect ( void )  
#注意和__construct 的区别,该方法不带参数。
::isConnected ( void )
::isPersistent ( void )
#是否持久链接为什么不是isPconnected ?-_-

如果使用持久链接,在类方法__destruct中释放(断开链接)

::setHost ( string $host )
::setPort ( int $port )
::setVhost ( string $vhost )
::setLogin ( string $login )
::setPassword ( string $password )
::getHost ( void )
::getPort ( void )
::getVhost ( void )
::getLogin ( void )
::getPassword ( void )
还有类似的::setTimeout ( int $timeout ) 、::getTime ( void )、::getReadTimeout ( void )、::setReadTimeout ( int $timeout )、::getWriteTimeout (void ) 、setWriteTimeout ( int $timeout)

上面的这些方法,就不细说了,还有2个不常见的方法

::getUsedChannels ( void )  #在链接会话期间最后一次使用的channel id
::getMaxChannels ( void )   #单次链接能创建最大channel数量

二、AMQPChannel

为了一个AMQP的链接,创建一个channel instance(通道实例),这里需要了解一下几个概念,以及她们之间的关系,本人做了一个罗列,水平有限,不对的地方请指正。

client :客户端
Broker:英文含义是代理,这里指代理服务器,也是通常说的server,也可以指Rabbit cluster的节点,
也就是在AMQPConnection的配置文件中的host、port
virtual_host:虚拟host,这个可以理解为在Broker中,虚拟出来的消息处理模块。一个Broker中
可以创建多个virtual_host.
exchange:逻辑上属于virtual_host的子模块,负责消息转发到queue中。
queue:逻辑上属于virtual_host的子模块。
channel:通道。链接virtual_host的通道。
routingkey:

virtual_host和channel的设置都是为了增加并发性能,降低资源消耗的。
参考文章:RabbitMQ中的AMQP协议规范

::__construct ( AMQPConnection $amqp_connection ) #创建channel
::isConnected ( void )
#这里和AMQPConnection::isConnected一样.

::getConnection ( void )
#返回的结果是AMQPConnection,从概念上个人理解,channel是建立在connection之上的,是个独立的链接。connection是指物理的连接,一个client与一个server之间有一个连接;一个连接上可以建立多个channel,可以理解为逻辑上的连接。一般应用的情况下,有一个channel就够用了,不需要创建更多的channel

::getChannelId ( void )
#返回channel id,在PHP中循环创建通道,发现通道ID始终只有1和2,不知是否和服务器的设置有关?

::setPrefetchSize ( int $size )、::getPrefetchSize ( void )
这个地方有个坑,设置prefetchSize不等于0,会提示not_implemented: “prefetch_size!=0 (65535)”,也就是说目前prefetchSize只能设置为0,还未实现不等于0的情况。

::setPrefetchCount ( int $count )、::getPrefetchCount ( void )
这个参数设置,接收消息端,接收的最大消息数量(包括使用get、consume),一旦到达这个数量,客户端不在接收消息。0为不限制。默认值为3.

::qos ( int $size , int $count )
同时设置prefetchSize、prefetchCount,但prefetchSize目前只能设置为0

 ::basicRecover ( boolen )
默认为true,没有确认的消息会被再次投递,这个地方和消息确认机制有关,后面再来说这个问题。

三、 AMQPExchange

::__construct ( AMQPChannel $amqp_channel )

::setName ( string $exchange_name )、::getName ( void )
这个命令的最好使用数字、字母、- _ . 组合(区分大小写,但最好使用小写)
也不要使用amq带头(内置的exchange都是使用amq.开头)
如果申明一个已存在的exchange name,如果type不一致会抛出异常.

::setType ( string $exchange_type )、::getType ( void )
这个是rabbitmq的核心,exchange的类型,决定了rabbit实现何种业务模型。
这个地方尤其要注意,AMQ本身没有默认的exchangeType,所以在申明exchange必需要设置type,否则会抛出异常 Could not declare exchange. Exchanges must have a type,
目前exchange支持4种类型  AMQP_EX_TYPE_DIRECT、 AMQP_EX_TYPE_FANOUT、AMQP_EX_TYPE_HEADERS 、 AMQP_EX_TYPE_TOPIC
可参考:RabbitMQ AMQP 消息模型攻略
记忆方法:amqp_exchange_type_direct、fanout、topic、headers

 ::setFlags ( int $flags )、::getFlags ( void )
默认AMQP_NOPARAM; 可选AMQP_DURABLE, AMQP_PASSIVE、AMQP_AUTODELETE

  1. passive:声明一个已存在的交换器的,如果不存在将抛出异常,这个一般用在consume端。因为一般produce端创建,在consume端建议设置成AMQP_PASSIVE,防止consume创建exchange
  2. durable:持久exchange,该交换器将在broker重启后生效,一般常使用该选项.
  3. auto_delete:该交换器将在没有消息队列绑定时自动删除。一个从未绑定任何队列的交换器不会自动删除。解释有点绕。说明下吧,当有队列bind到auto_delete的交换器上之后,删除该队列。此时交换器也会删除。一般创建为临时交换器。

::setArgument ( string $key , mixed $value )
::setArguments ( array $arguments )
::getArgument ( string $key )
::getArguments ( void )

::declareExchange( void ) //和PHP的官方有差别

::delete ($exchangeName = null, $flags = AMQP_NOPARAM) //和PHP的官方有差别
未指定$exchangeName,删除当前链接的exchange但无法删除system中内置的exchange.

::bind($exchange_name, $routing_key = ”, array $arguments = array())  //和PHP的官方有差别
::unbind($exchange_name, $routing_key = ”, array $arguments = array())  //和PHP的官方有差别
可以将消息同时发送到多个exchange,稍后补充实际的用法。

::publish ( string $message , string $routing_key [, int $flags = AMQP_NOPARAM [, array $attributes = array() ]] ) //发送消息方法 $attributes参考:RabbitMQ的PHP教程之RPC (六)

参数:$routint_key,在fanout下忽略。在direct类型下,如果该值为空,则bind到exchange都能收到消息,这个有点像fanout的广播模式。

参数:$flags可选值 AMQP_MANDATORY、AMQP_IMMEDIATE,默认:AMQP_NOPARAM,此处一般都设置为false,本人还未测出这2个flag的区别,这里有一篇文章关于这2个flag的介绍:
http://blog.csdn.net/jiao_fuyou/article/details/21594947

::getChannel()、getConnection() //不再介绍,见上面

四、AMQPQueue

::__construct(AMQPChannel $amqp_channel) //有PHP官方有区别

::setName ( string $queue_name )、::getName ( void ) //队列名称

::setFlags ( int $flags )、::getFlags ( void ) //重点
可选参数:AMQP_DURABLE, AMQP_PASSIVE,AMQP_EXCLUSIVE, AMQP_AUTODELETE
默认:auto_delete

auto_delete:当队列中有消费者,则队列存在,当没有消费者链接,则队列删除

durable:持久化,队列不删除,注意仅仅是队列持久,消息不持久(消息的持久在publish时的增加属性delivery_mode)。消费的消息,从队列里删除,未消费的消息保存在队列中,不需要关注是否有消费者。最实用

passive:声明一个1个已存在的队列。意义不大,如果队列不存在会抛出异常。

exclusive:排他队列,如果一个队列被声明为排他队列,该队列仅对首次声明它的连接可见,并在连接断开时自动删除。这里需要注意三点:

  1. 排他队列是基于连接可见的,同一连接的不同信道是可以同时访问同一个连接创建的排他队列的。
  2. “首次”,如果一个连接已经声明了一个排他队列,其他连接是不允许建立同名的排他队列的,这个与普通队列不同。
  3. 即使该队列是持久化的,一旦连接关闭或者客户端退出,该排他队列都会被自动删除的。这种队列适用于只限于一个客户端发送读取消息的应用场景。

::setArgument ( string $key , mixed $value )
::setArguments ( array $arguments )
::getArgument ( string $key )
::getArguments ( void )

::declareQueue ( void )
::bind ( string $exchange_name , string $routing_key )
::unbind ( string $exchange_name , string $routing_key )

::get ([ int $flags ] )
//非阻塞,从队列中检索下一个可用的消息,flags为空默认为amqp.auto_ack(在php.ini中可用设置默认值),可选参数:AMQP_AUTOACK

::consume ( callable $callback [, int $flags = AMQP_NOPARAM ] )
//阻塞,$callback支持数组的写法,flags为空默认为amqp.auto_ack(在php.ini中可用设置默认值),可选参数:AMQP_AUTOACK

::ack ( int $delivery_tag [, int $flags = AMQP_NOPARAM ] )  //复杂的业务逻辑,最好采用手动应答,防止消费处理业务逻辑复杂时,消息丢失。指定务必指定为AMQP_NOPARAM, 本人测试使用参数 AMQP_MULTIPLE,达不到预期的效果。此处有待高人解决。

更人性化的应答,在使用get、consume时,指定AMQP_NOPARAM时(未指定参数,则会调用PHP.ini的默认的设置),此时未对消费的消息做应答。在处理完复杂的业务逻辑后,使用ack做应答。此时队列中,会删除该条消息,如未应答,则会一直存在队列中。

::nack($delivery_tag, $flags = AMQP_NOPARAM)、reject($delivery_tag, $flags = AMQP_NOPARAM)
用法ack相反,表示not ack.

::purge ( void )
清空队列中的消息。

::delete ( void )
删除队列

::getChannel()、getConnection() //不再介绍,见上面

::cancel ([ string $consumer_tag = “” ] ) //暂时不知该如何使用

五、AMQPEnvelope

该类没有set的方法,各种get

::getBody ( void )
::getRoutingKey ( void )
::getDeliveryTag ( void )
::getDeliveryMode ( void )
::getExchangeName ( void )
::isRedelivery ( void )
::getContentType ( void )
::getContentEncoding ( void )
::getType ( void )
::getTimeStamp ( void )
::getPriority ( void )
::getExpiration ( void )
::getUserId ( void )
::getAppId ( void )
::getMessageId ( void )
::getReplyTo ( void )
::getCorrelationId ( void )
::getHeaders ( void )
::getHeader($header_key).
相关阅读:

  1. RabbitMQ的原理与操作示例
  2. RabbitMQ AMQP 消息模型攻略
  3. PHP中的AMQP类
  4. RabbitMQ的PHP教程之入门 (一)
  5. RabbitMQ的PHP教程之工作队列 (二)
  6. RabbitMQ的PHP教程之发布/订阅(三)
  7. RabbitMQ的PHP教程之Routing (四)
  8. RabbitMQ的PHP教程之topic (五)
  9. RabbitMQ的PHP教程之RPC (六)

PHP中的AMQP类》上有2条评论

  1. Pingback引用通告: RabbitMQ的PHP教程之入门 (一) | 精彩每一天

  2. Pingback引用通告: RabbitMQ的PHP教程之Routing (四) | 精彩每一天

发表评论

电子邮件地址不会被公开。 必填项已用*标注

您可以使用这些HTML标签和属性: <a href="" title=""> <abbr title=""> <acronym title=""> <b> <blockquote cite=""> <cite> <code> <del datetime=""> <em> <i> <q cite=""> <strike> <strong>