php-nsq

php-nsq 是nsq的php客戶端，采用c擴(kuò)展編寫，性能和穩(wěn)定性。

安裝 :

請?zhí)崆鞍惭blibevent

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?publish("test",?"nihao");
}
$nsq->closeNsqdConnection();

//?Deferred?publish?
//function?:?deferredPublish(string?topic,string?message,?int?millisecond);?
//millisecond?default?:?[0?connectNsqd($nsqdAddr);
for($i?=?0;?$i?deferredPublish("test",?"message?daly",?3000);?
}
$deferred->closeNsqdConnection();

?sub例子:

?"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，你也可以利用此函數(shù)做健康檢查

• closeNsqdConnection()?
  關(guān)閉nsq的連接

• publish($topic,$msg)?
  消息發(fā)送

• deferredPublish($topic,$msg,$msec)?
  延遲消息發(fā)送

• subscribe($nsq_lookupd,$config,$callback)?
  消息訂閱

Message 類方法與屬性：

• timestamp?
  消息時間戳

• attempts?
  消息的重試次數(shù)，（從1開始）

• message_id?
  消息id?

• payload?
  消息內(nèi)容

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

• touch($bev,$msg->message_id)?
  如果你消息執(zhí)行太長，可以利用次函數(shù)告知nsq 你還活著，一般用于執(zhí)行頻率比較規(guī)律的場景。

Tips :

1.如果callback內(nèi)需要外部變量，可以采用以下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.消息重試，只要拋異常就可以，切記不要陷入死循環(huán)，超過自己覺得可以的次數(shù) 要return:

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

????????if($msg->attempts?getMessage();
????????????return;
????????}
????}

});

3.如果你想增加 客戶端的心跳時間與消息的超時時間 ：

?第一步?在nsqd啟動時要加入相關(guān)參數(shù)，這個參數(shù)是最大的限制，比如--max-heartbeat-interval=1m30s?心跳時間最大不能超過1分30秒：

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

第二步??因?yàn)榈谝徊绞侵付ㄗ畲髸r間，所以還需要第二步在客戶端指定所需要的值?具體請看?example目錄中的identify開頭的文件例子。

4.如果你想增強(qiáng)消費(fèi)能力，可以加大rdy參數(shù)

5.你可以用supervisor管理，但是因?yàn)槭嵌噙M(jìn)程消費(fèi)，你需要在supervisor job的配置文件?添加:?

????stopasgroup=true
????killasgroup=true

Changes

• 3.0

  • 修復(fù)因libevent 超過4096消息被截斷問題

  • 增加identify指令功能，可以增加客戶端心跳時間 與 消息超時時間

• 2.4.0

  • 修復(fù) pub bug

  • 修復(fù) sub coredump

  • 修覆蓋 touch bug

  • 增加等待，當(dāng)剛初始化的topic沒消息時

• 2.3.1

  • pub支持域名

  • 修復(fù) pub coredump