NSQ简介

NSQ 是一个实时分布式消息平台，旨在大规模运行，每天处理数十亿条消息。

它提倡没有单点故障的分布式和分散式拓扑结构，实现容错和高可用性，同时保证可靠的消息传递。请看特点和保证。

在操作上，NSQ很容易配置和部署（所有参数都在命令行上指定，编译的二进制文件没有运行时的依赖性）。为了获得最大的灵活性，它与数据格式无关（消息可以是JSON、MsgPack、协议缓冲区或其他任何形式）。官方的Go和Python库是开箱即用的（还有许多其他的客户端库），如果你有兴趣建立自己的库，有一个协议规范。

特点:

• 追求简单部署

• 追求高可用、避免单点故障、无中心设计

• 确保消息送达

• 生产者消费者自动发现、消费者连接所有生产者、向消费者推的模式

• 提供 HTTP 接口

• 提供几乎所有编程语言的客户端开发包

NSQ是由知名短链接服务商bitly用Go语言开发的实时消息处理系统，具有高性能、高可靠、无视单点故障等优点，是一个非常不错的新兴的消息队列解决方案。

源代码地址:https://github.com/nsqio/nsq

下载和安装nsq

官方安装地址:https://nsq.io/overview/quick_start.html

官方下载地址:https://nsq.io/deployment/installing.html

#创建目录,并进入
mkdir -p /data/software
cd /data/software#下载nsq包
wget https://s3.amazonaws.com/bitly-downloads/nsq/nsq-0.3.8.linux-amd64.go1.6.2.tar.gz#解压
tar -zxvf nsq-0.3.8.linux-amd64.go1.6.2.tar.gz#复制到指定目录
cp -r /data/software/nsq-0.3.8.linux-amd64.go1.6.2 /data/beyond/nsq

启动时可配置文件的参考地址:
https://liushuchun.gitbooks.io/mixapi/content/nsq_config.html

启动相关服务

nohup /data/beyond/nsq/bin/nsqd &
#（守护进程；接收，缓存和投递消息给客户端）  如：nsqd -config=/home/nsq/bin/nsqd.cfgnohup /data/beyond/nsq/bin/nsqlookupd &
#（守护进程；为消费者提供运行时发现服务，来查找指定话题（topic）的生产者 nsqd） nohup /data/beyond/nsq/bin/nsqadmin &
#（提供 Web 页面用来实时的管理你的 NSQ 集群。它通过和 nsqlookupd 实例交流，来确定生产者）

指定端口启动
nohup /data/beyond/nsq/bin/nsqadmin --lookupd-http-address localhost:4160 &

指定配置文件启动
nohup /data/beyond/nsq/bin/nsqlookupd -config=/data/beyond/nsq/conf/nsqlookupd.cfg &

nohup /data/beyond/nsq/bin/nsqadmin -config=/data/beyond/nsq/conf/nsqadmin.cfg &

nohup /data/beyond/nsq/bin/nsqd -config=/data/beyond/nsq/conf/nsqd.cfg &

集群管理
记得谁说过，go调用nsq时，只能在代码里配置多个地址，去轮询调用

测试环境的配置,如:192.168.1.6

nsqd.cfg文件

cat nsqd.cfg
#配置如下## enable verbose logging
verbose = false## unique identifier (int) for this worker (will default to a hash of hostname)
id = 66## <addr>:<port> to listen on for TCP clients
tcp_address = "0.0.0.0:4150"## <addr>:<port> to listen on for HTTP clients
http_address = "0.0.0.0:4151"## <addr>:<port> to listen on for HTTPS clients
# https_address = "0.0.0.0:4152"## address that will be registered with lookupd (defaults to the OS hostname)
broadcast_address = "192.168.1.6"## cluster of nsqlookupd TCP addresses
nsqlookupd_tcp_addresses = ["192.168.1.6:4160"
]## duration to wait before HTTP client connection timeout
http_client_connect_timeout = "2s"## duration to wait before HTTP client request timeout
http_client_request_timeout = "5s"## path to store disk-backed messages
# data_path = "/var/lib/nsq"## number of messages to keep in memory (per topic/channel)
mem_queue_size = 10000## number of bytes per diskqueue file before rolling
max_bytes_per_file = 104857600## number of messages per diskqueue fsync
sync_every = 2500## duration of time per diskqueue fsync (time.Duration)
sync_timeout = "2s"## duration to wait before auto-requeing a message
msg_timeout = "60s"## maximum duration before a message will timeout
max_msg_timeout = "15m"## maximum size of a single message in bytes
max_msg_size = 1024768## maximum requeuing timeout for a message
max_req_timeout = "1h"## maximum size of a single command body
max_body_size = 5123840## maximum client configurable duration of time between client heartbeats
max_heartbeat_interval = "60s"## maximum RDY count for a client
max_rdy_count = 2500## maximum client configurable size (in bytes) for a client output buffer
max_output_buffer_size = 65536## maximum client configurable duration of time between flushing to a client (time.Duration)
max_output_buffer_timeout = "1s"## UDP <addr>:<port> of a statsd daemon for pushing stats
# statsd_address = "127.0.0.1:8125"## prefix used for keys sent to statsd (%s for host replacement)
statsd_prefix = "nsq.%s"## duration between pushing to statsd (time.Duration)
statsd_interval = "60s"## toggle sending memory and GC stats to statsd
statsd_mem_stats = true## message processing time percentiles to keep track of (float)
e2e_processing_latency_percentiles = [100.0,99.0,95.0
]## calculate end to end latency quantiles for this duration of time (time.Duration)
e2e_processing_latency_window_time = "10m"## path to certificate file
tls_cert = ""## path to private key file
tls_key = ""## set policy on client certificate (require - client must provide certificate,
##  require-verify - client must provide verifiable signed certificate)
# tls_client_auth_policy = "require-verify"## set custom root Certificate Authority
# tls_root_ca_file = ""## require client TLS upgrades
tls_required = false## minimum TLS version ("ssl3.0", "tls1.0," "tls1.1", "tls1.2")
tls_min_version = ""## enable deflate feature negotiation (client compression)
deflate = true## max deflate compression level a client can negotiate (> values == > nsqd CPU usage)
max_deflate_level = 6## enable snappy feature negotiation (client compression)
snappy = true

nsqlookupd.cfg文件

cat nsqlookupd.cfg
#配置如下## enable verbose logging
verbose = false## <addr>:<port> to listen on for TCP clients
tcp_address = "0.0.0.0:4160"## <addr>:<port> to listen on for HTTP clients
http_address = "0.0.0.0:4161"## address that will be registered with lookupd (defaults to the OS hostname)
broadcast_address = "192.168.1.6"## duration of time a producer will remain in the active list since its last ping
inactive_producer_timeout = "300s"## duration of time a producer will remain tombstoned if registration remains
tombstone_lifetime = "45s"

nsqadmin.cfg文件

cat nsqadmin.cfg
#配置如下## <addr>:<port> to listen on for HTTP clients
http_address = "0.0.0.0:4171"## graphite HTTP address
graphite_url = ""## proxy HTTP requests to graphite
proxy_graphite = false## prefix used for keys sent to statsd (%s for host replacement, must match nsqd)
statsd_prefix = "nsq.%s"## format of statsd counter stats
statsd_counter_format = "stats.counters.%s.count"## format of statsd gauge stats
statsd_gauge_format = "stats.gauges.%s"## time interval nsqd is configured to push to statsd (must match nsqd)
statsd_interval = "60s"## HTTP endpoint (fully qualified) to which POST notifications of admin actions will be sent
notification_http_endpoint = ""## nsqlookupd HTTP addresses
nsqlookupd_http_addresses = ["192.168.1.6:4161"
]## nsqd HTTP addresses (optional)
#nsqd_http_addresses = [
#    "192.168.1.6:4151",
#    "192.168.1.7:4151",
#]

测试环境的配置集群,第二台,如:192.168.1.7

注意: nsqlookupd_tcp_addresses 地址配置的是 192.168.1.6

nsqd.cfg文件

cat nsqd.cfg
#配置如下## enable verbose logging
verbose = false## unique identifier (int) for this worker (will default to a hash of hostname)
id = 77## <addr>:<port> to listen on for TCP clients
tcp_address = "0.0.0.0:4150"## <addr>:<port> to listen on for HTTP clients
http_address = "0.0.0.0:4151"## <addr>:<port> to listen on for HTTPS clients
# https_address = "0.0.0.0:4152"## address that will be registered with lookupd (defaults to the OS hostname)
broadcast_address = "192.168.1.7"## cluster of nsqlookupd TCP addresses
nsqlookupd_tcp_addresses = ["192.168.1.6:4160"
]## duration to wait before HTTP client connection timeout
http_client_connect_timeout = "2s"## duration to wait before HTTP client request timeout
http_client_request_timeout = "5s"## path to store disk-backed messages
# data_path = "/var/lib/nsq"## number of messages to keep in memory (per topic/channel)
mem_queue_size = 10000## number of bytes per diskqueue file before rolling
max_bytes_per_file = 104857600## number of messages per diskqueue fsync
sync_every = 2500## duration of time per diskqueue fsync (time.Duration)
sync_timeout = "2s"## duration to wait before auto-requeing a message
msg_timeout = "60s"## maximum duration before a message will timeout
max_msg_timeout = "15m"## maximum size of a single message in bytes
max_msg_size = 1024768## maximum requeuing timeout for a message
max_req_timeout = "1h"## maximum size of a single command body
max_body_size = 5123840## maximum client configurable duration of time between client heartbeats
max_heartbeat_interval = "60s"## maximum RDY count for a client
max_rdy_count = 2500## maximum client configurable size (in bytes) for a client output buffer
max_output_buffer_size = 65536## maximum client configurable duration of time between flushing to a client (time.Duration)
max_output_buffer_timeout = "1s"## UDP <addr>:<port> of a statsd daemon for pushing stats
# statsd_address = "127.0.0.1:8125"## prefix used for keys sent to statsd (%s for host replacement)
statsd_prefix = "nsq.%s"## duration between pushing to statsd (time.Duration)
statsd_interval = "60s"## toggle sending memory and GC stats to statsd
statsd_mem_stats = true## message processing time percentiles to keep track of (float)
e2e_processing_latency_percentiles = [100.0,99.0,95.0
]## calculate end to end latency quantiles for this duration of time (time.Duration)
e2e_processing_latency_window_time = "10m"## path to certificate file
tls_cert = ""## path to private key file
tls_key = ""## set policy on client certificate (require - client must provide certificate,
##  require-verify - client must provide verifiable signed certificate)
# tls_client_auth_policy = "require-verify"## set custom root Certificate Authority
# tls_root_ca_file = ""## require client TLS upgrades
tls_required = false## minimum TLS version ("ssl3.0", "tls1.0," "tls1.1", "tls1.2")
tls_min_version = ""## enable deflate feature negotiation (client compression)
deflate = true## max deflate compression level a client can negotiate (> values == > nsqd CPU usage)
max_deflate_level = 6## enable snappy feature negotiation (client compression)
snappy = true

其他环境的配置,如: 172.28.15.1

nsqd.cfg 文件

cat nsqd.cfg 
#配置如下## enable verbose logging
verbose = false## unique identifier (int) for this worker (will default to a hash of hostname)
id = 66 ## <addr>:<port> to listen on for TCP clients
tcp_address = "0.0.0.0:4150"## <addr>:<port> to listen on for HTTP clients
http_address = "0.0.0.0:4151"## <addr>:<port> to listen on for HTTPS clients
# https_address = "0.0.0.0:4152"## address that will be registered with lookupd (defaults to the OS hostname)
broadcast_address = "172.28.15.1"## cluster of nsqlookupd TCP addresses
nsqlookupd_tcp_addresses = ["172.28.15.1:4160"
]## duration to wait before HTTP client connection timeout
http_client_connect_timeout = "2s"## duration to wait before HTTP client request timeout
http_client_request_timeout = "5s"## path to store disk-backed messages
# data_path = "/var/lib/nsq"## number of messages to keep in memory (per topic/channel)
mem_queue_size = 10000## number of bytes per diskqueue file before rolling
max_bytes_per_file = 104857600## number of messages per diskqueue fsync
sync_every = 2500## duration of time per diskqueue fsync (time.Duration)
sync_timeout = "2s"## duration to wait before auto-requeing a message
msg_timeout = "60s"## maximum duration before a message will timeout
max_msg_timeout = "15m"## maximum size of a single message in bytes
max_msg_size = 1024768## maximum requeuing timeout for a message
max_req_timeout = "1h"## maximum size of a single command body
max_body_size = 5123840## maximum client configurable duration of time between client heartbeats
max_heartbeat_interval = "60s"## maximum RDY count for a client
max_rdy_count = 2500## maximum client configurable size (in bytes) for a client output buffer
max_output_buffer_size = 65536## maximum client configurable duration of time between flushing to a client (time.Duration)
max_output_buffer_timeout = "1s"## UDP <addr>:<port> of a statsd daemon for pushing stats
# statsd_address = "127.0.0.1:8125"## prefix used for keys sent to statsd (%s for host replacement)
statsd_prefix = "nsq.%s"## duration between pushing to statsd (time.Duration)
statsd_interval = "60s"## toggle sending memory and GC stats to statsd
statsd_mem_stats = true## message processing time percentiles to keep track of (float)
e2e_processing_latency_percentiles = [100.0,99.0,95.0
]## calculate end to end latency quantiles for this duration of time (time.Duration)
e2e_processing_latency_window_time = "10m"## path to certificate file
tls_cert = ""## path to private key file
tls_key = ""## set policy on client certificate (require - client must provide certificate,
##  require-verify - client must provide verifiable signed certificate)
# tls_client_auth_policy = "require-verify"## set custom root Certificate Authority
# tls_root_ca_file = ""## require client TLS upgrades
tls_required = false## minimum TLS version ("ssl3.0", "tls1.0," "tls1.1", "tls1.2")
tls_min_version = ""## enable deflate feature negotiation (client compression)
deflate = true## max deflate compression level a client can negotiate (> values == > nsqd CPU usage)
max_deflate_level = 6## enable snappy feature negotiation (client compression)
snappy = true

nsqlookupd.cfg文件

cat nsqlookupd.cfg
#配置如下## enable verbose logging
verbose = false## <addr>:<port> to listen on for TCP clients
tcp_address = "0.0.0.0:4160"## <addr>:<port> to listen on for HTTP clients
http_address = "0.0.0.0:4161"## address that will be registered with lookupd (defaults to the OS hostname)
broadcast_address = "172.28.15.1"## duration of time a producer will remain in the active list since its last ping
inactive_producer_timeout = "300s"## duration of time a producer will remain tombstoned if registration remains
tombstone_lifetime = "45s"

nsqadmin.cfg文件

cat nsqadmin.cfg
#配置如下## <addr>:<port> to listen on for HTTP clients
http_address = "0.0.0.0:4171"## graphite HTTP address
graphite_url = ""## proxy HTTP requests to graphite
proxy_graphite = false## prefix used for keys sent to statsd (%s for host replacement, must match nsqd)
statsd_prefix = "nsq.%s"## format of statsd counter stats
statsd_counter_format = "stats.counters.%s.count"## format of statsd gauge stats
statsd_gauge_format = "stats.gauges.%s"## time interval nsqd is configured to push to statsd (must match nsqd)
statsd_interval = "60s"## HTTP endpoint (fully qualified) to which POST notifications of admin actions will be sent
notification_http_endpoint = ""## nsqlookupd HTTP addresses
nsqlookupd_http_addresses = ["172.28.15.1:4161"
]## nsqd HTTP addresses (optional)
#nsqd_http_addresses = [
#    "192.168.1.6:4151",
#    "192.168.1.7:4151",
#]