42

PHP-NSQ —— 基于 C 的 NSQ 的 PHP 客户端

 2年前 阅读数 65
以下为 快照 页面,建议前往来源网站查看,会有更好的阅读体验。
原文链接: https://www.oschina.net/p/php-nsq?amp%3Butm_medium=referral

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


猜你喜欢

  • 24
    • studygolang.com 2年前
    • 快照

    NSQ源码-Nsq客户端

    看完lookupd和nsqd之后我们再来看下nsq client端的代码。 我是想把nsq系统完完整整的看一遍,从而对他形成一个更整体的 认识。对message queue来说他的client端就是生产者和消费者,生产者负责想nsq中投递消息,消费者负责...

  • 36
    • 微信 mp.weixin.qq.com 2年前
    • 快照

    NSQ最佳实践

  • 78
    • 微信 mp.weixin.qq.com 2年前
    • 快照

    NSQ 最佳实践

    目前,全新的异步任务服务每天高效稳定的为唱吧提供数亿次的调用。服务器团队用全新的方式重新定义了异步任务实现方式,以为云计算而生的NSQ、成熟的PHP执行者PHP-FPM、自主开发的中间件NSQProxy以及admin管理后台共同组成了异步任务的队...

  • 62
    • bridgeforyou.cn 2年前
    • 快照

    MQ(6) —— Nsq vs Kafka

    Nsq vs Kafka 正如之前说的,Nsq是一款极简的消息中间件,通过学习Nsq,我们可以通过对比的方式,学习其他的Mq。 这一节,就让我们在对比中,学习另一种Mq,Kafka,在对比中,加深对Mq的理解。 首先,先放上...

  • 29
    • studygolang.com 2年前
    • 快照

    NSQ源码-NSQD

    看完了nsqlookupd我们继续往下看, nsqd才是他的核心. 里面大量的使用到了go channel, 相信看完之后对你学习go有很大的帮助.相较于lookupd部分无论在代码逻辑和实现上都要复杂很多. 不过基本的代码结构基本上都是一样的, 进...

  • 50
    • studygolang.com 2年前
    • 快照

    NSQ源码-nsqlookupd

    为什么选择nsq 之前一直在用erlang做电信产品的开发,对erlang的一些生态也比较了解,和erlang相关的产品在互联网公司使用最多的应该就是rabbitmq了,也许很多人听说过erlang就是因为他们公司在使用rabbitmq。在之前也看过一点...

  • 46
    • www.tuicool.com 1年前
    • 快照

    NSQ v0.1.5 源码分析

    NSQ 是实时的分布式消息处理平台,其设计的目的是用来大规模地处理每天数以十亿计级别的消息。NSQ 具有分布式和去中心化拓扑结构,该结构具有无单点故障、故障容错、高可用性以及能够保证消息的可靠传递的特征,是一个成熟的、已在大规模...

  • 65
    • www.tuicool.com 1年前
    • 快照

    nsq源码review:go-nsq producer

    nsq是一个实时分布式的消息队列平台。 核心部分是一个叫nsqd的模块,它负责接收和转发消息。同时在go-nsq的包中,提供了consumer和producer的核心接口。在读nsq源码的时候,很好奇它的数据是怎么从producer给到了consumer的,于...

  • 64
    • www.tuicool.com 1年前
    • 快照

    使用NSQ(附Golang代码)

    上篇文章已经了解了消息中间件相关的知识,这篇文章学习一下Golang语言编写的知名消息中间件 NSQ 。 nsq最初是由bitly公司开源出来的一款简单易用的消息中间件,它可用于大规模系...

  • 46
    • www.tuicool.com 1年前
    • 快照

    golang nsq

    1.安装 https://nsq.io/deployment/installing.html 2.快速入门 在一个终端中,运行

关于极客头条


聚合每日国内外有价值,有趣的链接。

AD