Mar2

【原创】PHP RabbitMQ消息队列应用高级教程

Author: leeon  Click: 1835   Comments: 0 Category: 架构  Tag: rabbitmq,php,amqp

    最近在看rabbitmq的设计哲学和相关概念,有很多需要总结的学习心得可以分享一下。相比以往使用的轻量级高性能beanstalkd队列服务组件,rabbitmq提供了更更丰富的功能,当然他也是可以做到数据持久化的。

    参考了不少资料,但是在PHP下使用rabbitmq的案例并不多,有些代码也略显陈旧,今天主要从一下几个方面开始介绍和总结。

 1. php-amqp 扩展安装

 2. exchange,queue,routing_key, name概念的相关梳理

 3. php-amqp相关API解析

 4. rabbitmq 持久化使用

   

php-amqp扩展安装

    rabbitmq对应php的api其实有好几个,从性能上当然首选pecl的最佳。php-amqp是抽象的消息队列扩展,并不是只针对rabbitmq提供访问服务,只要是支持AMQP协议的消息队列服务程序,都能通过php-amqp插件获得访问能力。
    在安装php-amqp扩展的时候需要先安装了librabbitmq依赖,如果你正在使用centos的系统,我的建议是最好不要使用yum仓库中提供的librabbitmq包,最好手工去下载最新版本的文件进行手工编译安装,因为楼主在用yum安装了ibrabbitmq-devel的时候发现了兼容性问题,导致php始终连接不上rabbitmq。
    官方源文件rabbit-c的安装地址在https://github.com/alanxz/rabbitmq-c 这里,git下来后记得在configure的时候制定prefix安装路径,这样在pecl安装amqp扩展的时候才能让php-amqp包识别到librabbitmq的存在,如果直接configure的话会导致找不到librabbitmq相关文件。
  rabbitmq-安装成功后,我们假定安装路径为/usr/local/librabbitmq ,那么接着安装php-amqp的时候步骤如下:
pecl install amqp
此时提示如下信息时:Set the path to librabbitmq install prefix [autodetect] : 
输入刚才指定的安装路径/usr/local/librabbitmq 回车即可。
编译安装完成后将extension=amqp.so 放入php.ini文件中。

注意在使用rabbitmq的时候默认的账号是guest,但是这个账号限定只能本地访问操控,因此在我们使用rabbitmq远程访问的时候最好新建独立的账号,并进行相应授权,这个操作使用rabbitmqctl命令来控制。

exchange,queue,routing key, name概念的相关梳理

exchange相当于生产者的接收者,同时也相当于接收者数据派送者,类似于一个消息中转站。通常exchange的出现需要和queue配对,并且需要事先申明。exchange类型又主要分为三类:direct(定向直接投递),fanout(广播投递),topic(模糊主题投递)。exchange在申明创建的时候可以定义一个name名字,也可以不定义,不定义的时候rabbitmq会默认赋值一个固定的name。同时对exchange的属性也做了定义:AMQP_DURABLE(持久化,rabbitmq重启后exchange依然存在)和AMQP_PASSIVE(不自动创建exchange,而是检查exchange是否存在) 。

当我们需要检查一个name为Test的exchange是否存在的时候我们可以使用如下代码:
[code="php"]
$connection = new AMQPConnection();
$connection->setHost('127.0.0.1');
$connection->setLogin('leeon');
$connection->setPassword('leeon');
$connection->connect();
$channel = new AMQPChannel($connection);
$exchange = new AMQPExchange($channel);
$exchange->setName('Test'); // 设定要检查的exchange的名字
$exchange->setType(AMQP_EX_TYPE_DIRECT);
$exchange->setFlags(AMQP_PASSIVE); // 使用PASSIVE参数
$exchange->declareExchange();
[/code]

如果Test存在则不会抛出任何异常,如果不存在则会抛出异常
Fatal error: Uncaught AMQPExchangeException: Server channel error: 404, message: NOT_FOUND - no exchange 'Test' in vhost '/'

AMQP_DURABLE参数的设定将会保证rabbitmq的重启不会导致申明定义的exchange丢失。例如我们正常创建了一个exchange后可以在rabbitmq的web控制端看到此状态:


queue的定义主要在消费者这里做定义,定义queue的时候也可以定义一个name,这个name仅仅是对队列名字的命名,不同队列的名字name是不一样的。同样queue的属性定义也有好几个:
AMQP_DURABLE (持久化)
AMQP_PASSIVE (不自动创建queue,而是检查queue是否存在,不存在抛出异常)
AMQP_EXCLUSIVE (排他队列,以此定义命名的队列只有一个)
AMQP_AUTODELETE(当queue释放的时候是否自动删除这个队列在rabbitmq中的记录)

routing key 这个好比一个协商key,当发布者和消费者有协商一致的key策略的时候,消费者就可以合法从生产者手中获取数据。这个routing key主要当exchange设定为direct和topic模式的时候使用,fanout模式不使用routing key,那么这三个模式又有何区别呢?我们来看看如下三个图:




这三个图分别展示了direct,fanout,topic的消息投递规则。

php-amqp相关API解析

php-amqp在最近的版本中改动比较大,例如对于exchange和queue的delcare方法做了改变,修改为了declareExchange和declareQueue。因此网上很多相关的PHP教程事例代码并不能直接拿来学习了。
这里我们主要对publish,consume两个api方法进行讲解。
publish方法:
[code="php"]
public function publish(
$message,
$routing_key = null,
$flags = AMQP_NOPARAM,
array $attributes = array()
) {
}
[/code]
publish方法在生产者的代码逻辑中使用,message参数是需要传递的具体消息内容,这里我们可以通过序列化或者json字符串的形式进行复杂数据的封装。routing_key参数就是前文所说的exchange和queue之间的路由协商key。当我们使用fanout模式的时候,routing_key定义为null。 flags参数包含了两个定义:AMQP_MANDATORY 和 AMQP_IMMEDIATE. 这两个参数在当前php-amqp(1.8.0)版本中并没有实际的作用,因为php-amqp里面的实现调用的是异步模式,并不会等待结果的返回。
[code="cpp"]
/* NOTE: basic.publish is asynchronous and thus will not indicate failure if something goes wrong on the broker */
int status = amqp_basic_publish(
channel_resource->connection_resource->connection_state,
channel_resource->channel_id,
(Z_TYPE_P(exchange_name) == IS_STRING && Z_STRLEN_P(exchange_name) > 0 ? amqp_cstring_bytes(Z_STRVAL_P(exchange_name)) : amqp_empty_bytes), /* exchange */
(key_len > 0 ? amqp_cstring_bytes(key_name) : amqp_empty_bytes), /* routing key */
(AMQP_MANDATORY & flags) ? 1 : 0, /* mandatory */
(AMQP_IMMEDIATE & flags) ? 1 : 0, /* immediate */
&props,
php_amqp_long_string(msg, msg_len) /* message body */
);
[/code]

我们重点需要关注的是attributes的参数设定,这里是一个数组,可以同时定义多个属性,包括如下信息:
content_type: MIME类型定义,例如gzip
message_id: 消息唯一识别id
user_id: 用来记录这个消息的发送者唯一识别id
app_id: 用来记录这个消息的生产者的唯一识别id
delivery_mode: 消息的传送模式 1代表非持久态 2代表持久态(如果消息需要高可用,例如rabbitmq宕机后数据不丢失,name这里就要设置成2)
priority: 消息的优先级,设定值从0到9
timestamp: 消息当前被发送的时间的时间戳
expiration: 消息的过期时间,这个时间单位是毫秒
type: 消息的类型名字
reply_to: 通常用在RPC中,告知需要回复的队列
headers: 消息头部数据表,可以定义多个自定义数据。

consume方法:
[code="php"]
public function consume(
callable $callback = null,
$flags = AMQP_NOPARAM,
$consumerTag = null
) {
}
[/code]

consume方法需要传递一个回调函数,这里不支持以前我们php中的惯用法,传递一个方法名,而是需要传递一个匿名含函数的变量值。flags参数可选AMQP_AUTOACK设定,如果设定AMQP_AUTOACK则当消费者接收到消息后向rabbitmq自动确认消息接收成功,如果不设定AMQP_AUTOACK,则需要我们在代码中调用ack方法来确认消息接收状态。例如如下一段代码:
[code="php"]
$connection = new AMQPConnection();
$connection->setHost('127.0.0.1');
$connection->setLogin('leeon');
$connection->setPassword('leeon');
$connection->connect();

$channel = new AMQPChannel($connection);
$exchange = new AMQPExchange($channel);

$queue = new AMQPQueue($channel);
$queue->setName('GGGGGG');
$queue->setFlags(AMQP_EXCLUSIVE);
$queue->declareQueue();

$queue->bind('topic','aaa.#');
$callback = function (AMQPEnvelope $message, AMQPQueue $queue) {
echo $message->getBody() . PHP_EOL;
$queue->ack($message->getDeliveryTag());
};

$queue->consume($callback);
[/code]

     当我们在编写消费者代码的时候,应当都需要调用bind方法来对exchange进行绑定,注意exchange的name一定要已经创建,否则调用bind方法会抛出异常。当使用topic和direct模型时,则需要制定明确的routing key。所以为什么大家在网上看到的PHP AMQP示例代码中,对于生产者和消费者都会申明一次exchange的缘由了,因为PHP的无状态性,为了保证程序代码的完整性,我们需要养成一个习惯,在生产者和消费者中都申明一次exchange和queue。
    为何也要同时申明queue呢?因为消息的投递一定要有明确的queue队列来接收,当我们使用publish来发布后如果找不到合法的queue,这条消息就被rabbitmq丢弃掉了。

rabbitmq 持久化使用

rabbitmq的持久化分为三个部分:exchange,queue 和消息体。当需要一个高可用的消息队列服务时,我们需要同时对exchange和queue在创建的时候申明durable状态,在生产者发送的消息中定义delivery_mode为2即可。例如如下代码:
[code="php"]
$connection = new AMQPConnection();
$connection->setHost('127.0.0.1');
$connection->setLogin('leeon');
$connection->setPassword('leeon');
$connection->connect();
$channel = new AMQPChannel($connection);
$exchange = new AMQPExchange($channel);
$exchange->setName('ex');
$exchange->setType(AMQP_EX_TYPE_DIRECT);
$exchange->setFlags(AMQP_DURABLE);
$exchange->declareExchange();

try {

$queue = new AMQPQueue($channel);
$queue->setName('AAA');
$queue->setFlags(AMQP_DURABLE);
$queue->declareQueue();
$queue->bind('ex','keykey');

var_dump( $exchange->publish(
'hello world!',
'keykey',
AMQP_NOPARAM,
[
'delivery_mode'=>2
]
));

}catch (Exception $e){
var_dump($e);
}
[/code]

Mar2

【原创】墨水湖绿道骑行路线

Author: leeon  Click: 1025   Comments: 0 Category: 生活  Tag: 墨水湖绿道,骑行

     最近环湖骑行了一下墨水湖绿道,发现修的相当不错了,风景绝对好,在武汉三环内能有如此大规模的绿道除了东湖应该当属墨水湖绿道最大了。不过墨水湖绿道本身是没有完全完工的,迫不及待的小伙伴们可以参考我的骑行路线进行环湖。墨水湖绿道北岸和南岸完工进度差不多在90%了,但是东西两岸工程进度还不行。西岸完工差不多70%,东岸的绿道目前还不能说是绿道。

北岸的出口在动物园旁边的东方华尔兹后面路过,然后达到江城大道和墨水湖北路的交叉口位置。从北岸出来后沿着墨水湖北路到达马沧湖路,继续骑行到江提中路,到江欣苑路后右转直行,可以看到这里有个没有正式通的路,旁边是碧水科技有限公司。这段没有通的路还在施工,但不妨碍自行车过去,按照图上所示过去就可以看到南岸绿道的接线处了。剩下的路线就按照图示骑行吧。对了最后说一句整个环湖绿道全长差不多在15公里左右,除了靠近汉阳区政府这一段人比较多,其他的地方人少路宽敞。

Feb16

【原创】如何通过phpstorm查看某一行代码的变更记录

Author: leeon  Click: 1100   Comments: 0 Category: 其他  Tag: phpstorm,php,git

多人维护的工程项目,往往对历史代码有些不理解,phpstorm在新版本中集成了一个很有用的功能

“Show History for Selection”

只要我们对有疑问的代码进行框选,然后右键点击git->Show History for Selection ,phpstorm就会去git history中去检查这行代码在历史记录中出现的状态。

Feb15

【原创】最佳PHP框架选择(phalcon,yaf,laravel,thinkphp,yii)

Author: leeon  Click: 1772   Comments: 1 Category: php  Tag: php,yaf,phalcon,laravel,thinkphp,yii

     最近面试了不少人,绝大部分使用的是thinkphp和laravel这两个框架,真正高性能的框架yaf和phalcon反而没有人使用。我认为使用框架主要基于这几点因素:

1. 工程结构规范化

2. 代码格式规范化

3. 功能抽象化

4. 逻辑封装化

5. 底层核心透明化

     tp,laravel,yii的简单实用的上手难度不高,但是复杂封装又各有特点需要深入学习,虽然可以快速通过其丰富的组件功能来构建一个基本的网站,但是后期的学习曲线也比较高。不管是用哪种框架,都是基于MVC模式来设计。本质上都是通过统一的路由管理器来派发不同的控制器任务。本文我将通过直接的性能测评来选出PHP7版本下认为在性能和工程化上最优的PHP框架。

 框架 版本  说明 
 yaf 3.0.4   通过yaf自带的yaf_cg来初始化一个基本工程
 thinkphp 5.0.6   composer创建一个基本工程,关闭debug
 phalcon  3.0.3  通过phalcon-devtools工具创建一个基本工程
 laravel  5.4  composer创建一个基本工程,关闭debug
 yii  2.0.11  yii手工下载basic版本工程,关闭debug

      我们通过将5个工程分别部署成5个网站,并修改里面的首页控制器,均改为输出到页面“hello”字段来评测裸框架下的基准性能。基础测试环境采用负载稳定的树莓派3来运行,并通过另外一台树莓派2来进行ab压力测试。fpm进程设定为static模式,并设定fpm进程数为5个,opcache设为开启状态。

ab运行命令类似如下,并连续采样10次来分析持续压力下的框架稳定性。

ab -k  -c 20 -n500 -q http://127.0.0.1/

压力测试结果如下:

laravel phalcon 原生php yaf thinkphp yii

87.9 670.66 1258.67 1061.76 545.93 61.65

100.4 924.2 1331.67 1126.23 618.78 65.88

81.19 921.87 1352.38 1129.91 608.41 77.84

111.39 930.88 1355.48 1159.62 601.55 79.05

53.89 929.71 1367.68 1165.19 607.24 72.21

94.67 918.72 1482.87 1107 607.24 82

62.21 932.51 1374.17 1130.23 610.72 76.42

85.63 925.01 1674.64 1124.18 609.11 69.82

67.05 913.12 1828.11 1128.22 608.72 75.67

89.21 879.26 1809.36 1159.38 601.18 77.93

     我们可以看到原生的PHP依然是相当强悍的,yaf的性能比phalcon高,这是因为yaf框架只 包含最基本的路由配置加载,并未有任何高级封装,相比phalcon框架丰富的组件,yaf在内存加载启动上必然比phalcon要轻量化。我们可以看到PHP实现的thinkphp性能表现不俗,但lavarel和yii的性能确实有点差强人意了。

    yaf属于超轻量级的c级别裸框架,只集成了基本的视图和路由控制,如果要完成一个中等以上规模的网站的话,还需要集成非常多的第三方类库,如果在线上运营环境中使用,会因为php加载大量的第三方类库而影响到性能。

    phalcon基于c的实现几乎完整的实现了yii,laravel,thinkphp具备的功能特点,同时在性能上完胜了这三个框架,在性能和功能上做了一个比较好的平衡。对于一个可扩展架构的高性能PHP框架而言,phalcon具备了比PHP实现的框架更好的稳定性和开发效率。phalcon虽然在入门门槛上偏高,但是我们只要熟悉了他的框架结构,我们可以很快速的去实现任何web服务业务模型。phalcon具体的介绍将在下一章节中进行介绍。

分类

标签

归档

最新评论

呵呵哒啦在15:06:33评论了
【原创】最佳PHP框架选择(phalcon,yaf,laravel,thinkphp,yii)
jayden在23:39:32评论了
shell中使用while循环ssh的注意事项
ligaofeng在11:55:22评论了
file_get_contents超时问题的解决方法
我也不知道叫个啥好在00:59:46评论了
shell中使用while循环ssh的注意事项
司马成空在16:14:13评论了
【原创】ZendStudio中格式化HTML代码错位问题修正

我看过的书

链接

其他

访问本站种子 本站平均热度:6343 c° 本站链接数:27 个 本站标签数:431 个 本站被评论次数:89 次