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.cfg
nohup /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",
#]
网友评论