golang mq NSQ安装及API基本使用

「爱情、让人受尽委屈。」 2024-03-26 00:25 65阅读 0赞

## 安装 ##

curl -fsSL https://github.com/nsqio/nsq/releases/download/v1.2.1/nsq-1.2.1.linux-amd64.go1.16.6.tar.gz -o /D/Develop/
    
    cd /D/Develop/
    
    tar zxf nsq-1.2.1.linux-amd64.go1.16.6.tar.gz -C /D/Develop/
    
    cd nsq-1.2.1.linux-amd64.go1.16.6/
    
    # 启动nsqlookup
    ./bin/nsqlookupd &
    
    # 启动nsqd
    ./bin/nsqd --lookupd-tcp-address=127.0.0.1:4160 &
    
    # 运行 nsqadmin 管理
    ./bin/nsqadmin --lookupd-http-address=0.0.0.0:4161 --http-address=0.0.0.0:8761 &
    
    # 访问web界面
    http://localhost:8761
    
    # 客户端写入消息
    curl -d 'hello world 1' 'http://127.0.0.1:4151/pub?topic=hello-word'
    
    # 队列消息写入文件
    ./bin/nsq_to_file --topic=hello-world --output-dir=/tmp/nsq-log --lookupd-http-address=127.0.0.1:4161

## golang发送、消费消息 ##

![在这里插入图片描述][b9c0c5279c054832a9dc6d8dbe7bb370.png]

### producer ###

![在这里插入图片描述][6585afd443d0462aa27cc1f4e30ba72e.png]

// nsq-demo/product.go
    package main
    
    import (
    	"github.com/nsqio/go-nsq"
    	"io/ioutil"
    	"log"
    	"strconv"
    )
    
    var nullLogger = log.New(ioutil.Discard, "", log.LstdFlags)
    
    func sendMessage() {
        
    	config := nsq.NewConfig() // 1. 创建生产者
    	producer, err := nsq.NewProducer("127.0.0.1:4150", config)
    	if err != nil {
        
    		log.Fatalln("连接失败: (127.0.0.1:4150)", err)
    	}
    
    	errPing := producer.Ping() // 2. 生产者ping
    	if errPing != nil {
        
    		log.Fatalln("无法ping通: 127.0.0.1:4150", errPing)
    	}
    
    	producer.SetLogger(nullLogger, nsq.LogLevelInfo) // 3. 设置不输出info级别的日志
    
    	for i := 0; i < 5; i++ {
         // 4. 生产者发布消息
    		message := "消息发送测试 " + strconv.Itoa(i+10000)
    		err2 := producer.Publish("one-test", []byte(message)) // 注意one-test　对应消费者consumer.go　保持一致
    		if err2 != nil {
        
    			log.Panic("生产者推送消息失败!")
    		}
    	}
    
    	producer.Stop() // 5. 生产者停止执行
    }
    
    func main() {
        
    	sendMessage()
    }

### consumer ###

![在这里插入图片描述][9d9dfb4bfc7c40a1b064943c1c9c4a1a.png]

// nsq-demo/consumer.go
    package main
    
    import (
    	"fmt"
    	"github.com/nsqio/go-nsq"
    	"log"
    )
    
    func doConsumerTask() {
        
    	// 1. 创建消费者
    	config := nsq.NewConfig()
    	q, errNewCsmr := nsq.NewConsumer("one-test", "ch-one-test", config)
    	if errNewCsmr != nil {
        
    		fmt.Printf("fail to new consumer!, topic=%s, channel=%s", "one-test", "ch-one-test")
    	}
    
    	// 2. 添加处理消息的方法
    	q.AddHandler(nsq.HandlerFunc(func(message *nsq.Message) error {
        
    		log.Printf("message: %v", string(message.Body))
    		message.Finish()
    		return nil
    	}))
    
    	// 3. 通过http请求来发现nsqd生产者和配置的topic（推荐使用这种方式）
    	lookupAddr := []string{
        
    		"127.0.0.1:4161",
    	}
    	err := q.ConnectToNSQLookupds(lookupAddr)
    	if err != nil {
        
    		log.Panic("[ConnectToNSQLookupds] Could not find nsqd!")
    	}
    
    	// 4. 接收消费者停止通知
    	<-q.StopChan
    
    	// 5. 获取统计结果
    	stats := q.Stats()
    	fmt.Sprintf("message received %d, finished %d, requeued:%s, connections:%s",
    		stats.MessagesReceived, stats.MessagesFinished, stats.MessagesRequeued, stats.Connections)
    }
    
    func main() {
        
    	doConsumerTask()
    }

参考资料：  
[https://www.cnblogs.com/phpper/p/13408463.html][https_www.cnblogs.com_phpper_p_13408463.html]

[b9c0c5279c054832a9dc6d8dbe7bb370.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/25/89c3fa30549044c5b2547293d34f8fbf.png
[6585afd443d0462aa27cc1f4e30ba72e.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/25/cd422bd074d64670866695542de7a89d.png
[9d9dfb4bfc7c40a1b064943c1c9c4a1a.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/25/e829096c6c3d4ee7994cfcc77e7203f0.png
[https_www.cnblogs.com_phpper_p_13408463.html]: https://www.cnblogs.com/phpper/p/13408463.html