nsq学习

摘一幅图：

注意点：

-broadcast-address：
这个选项在nsqd和nsqlookupd中都有出现，这个不仅仅是一个名字，更是外部连入的一个地址，所以不能错了，如：nsqd向nsqlookupd注册的就是这个，而consumer从nslookupd查询到的需要连接的nsqd的地址也是这个

$ curl "http://10.79.80.245:4161/lookup?topic=test" 
{"status_code":200,"status_txt":"OK","data":{"channels":["c1"],"producers":[{"remote_address":"10.79.80.245:47975","hostname":"luna245","broadcast_address":"luna245","tcp_port":4150,"http_port":4151,"version":"0.2.30"},{"remote_address":"10.79.80.245:47976","hostname":"luna245","broadcast_address":"luna245","tcp_port":5150,"http_port":5151,"version":"0.2.30"}]}}

$ curl "http://10.79.80.245:4161/lookup?topic=test"

{"status_code":200,"status_txt":"OK","data":{"channels":["c1"],"producers":[{"remote_address":"10.79.80.245:47975","hostname":"luna245","broadcast_address":"luna245","tcp_port":4150,"http_port":4151,"version":"0.2.30"},{"remote_address":"10.79.80.245:47976","hostname":"luna245","broadcast_address":"luna245","tcp_port":5150,"http_port":5151,"version":"0.2.30"}]}}

注意：上面的broadcast_address 就是由于启动nsqd的时候没有指定该选项导致的，consumer如果不能解析该地址的话就不能获取到消息

-worker-id：
nsqd 会根据worker-id来命名数据文件，即： nsqd.${worker_id}.dat
如： /data1/nsq_data/nsqd.1.dat
如果涉及到 topic和channel的话，不同的nsqd示例就不能（？）使用相同的data目录了，因为topic和channel的meta存储文件名中没有worker-id，如下：
/data1/nsq_data/test:c1.diskqueue.meta.dat
/data1/nsq_data/test.diskqueue.meta.dat由此，也可以看出，每个channel并没有一份重复的存储，应该
nsqd存储

启动nsq相关的一堆命令太麻烦，写一个脚本来start、stop：

#!/usr/local/bin/php
<?php
$type = $argv[1];

$ip = "10.10.10.110";
$worker_id = 1;
if ($type == "start") {
        myexec("nsqlookupd  -http-address=\"0.0.0.0:4161\" -tcp-address=\"0.0.0.0:4160\" -broadcast-address=\"$ip\" ");
        foreach(array(4150, 5150) as $tcp_port) {
                $http_port = $tcp_port + 1;
                $data_dir = "/data1/nsq_data/$tcp_port";
                if(!file_exists($data_dir)) {
                        mkdir($data_dir);
                }
                myexec("nsqd -data-path=$data_dir -http-address=\"0.0.0.0:$http_port\" -tcp-address=\"0.0.0.0:$tcp_port\" -broadcast-address=\"$ip\" -worker-id=$worker_id -lookupd-tcp-address=$ip:4160");
                $worker_id++;
        }
        myexec("nsqadmin -http-address=\"0.0.0.0:4171\" -lookupd-http-address=localhost:4161");
        myexec("nsq_pubsub -lookupd-http-address=localhost:4161 -http-address=:8181");
}
if ($type == "stop") {
        mykill("nsqlookupd");
        mykill("nsqd");
        mykill("nsqadmin");
        mykill("nsq_pubsub");
}
if ($type == "show") {
        mygrep("nsqlookupd");
        mygrep("nsqd");
        mygrep("nsqadmin");
        mygrep("nsq_pubsub");
}
function mygrep($name) {
        echo `ps aux|grep $name|grep -v grep`;
}

function mykill($name) {
        system("ps aux|grep $name|grep -v grep | awk '{print $2}'|xargs -n 1 kill -9");
}

function myexec($cmd) {
        $arr = explode(" ", $cmd);
        $name = $arr[0];
        $pid = pcntl_fork();
        if ($pid == 0) {
                system("nohup $cmd >$name.log 2>&1 &", $ret);
                if ($ret != 0) {
                        die("exec cmd fail: $cmd\n");
                }
        }
}

#!/usr/local/bin/php

<?php

$type = $argv[1];

$ip = "10.10.10.110";

$worker_id = 1;

if ($type == "start") {

myexec("nsqlookupd -http-address=\"0.0.0.0:4161\" -tcp-address=\"0.0.0.0:4160\" -broadcast-address=\"$ip\" ");

foreach(array(4150, 5150) as $tcp_port) {

$http_port = $tcp_port + 1;

$data_dir = "/data1/nsq_data/$tcp_port";

if(!file_exists($data_dir)) {

mkdir($data_dir);

}

myexec("nsqd -data-path=$data_dir -http-address=\"0.0.0.0:$http_port\" -tcp-address=\"0.0.0.0:$tcp_port\" -broadcast-address=\"$ip\" -worker-id=$worker_id -lookupd-tcp-address=$ip:4160");

$worker_id++;

}

myexec("nsqadmin -http-address=\"0.0.0.0:4171\" -lookupd-http-address=localhost:4161");

myexec("nsq_pubsub -lookupd-http-address=localhost:4161 -http-address=:8181");

}

if ($type == "stop") {

mykill("nsqlookupd");

mykill("nsqd");

mykill("nsqadmin");

mykill("nsq_pubsub");

}

if ($type == "show") {

mygrep("nsqlookupd");

mygrep("nsqd");

mygrep("nsqadmin");

mygrep("nsq_pubsub");

}

function mygrep($name) {

echo `ps aux|grep $name|grep -v grep`;

}

function mykill($name) {

system("ps aux|grep $name|grep -v grep | awk '{print $2}'|xargs -n 1 kill -9");

}

function myexec($cmd) {

$arr = explode(" ", $cmd);

$name = $arr[0];

$pid = pcntl_fork();

if ($pid == 0) {

system("nohup $cmd >$name.log 2>&1 &", $ret);

if ($ret != 0) {

die("exec cmd fail: $cmd\n");

}

留下评论

取消回复