码迷,mamicode.com
首页 > 其他好文 > 详细

NSQ部署

时间:2015-11-05 15:18:20      阅读:486      评论:0      收藏:0      [点我收藏+]

标签:

一、      简介

NSQ主要有三个主要程序和一个Web服务程序:

  1. nsqd:是守护进程,接收,缓存,并投递消息给客户端
  2. nsqlookupd:是一个守护进程,为消费者提供运行时发现服务,来查找指定话题(topic)的生产者 nsqd
  3. nsq_to_http:消费指定的话题(topic)/通道(channel)和执行 HTTP requests (GET/POST) 到指定的端点。
  4. nsqadmin:是 Web 服务,用来实时的管理你的 NSQ 集群。它通过和 nsqlookupd 实例交流,来确定生产者

除了nsqadmin是独立提供WEB信息以外,三个主要程序组成类似下面的网络拓扑,以提供消息服务:

技术分享

二、      NSQ获取

下载最新的程序。并将NSQ所在目录加入${PATH}中。

三、      部署

以下假设我们要在两台机器上部署NSQ:

192.168.30.54(简称54)

192.168.30.55(简称55)。

其中nsqlookupd和nsqadmin部署到54,nsq两台机器上都部署,以冗余nsqd来实现容灾。

 

以下假设nsq工具集的路径已包含在${PATH}环境变量中。

 

1)     54的部署:

  技术分享

当程序未指定—config参数时,程序会尝试从配置文件中读取配置信息。配置文件是以cfg为扩展名,且与程序名同名的文本文件。

nsqlookupd.cfg

一般情况下nsqlookupd不需要配置文件,我们使用它的缺省配置即可,如果需要它在特殊端口上监听或修改某些配置,可以修改配置文件

技术分享

nsqadmin.cfg

技术分享

nsqd.cfg

技术分享

2)     55的部署:

55上只需启动nsqd即可

技术分享

该上面的配置文件,除了id被设置成2外,与54上全部相同。

四、      监控

4.1 可视化监控

通过http浏览器,我们可以看到NSQ的运行情况,见下图。该图是由nsqadmin提供的。

 技术分享

4.2 自动监控

自动监控脚本,见附录。

自动监控目前主要监控两个指标:

1)  运行中的nsqd的数量。主要监控其是否正常运行。

2)  nsqd上各channel最大堆积的消息数量。主要监控下游处理情况。

 

自动监控输出示例:

技术分享

说明:

第一行主要监控nsqd数量,特征字段为nsqd_count,当前系统中有2个nsqd。

第二、三行监控nsqd运行情况,特征字段为nsqd。

warning表明是否有报警。off – 不报警,监控项运行正常;on – 报警,该项运行异常。

 

例:

第一行,当前运行的nsqd有两个(nsqd_count),系统中应有nsqd数量为2(should_be),不报警(warning[off])

第二行,nsqd地址为192.168.30.54:4151,nsqd运行良好(health[OK], 不报警(warning[off])。

五、      异常处理

当NSQ发生异常时,可以通过http://192.168.30.54:4171观察到,或通过自动监控机制接收到报警。这时可以重启有问题的进程,以使系统恢复。

 

例如,如果nsqd运行异常,则可以运行如下命令重启nsqd:

技术分享

目前已知异常的处理:

1)  系统时间向后调整后, nsqd占用100%的CPU

描述:将系统时间由2015-08-25 11:02:00 向后调整2015-08-24 11:02:00后,nsqd占用100%的CPU

分析:由于是时间向后调整,因此造成nsqd定时器,向前追赶新时间,造成定时器被密集调用。

解决方法:不管是向前(跨度较大时),还是向后调整时间,都建议重启nsqd。

六、下游消息处理(nsq_to_http)(SUB)

6.1 GET推送

典型的命令行参数:

nsq_to_http --topic=demo --lookupd-http-address=127.0.0.1:4161 --get=http://localhost/nsqphp/sub?msg=%s

参数—get表明消息将以GET请求向WebServer(SUB)推送。%s以可打印的形式被实际消息替代。

nsq_to_http需要三元组(topic, lookupd的http地址,get推送地址)

 

6.2 POST推送(非表单数据)

典型的命令行参数

nsq_to_http --topic=demo --lookupd-http-address=127.0.0.1:4161 --post=http://localhost/nsqphp/sub

参数—post表明消息将以POST请求向WebServer(SUB)推送。消息在http的body部分。

此时推送的数据将以二进制推送,也即HTTP请求的Content-Type:application/octet-stream。如果WebServer是以php实现的,则需要以file_get_contents(“php://input”)的形式读取数据。

6.3 POST推送(表单数据)

典型的命令行参数

nsq_to_http --topic=demo --lookupd-http-address=127.0.0.1:4161 --post=http://localhost/nsqphp/sub --content-type="application/x-www-form-urlencoded"

参数—post表明消息将以POST请求向WebServer(SUB)推送。消息在http的body部分。

注意—content-type参数,该参数只有在—post的情况下才有效。该参数表明post的消息是以表单的形式组织的,也即可以通过$_POST[]或$_REQUEST[]的方式取到消息

七、其它

1)   nsqphp向nsqd发送信息时各阶段超时时间

名称

超时时间

说明

connectionTimeout

3秒

连接nsqd的超时时间

readWriteTimeout

3秒

收发数据的超时时间

readWaitTimeout

5秒

接收回应的超时时间

 

附录:

自动监控脚本: 

<?php
/**
 * Created by PhpStorm.
 * User: Lenovo
 * Date: 2015/8/22
 * Time: 18:05
 */

error_reporting(E_ALL || ~E_NOTICE); //ÏÔʾ³ýÈ¥ E_NOTICE Ö®ÍâµÄËùÓдíÎóÐÅÏ¢

// http://192.168.30.54:4161/nodes

function queryLookupdForNsqd($lookupAddrs) {
    foreach ($lookupAddrs as $addr) {
        $text = file_get_contents("http://$addr/nodes");
        // $nodes¸ñʽ
        // {"status_code":200,"status_txt":"OK","data":{"producers":[{"remote_address":"192.168.30.54:38631","hostname":"localhost.localdomain","broadcast_address":"192.168.30.54","tcp_port":4150,"http_port":4151,"version":"0.3.6-alpha","tombstones":[false,false],"topics":["demo","hancj"]},{"remote_address":"192.168.30.55:35621","hostname":"localhost.localdomain","broadcast_address":"192.168.30.55","tcp_port":4150,"http_port":4151,"version":"0.3.6-alpha","tombstones":[false,false],"topics":["demo","hancj"]}]}}
        $nodes = json_decode($text, true);

        $arrNode = array();
        foreach ($nodes[‘data‘][‘producers‘] as $nsqd) {
            $addr = "{$nsqd[‘broadcast_address‘]}:{$nsqd[‘http_port‘]}";
            //echo $addr, PHP_EOL;
            $arrNode[] = $addr;
        }

        return $arrNode;
    }
}

function queryNsqd($nsqdAddr) {
    $httpAddr = "http://$nsqdAddr/stats?format=json";
    // infoµÄ¸ñʽ
    // {"status_code":200,"status_txt":"OK","data":{"version":"0.3.6-alpha","health":"OK","start_time":1440233123,"topics":[{"topic_name":"demo","channels":[{"channel_name":"nsq_to_http","depth":0,"backend_depth":0,"in_flight_count":0,"deferred_count":0,"message_count":7010004,"requeue_count":1046,"timeout_count":649,"clients":[],"paused":false,"e2e_processing_latency":{"count":0,"percentiles":null}}],"depth":0,"backend_depth":0,"message_count":7010004,"paused":false,"e2e_processing_latency":{"count":0,"percentiles":null}},{"topic_name":"hancj","channels":[{"channel_name":"nsq_to_http","depth":1,"backend_depth":0,"in_flight_count":0,"deferred_count":0,"message_count":0,"requeue_count":0,"timeout_count":0,"clients":[],"paused":false,"e2e_processing_latency":{"count":0,"percentiles":null}}],"depth":0,"backend_depth":0,"message_count":0,"paused":false,"e2e_processing_latency":{"count":0,"percentiles":null}}]}}
    $info = json_decode(file_get_contents($httpAddr), true);

    $nsqd = array(‘health‘ => $info[‘data‘][‘health‘], ‘channels‘ => 0, ‘topics‘ => 0);

    foreach ($info[‘data‘][‘topics‘] as $topic) {
        $nsqd[‘topics‘] += 1;
        foreach ($topic[‘channels‘] as $channel) {
            $nsqd[‘channels‘] += 1;
            if ($channel[‘channel_name‘] == ‘nsq_to_http‘) {
                if (!$nsqd[‘max_channel‘] || $nsqd[‘max_channel‘][‘depth‘] < $channel[‘depth‘]) {
                    $nsqd[‘max_channel‘] = $channel;
                }
            }
        }
    }

    return $nsqd;
}

function main($argv) {
    if (count($argv) < 3) {
        echo "Usage: {$argv[0]} nsqd_count nsqlookupd_address...\n";
        exit;
    }

    $nsqdCount = $argv[1];
    $nsqlookupd = array_slice($argv, 2);

    $now = date(‘Y-m-d H:i:s‘);
    $arrNSQD = queryLookupdForNsqd($nsqlookupd);//array("192.168.30.54:4161",));
    $warn = (count($arrNSQD) == $nsqdCount ? "off" : "on");
    echo "$now nsqd_count[" . count($arrNSQD) . "] should_be[{$nsqdCount}] warning[$warn]", PHP_EOL;
    foreach ($arrNSQD as $nsqdAddr) {
        $info = queryNsqd($nsqdAddr);

        $warn = ((!$info || $info[‘health‘] != ‘OK‘ || $info[‘max_channel‘][‘depth‘] > 100000) ? "on" : "off");
        echo "$now nsqd[$nsqdAddr] health[{$info[‘health‘]}] warning[$warn] topics[{$info[‘topics‘]}]";
        echo " channels[{$info[‘channels‘]}] depth[{$info[‘max_channel‘][‘depth‘]}]";
        echo PHP_EOL;
    }
}

main($argv);

  

NSQ官网地址:http://nsq.io/

NSQ部署

标签:

原文地址:http://www.cnblogs.com/lixiuran/p/4939437.html

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!