Go之NSQ简介,原理和使用

/*
		1. NSQ提倡分布式和分散的拓扑,没有单点故障,支持容错和高可用性, 并提供可靠的消息交付保证.
		2. NSQ支持横向扩展,没有任何集中式代理
		3. NSQ易于配置和部署,并且内置了管理界面
*/

/*
		1. 支持无 SPOF 的分布式拓扑
		2. 水平扩展(没有中间件，无缝地添加更多的节点到集群)
		3. 低延迟消息传递 (性能)
    4. 结合负载均衡和多播消息路由风格
    5. 擅长面向流媒体(高通量)和工作(低吞吐量)工作负载
    6. 主要是内存中(除了高水位线消息透明地保存在磁盘上)
    7. 运行时发现消费者找到生产者服务(nsqlookupd)
    8. 传输层安全性 (TLS)
    9. 数据格式不可知
    10. 一些依赖项(容易部署)和健全的，有界，默认配置
    11. 任何语言都有简单 TCP 协议支持客户端库
    12. HTTP 接口统计、管理行为和生产者(不需要客户端库发布)
    13. 为实时检测集成了 statsd
    14. 健壮的集群管理界面 (nsqadmin)
*/

/*
		1. 消息默认不持久化, 可以配置成持久化模式, nsq采用的方式是内存+硬盘的模式,当内存到一定程度就会持久化到硬盘.
				如果将 --mem-queue-size设置为0, 所有的消息将会存储到磁盘.
				服务器重启时也会将在内存中的消息持久化
		2. 每条消息至少传递一次
		3. 消息不保证有序.
*/

/*
		特性:
		1. 对订阅了同一个topic，同一个channel的消费者使用负载均衡策略（不是轮询）
		
		2. 只要channel存在，即使没有该channel的消费者，也会将生产者的message缓存到队列中（注意消息的过期处理）
		
		3. 保证队列中的message至少会被消费一次，即使nsqd退出，也会将队列中的消息暂存磁盘上(结束进程等意外情况除外)

		4. 限定内存占用，能够配置nsqd中每个channel队列在内存中缓存的message数量，一旦超出，message将被缓存到磁盘中

		5. topic，channel一旦建立，将会一直存在，要及时在管理台或者用代码清除无效的topic和channel，避免资源的浪费
*/

/*
		特性:
		
		1. 唯一性，在一个Nsq服务中只有一个nsqlookupd服务。当然也可以在集群中部署多个nsqlookupd，但它们之间是没有关联的.
		2. 去中心化，即使nsqlookupd崩溃，也会不影响正在运行的nsqd服务
		3. 充当nsqd和naqadmin信息交互的中间件
		4. 提供一个http查询服务，给客户端定时更新nsqd的地址目录.
*/

/*
		特性:
		
		1. 提供一个对topic和channel统一管理的操作界面以及各种实时监控数据的展示，界面设计的很简洁，操作也很简单
		2. 展示所有message的数量
		3. 能够在后台创建topic和channel
		4. nsqadmin的所有功能都必须依赖于nsqlookupd，nsqadmin只是向nsqlookupd传递用户操作并展示来自nsqlookupd的数据
		
*/

wget https://s3.amazonaws.com/bitly-downloads/nsq/nsq-1.2.0.linux-amd64.go1.12.9.tar.gz


tar xf nsq-1.2.0.linux-amd64.go1.12.9.tar.gz -C /usr/local/

[root@nsq-47 ~]# tail -1 /etc/hosts
192.168.43.47 nsq-47

# 开三个终端,分别按顺序启动
./nsqlookupd 

./nsqd --lookupd-tcp-address=192.168.43.47:4160
  
./nsqadmin --lookupd-http-address=192.168.43.47:4161
  
# 访问
http://192.168.43.47:4171

/*
		go get -u github.com/nsqio/go-nsq
*/

// nsq_producer/main.go
package main

import (
	"bufio"
	"fmt"
	"os"
	"strings"

	"github.com/nsqio/go-nsq"
)

// NSQ Producer Demo

var producer *nsq.Producer

// 初始化生产者
func initProducer(str string) (err error) {
	config := nsq.NewConfig()
	producer, err = nsq.NewProducer(str, config)
	if err != nil {
		fmt.Printf("create producer failed, err:%v\n", err)
		return err
	}
	return nil
}

func main() {
	nsqAddress := "192.168.43.47:4150"
	err := initProducer(nsqAddress)
	if err != nil {
		fmt.Printf("init producer failed, err:%v\n", err)
		return
	}

	reader := bufio.NewReader(os.Stdin) // 从标准输入读取
	for {
		data, err := reader.ReadString('\n')
		if err != nil {
			fmt.Printf("read string from stdin failed, err:%v\n", err)
			continue
		}
		data = strings.TrimSpace(data)
		if strings.ToUpper(data) == "Q" { // 输入Q退出
			break
		}
		// 向 'topic_demo' publish 数据
		err = producer.Publish("topic_demo", []byte(data))
		if err != nil {
			fmt.Printf("publish msg to nsq failed, err:%v\n", err)
			continue
		}
	}
}

// nsq_consumer/main.go
package main

import (
	"fmt"
	"os"
	"os/signal"
	"syscall"
	"time"

	"github.com/nsqio/go-nsq"
)

// NSQ Consumer Demo

// MyHandler 是一个消费者类型
type MyHandler struct {
	Title string
}

// HandleMessage 是需要实现的处理消息的方法
func (m *MyHandler) HandleMessage(msg *nsq.Message) (err error) {
	fmt.Printf("%s recv from %v, msg:%v\n", m.Title, msg.NSQDAddress, string(msg.Body))
	return
}

// 初始化消费者
func initConsumer(topic string, channel string, address string) (err error) {
	config := nsq.NewConfig()
	config.LookupdPollInterval = 15 * time.Second
	c, err := nsq.NewConsumer(topic, channel, config)
	if err != nil {
		fmt.Printf("create consumer failed, err:%v\n", err)
		return
	}
	consumer := &MyHandler{
		Title: "沙河1号",
	}
	c.AddHandler(consumer)

	// if err := c.ConnectToNSQD(address); err != nil { // 直接连NSQD
	if err := c.ConnectToNSQLookupd(address); err != nil { // 通过lookupd查询
		return err
	}
	return nil

}

func main() {
	err := initConsumer("topic_demo", "first", "192.168.43.47:4161")
	if err != nil {
		fmt.Printf("init consumer failed, err:%v\n", err)
		return
	}
	c := make(chan os.Signal)        // 定义一个信号的通道
	signal.Notify(c, syscall.SIGINT) // 转发键盘中断信号到c
	<-c                              // 阻塞
}