> ### 十三例 NSQ操作
* 軟件下載(服務) :[https://nsq.io/deployment/installing.html](https://nsq.io/deployment/installing.html)
* * *
* nsqlookupd.exe
* nsqd.exe --lookupd-tcp-address=127.0.0.1:4160
* nsqadmin.exe --lookupd-http-address localhost:4161
* * *
* 一個節點下創建多個消息隊列, 給多個模塊消費

> ### 生產者
~~~
package main
import (
"bufio"
"fmt"
"os"
"strings"
"github.com/nsqio/go-nsq"
)
func main() {
nsqAddress := "127.0.0.1:4150"
config := nsq.NewConfig()
producer, err := nsq.NewProducer(nsqAddress, config)
if err != nil{
fmt.Println(err)
}
//讀取控制臺輸入
reader := bufio.NewReader(os.Stdin)
for {
data, err := reader.ReadString('\n')
if err != nil {
fmt.Printf("read string failed, err:%v\n", err)
continue
}
data = strings.TrimSpace(data)
if data == "stop" {
break
}
err = producer.Publish("queue", []byte(data))
if err != nil {
fmt.Printf("publish message failed, err:%v\n", err)
continue
}
fmt.Printf("publish data:%s succ\n", data)
}
}
~~~
> ### 消費者
~~~
package main
import (
"fmt"
"os"
"syscall"
"time"
"os/signal"
"github.com/nsqio/go-nsq"
)
// 消費者
type Consumer struct {
}
//處理消息
func (*Consumer) HandleMessage(msg *nsq.Message) error {
fmt.Println("receive", msg.NSQDAddress, "message:", string(msg.Body))
return nil
}
//初始化消費者
func initConsumer(topic string, channel string, address string) error {
cfg := nsq.NewConfig()
cfg.LookupdPollInterval = 15 * time.Second //設置服務發現的輪詢時間
c, err := nsq.NewConsumer(topic, channel, cfg) // 新建一個消費者
if err != nil {
return err
}
consumer := &Consumer{}
c.AddHandler(consumer) // 添加消費者接口
//建立NSQLookupd連接
if err := c.ConnectToNSQLookupd(address); err != nil {
return err
}
return nil
}
// 主函數
func main() {
err := initConsumer("queue", "channel1", "127.0.0.1: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
}
~~~
- 第一例 留言板
- 第二例 gRPC使用例子
- 第三例 基于go-micro做服務注冊和服務發現
- 第四例 聊天室
- 第五例 工具庫 第五例 并發安全字典
- dao
- common
- common.go
- config
- config.go
- gorm
- grom.go
- sqlx
- sqlx.go
- kafka
- kafka.go
- log
- log.go
- log2.go
- redis
- redis.go
- zookeeper
- zookeeper.go
- init
- main.go
- 第六例 原生sql操作
- 第七例 sqlx操作
- 第八例 Redis數據庫(gomodule/redigo)
- 第九例 Redis消息隊列
- 第十例 Redis集群連接
- 第十一例 Zookeeper操作
- 第十二例 Kafka操作
- 第十三例 NSQ操作
- 第十四例 二分查找
- 第十五例 交換排序 - 冒泡排序
- 第十六例 插入排序 - 直接插入排序
- 第十七例 插入排序 - 希爾排序
- 第十八例 交換排序 - 快速排序
- 第十九例 算法求解應用
- 第二十例 pprof性能分析
- 第二一例 CPU信息采集
- 第二二例 Heap信息采集
- 第二三例 Http信息采集
- 第二四例 單元測試(功能測試)
- 第二五例 基準測試(壓力測試/性能測試)
- 第二六例 gdb調試
- 第二七例 json序列化和反序列化
- 第二八例 protobuf序列化和反序列化
- 第二九例 包管理工具 go vendor
- 第三十例 包管理工具 go mod
- 第三一例 zip壓縮
- 第三二例 交叉編譯
- 第三三例 線上環境部署
- 第三四例 業務:實現固定周期維護
- 第三五例 聊天室(精簡版)
- 第三六例 并發安全字典
- 第三七例 導出Excel表格
- 第三八例 導出CSV表格
- 第三九例 聊天室(高并發)
- 第四十例 JWT (Json Web Token)
- 第四一例 雪花算法生成 Id
- 第四二例 對稱加密 AES
- 第四三例 非對稱加密 RSA
- 第四四例 簽名算法 SHA1
- 第四五例 數據庫操作 gorm
- 第四六例 數據庫操作 gorm 集合
- 數據庫連接和創建表
- 查詢 - 分頁
- 查詢所有數據
- 查詢單條數據
- 插入一條或多條數據
- 更新一條或多條數據
- 更新一條或多條數據(有零值)
- 第四七例 RSA(MD5WithRSA 算法)簽名和驗簽方式
- 第四八例 線上部署腳本
- 第四九例 Elasticsearch
- 第五十例 對象池
- 第五十一例 相關閱讀