NSQ 的 PHP 客户端 php-nsq

未知
C/C++ PHP
跨平台
2018-06-21
吴振宇_yunnian

php-nsq

php-nsq 是nsq的php客户端,采用c扩展编写,性能和稳定性。

安装 :

请提前安装libevent

Dependencies: libevent  (apt-get install libevent-dev ,yum install libevent-devel)

1. sudo phpize
2. ./configure 
3. make  
4. make install  

add in your php.ini:

extension = nsq.so;

 pub例子:

$nsqdAddr = array(
    "127.0.0.1:4150",
    "127.0.0.1:4154"
);

$nsq = new Nsq();
$isTrue = $nsq->connectNsqd($nsqdAddr);

for($i = 0; $i < 10000; $i++){
    $nsq->publish("test", "nihao");
}
$nsq->closeNsqdConnection();

// Deferred publish 
//function : deferredPublish(string topic,string message, int millisecond); 
//millisecond default : [0 < millisecond < 3600000]

$deferred = new Nsq();
$isTrue = $deferred->connectNsqd($nsqdAddr);
for($i = 0; $i < 20; $i++){
    $deferred->deferredPublish("test", "message daly", 3000); 
}
$deferred->closeNsqdConnection();

 sub例子:

<?php 

//sub

$nsq_lookupd = new NsqLookupd("127.0.0.1:4161"); //the nsqlookupd http addr
$nsq = new Nsq();
$config = array(
    "topic" => "test",
    "channel" => "struggle",
    "rdy" => 2,                //optional , default 1
    "connect_num" => 1,        //optional , default 1   
    "retry_delay_time" => 5000,  //optional, default 0 , if run callback failed, after 5000 msec, message will be retried
    "auto_finish" => true, //default true
);

$nsq->subscribe($nsq_lookupd, $config, function($msg,$bev){ 

    echo $msg->payload;
    echo $msg->attempts;
    echo $msg->message_id;
    echo $msg->timestamp;


});

Nsq 类方法:

  • connectNsqd($nsqdAddrArr) 
    pub的时候连接nsq,你也可以利用此函数做健康检查

  • closeNsqdConnection() 
    关闭nsq的连接

  • publish($topic,$msg) 
    消息发送

  • deferredPublish($topic,$msg,$msec) 
    延迟消息发送

  • subscribe($nsq_lookupd,$config,$callback) 
    消息订阅

Message 类方法与属性:

  • timestamp 
    消息时间戳

  • attempts 
    消息的重试次数,(从1开始)

  • message_id 
    消息id 

  • payload 
    消息内容

  • finish($bev,$msg->message_id) 
    主动的 ack消息方法

  • touch($bev,$msg->message_id) 
    如果你消息执行太长,可以利用次函数告知nsq 你还活着,一般用于执行频率比较规律的场景。

Tips :

1.如果callback内需要外部变量,可以采用以下use的写法: 

$nsq->subscribe($nsq_lookupd, $config, function($msg,$bev) use ($you_variable){ 

    echo $msg->payload;
    echo $msg->attempts;
    echo $msg->message_id;
    echo $msg->timestamp;


});

 2.消息重试,只要抛异常就可以,切记不要陷入死循环,超过自己觉得可以的次数 要return: 

subscribe($nsq_lookupd, $config, function($msg){ 
    try{
        echo $msg->payload . " " . "attempts:".$msg->attempts."\n";
        //do something
    }catch(Exception $e){

        if($msg->attempts < 3){
            //the message will be retried after you configure retry_delay_time 
            throw new Exception(""); 
        }else{
            echo $e->getMessage();
            return;
        }
    }

});

3.如果你想增加 客户端的心跳时间与消息的超时时间 :

 第一步 在nsqd启动时要加入相关参数,这个参数是最大的限制,比如--max-heartbeat-interval=1m30s 心跳时间最大不能超过1分30秒:

      nsqd --lookupd-tcp-address=127.0.0.1:4160 --max-heartbeat-interval=1m30s --msg-timeout=10m30s

第二步  因为第一步是指定最大时间,所以还需要第二步在客户端指定所需要的值 具体请看 example目录中的identify开头的文件例子。

4.如果你想增强消费能力,可以加大rdy参数

5.你可以用supervisor管理,但是因为是多进程消费,你需要在supervisor job的配置文件 添加: 

    stopasgroup=true
    killasgroup=true

Changes

  • 3.0

    • 修复因libevent 超过4096消息被截断问题

    • 增加identify指令功能,可以增加客户端心跳时间 与 消息超时时间

  • 2.4.0

    • 修复 pub bug

    • 修复 sub coredump

    • 修覆盖 touch bug

    • 增加等待,当刚初始化的topic没消息时

  • 2.3.1

    • pub支持域名

    • 修复 pub coredump

的码云指数为
超过 的项目
加载中

评论(7)

i
inject
感觉还是有很多问题,拿官方demo跑,发布和订阅都有问题。
战场原礼亚
战场原礼亚
为了用这个扩展,开始学习nsq
OSC_hzOkiS
OSC_hzOkiS
After I finished reading your article, I have a question for you.
高久峰
高久峰
支持
宇润
宇润
能支持swoole协程的话就太好了
su-xiaolin
su-xiaolin
为php社区作出了贡献,php的发展更壮大了。
吴振宇_yunnian
吴振宇_yunnian 软件作者
投入了很大的精力与时间,希望对php社区有帮助

php-nsq 3.4.3 线上稳定版发布

php-nsq 3.4.3 线上稳定版发布了。 3.4.3 版本修复了消费功能一些不稳定因素, 此版本经历美篇线上核心功能实践后发布,稳定运行,子进程无论是代码问题导致内存溢出挂掉,还是手动 kill 掉都...

09/19 14:58

php-nsq 3.3 发布,增加进程管理功能

3.3 版本主要增加了进程管理功能,使客户端更加强壮,另外之前版本主进程被kill时 ,子进程变为了孤儿进程被init接管,对这种情况也做了优化。 更新内容: 当一个子进程(与nsqd连接的消费进...

2018/10/21 16:51

php-nsq 3.0 线上稳定版发布,NSQ 的 PHP 客户端

php-nsq 是nsq的php扩展版本的客户端,经过大半年的迭代,功能逐渐完善。此版本是线上稳定版本. 新版本3.0更新内容: 增加了identify 指令功能,可以用于增加客户端的心跳时间 与 消息超时时...

2018/06/21 00:48

没有更多内容

加载失败,请刷新页面

没有更多内容

暂无问答

Linux安装NSQ

1.相关参考资料 https://github.com/yunnian/php-nsq 2.新建安装目录 mkdir /usr/local/nsq 3.在~/.bashrc中添加 export LC_ALL=C 4.执行配置文件 source ~/.bashrc 5.安装依赖 yum install ...

2018/08/16 17:17
129
0
golang使用Nsq

如果一个channel有多个订阅者,NSQ会使用负载均衡的策略,给其它一个订阅者发消息。 https://www.cnblogs.com/swarmbees/p/6635467.html NSQ 是实时的分布式消息处理平台,其设计的目的是用来...

2018/01/24 13:33
26
0
去中心化分布式服务实时消息平台-NSQ

NSQ是一个基于Go语言的开源的分布式实时消息平台 NSQ可用于大规模系统的实时消息服务,它的设计目标是为在分布式环境下提供一个强大的去除中心化的分布式服务架构,可以每天处理数以亿计的实...

2016/10/27 09:30
225
0
go-nsq 使用

## 例子 首先就是一个生产者消费者的模式,自己创建的消费者只要有HandleMessage方法就可以 ``` package main import ( "fmt" "time" "github.com/nsqio/go-nsq" ) // nsq发布消息 func Pro...

2016/11/11 16:59
104
0
nsq简要介绍

nsq服务端 nsq服务包含三个主要部分:lookupd(),nsqd(),admin(管理界面) lookupd 官方理解: lookupd是守护进程负责管理拓扑信息。客户端通过查询 lookupd 来发现指定 topic的生产者,并且...

01/13 00:43
68
0
基于 RocketMQ 的同城双活架构在美菜网的挑战与实践

本文整理自李样兵在北京站 RocketMQ meetup分享美菜网使用 RocketMQ 过程中的一些心得和经验,偏重于实践。 嘉宾李样兵,现就职于美菜网基础服务平台组,负责 MQ ,配置中心和任务调度等基础...

09/16 14:41
11
0
基于 RocketMQ 的同城双活架构在美菜网的挑战与实践

本文整理自李样兵在北京站 RocketMQ meetup分享美菜网使用 RocketMQ 过程中的一些心得和经验,偏重于实践。 嘉宾李样兵,现就职于美菜网基础服务平台组,负责 MQ ,配置中心和任务调度等基础...

09/16 16:03
21
0

没有更多内容

加载失败,请刷新页面

返回顶部
顶部