【goalng】消息订阅收发消息中间件 NSQ 部署使用

nsq是用golang写的一款非常成熟的支持分布式的高可用的消息传递中间件。它非常的好集成具体多好集成，在使用docker 的情况下三条命令搞定一个完整的实例，docker之前从来没有用过，接触的也不多，但是真用起来才觉得以前关于配置环境这种头疼的事情都可以省略了。
首先在NSQ中有三个组件是你必须知道的 nsqd nsqlookupd nsqadmin

nsqd nsqd是nsq服务的守护进程，他接收并且排着队向客户端发送消息。

nsqlookupd 这个组件从字面意思上来看就知道和查看相关，nsqd是管理整个消息拓扑图的守护程序，客户端查询nsqlookupd以发现nsqd特定主题的生产者。

nsqadmin 这是一个管理控制台，就是个简单的web ui ,让你可以清楚的看到各种信息。

了解完这些之后我们的思路就清楚了，如果我要使用nsq那么就需要把nsqlookupd启动起来让它去管理整个拓扑网络结构，与此同时我们还需要新建一个nsqd来做事情，然后启动nsqadmin查看任务详情、状态。
首先确保你的电脑上成功安装了docker,然后通过命令来加载镜像，这一部分代码没有特殊性和机器、环境没有关系。

docker pull nsqio/nsq
docker run --name lookupd -p 4160:4160 -p 4161:4161 nsqio/nsq /nsqlookupd

接下来，重新开一个命令行，或者在让lookupd 后台运行，然后运行一个nsqd容器，这一部分需要你填入host和port ,host就是你的ip地址，port就是服务监听的端口。

docker pull nsqio/nsq
docker run --name nsqd -p 4150:4150 -p 4151:4151 \
    nsqio/nsq /nsqd \
    --broadcast-address=<host> \
    --lookupd-tcp-address=<host>:<port>

到这一步就已经建立了一个超级简单的消息中间件，包含了所有你需要的功能，现在我们想要给他安装一个ui方便用户使用，记得修改成自己真是的host和port

docker run -d --name nsqadmin -p 4171:4171 nsqio/nsq /nsqadmin  --lookupd-http-address=<host>:<port>

【goalng】消息订阅收发消息中间件 NSQ 部署使用
上面就是安装好的，超级简洁。

接下来我们了解消息订阅收发机制，知乎上有个例子讲的很形象，大体意思知道，表述出来可能有出入，说是第一个月额时候小明的姐姐小红每次都按照妈妈的吩咐把自己的书给弟弟让他看，等弟弟看完了在给他，这样一个月下来，小明看了2本书，小红要每天都去检查小明看没看书然后给妈妈汇报，这样小红花了很多时间到不是自己的事情上；第二个月小红想我直接把书放在书架上让弟弟拿好了，她和弟弟说好之后每次弟弟书看完了都会去书架上看看姐姐有没有给她书，一个月下来，小明还是读了两本书，但是姐姐的时间全部被节省了下来做别的事情去了，这时候也会出现新的问题，小红没有办法及时向妈妈汇报小明的进度了，但是知道终究会看完的，所以这种方式会带来消息的延迟性即结果是可预期的但是什么时候会有结果是不知道的；第三个月，小明邀请小伙伴和他一起看书，姐姐告诉他们，书架第一层是给小明的，第二层是给小明朋友的，这样小明每次拿书就往第一层那，小明朋友就去第二层拿。这就是一个消息机制的进步史，第一个月的消息传递就相当于我们之前的代码，发来请求处理请求然后返回，第二个月高级了一点，发来请求，直接返回，并且向来的地方说这个东西我会干的但是需要等我有时间再干。第三个月更高级了支持好几个人同时来。
这样从中我们就可以抽取出来一些概念，在nsq中是topic 和channel的概念，一个topic就相当于例子中的那个书架，channel相当于书架上的第几层。这样通过一个topic和channel就能唯一确定一个用户了。废话不说了下面直接上代码，在这个代码中我们可以实现一个简单的聊天室的功能，查看效果的时候我们需要开三个终端，第一个运行main函数中的小明第二个运行main函数中的小红，然后第三个我们运行Publish函数来发送消息，记得把host填上自己的。

package main

// 消息收发机制的例子
import (
	"fmt"
	"sync"

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

//nsqd 节点地址和端口
var (
	//nsqd的地址，使用了tcp监听的端口
	tcpNsqdAddrr = "<host>:4150"
)

func main() {
	// 新建消费者   订阅all 主题  使用通道 c1
	// NewCustomer("小明", "all", "xiaoming")
	// NewCustomer("小红", "private", "xiaohong")

	// 新建生产者  生产一条问候消息  主题为all
	// Publish("private", "小红你好， 我是小A")

}

// Publish 发布函数   发布消息到指定的topic
// topic 主题
// content 内容
func Publish(topic, content string) {
	//初始化NSQ配置
	config := nsq.NewConfig()
	tPro, err := nsq.NewProducer(tcpNsqdAddrr, config)
	if err != nil {
		fmt.Println(err)
	}
	// 发布消息
	err = tPro.Publish(topic, []byte(content))
	if err != nil {
		fmt.Println(err)
	} else {
		fmt.Println("发送成功")
	}

}

// NsqHandler 声明一个结构体，实现HandleMessage接口方法
type NsqHandler struct {
	name string
	//标识ID
	// nsqHandlerID string
}

// HandleMessage 实现HandleMessage方法
// message是接收到的消息
func (s *NsqHandler) HandleMessage(message *nsq.Message) error {
	// 打印消息的一些基本信息
	fmt.Printf(s.name, "  收到了 ----->", string(message.Body))
	return nil
}

// NewCustomer 新建一个消费者
func NewCustomer(name, topic, channel string) {

	// 初始化配置
	config := nsq.NewConfig()
	// 创造消费者，参数一时订阅的主题，参数二是使用的通道
	com, err := nsq.NewConsumer(topic, channel, config)
	if err != nil {
		fmt.Println(err)
	}
	// 添加处理回调
	com.AddHandler(&NsqHandler{name: name})
	// 连接对应的nsqd
	err = com.ConnectToNSQD(tcpNsqdAddrr)
	if err != nil {
		fmt.Println(err)
	} else {
		fmt.Println("消费者", name)
	}
	// 只是为了不结束此进程，这里没有意义
	var wg = &sync.WaitGroup{}
	wg.Add(1)
	wg.Wait()

}

【goalng】消息订阅收发 消息中间件 NSQ 部署 使用

相关推荐

【goalng】消息订阅收发消息中间件 NSQ 部署使用