• 安裝Nsq

            1.安裝包管理器

go get github.com/tools/godep

            2.安裝依賴包

go get github.com/bmizerany/assert

            3.安裝Nsq

https://github.com/nsqio/nsq/releases

• Producer

type Producer struct {
    P  *nsq.Producer
    ip string
}

// 初始化生產者,最好是在專案初始化的時候就呼叫,否則可能會出現orm註冊model衝突的情況
func InitProducer (addr string, mysql string) (wr *Producer) {
    // 註冊資料庫
    _, err := orm.GetDB("default_nsq")
    if err != nil {
        fmt.Println("orm.RegisterDataBase = ", mysql)
        err = orm.RegisterDataBase("default_nsq", "mysql", mysql, 10, 10)
    }
    if err != nil {
        panic("註冊資料庫失敗:" + err.Error())
    }

    // 生產者配置,這裡使用基本配置
    cfg := nsq.NewConfig()
    cfg.AuthSecret = ""                  // nsq認證金鑰,暫時不需要
    p, err := nsq.NewProducer(addr, cfg) // 新建一個生產者
    if err != nil {
        panic(err)
    }

    // 設定日誌級別
    p.SetLogger(log.New(os.Stdout, "nsq producer:", 0), nsq.LogLevelInfo)
    wr = &Producer{P: p}

    // 獲取生產者ip
    ip, err := getLocalIp()
    if err != nil {
        panic("nsq InitProducer:" + err.Error())
    }
    wr.ip = ip

    return wr
}

func (p *Producer) Stop() {
    p.P.Stop()
}

// 釋出訊息
func (p *Producer) Publish(topic string, body []byte) error {
    // 不能釋出空位元組陣列,否則會導致error
    if len(body) == 0 {
        return nil
    }
    go p.PublishLog(topic, uuid, body)
    return p.P.Publish(topic, body)
}

// 新增日誌 (需要編寫models)
func (p *Producer) PublishLog(topic, uuid string, body []byte) (int64, error) {
    log := &models.NsqPublishLog{}
    log.Message = string(body)
    log.NsqdUrl = p.P.String()
    log.Topic = topic
    log.ProducerIp = p.ip
    log.MessageId = uuid
    return models.AddNsqPublishLog(log)
}
 

• Consumer

type HandlerRegist struct {
    h       nsq.Handler
    topic   string
    channel string
    nsqd    string // address
    ip      string
}

// 初始化消費者 最好是在專案初始化的時候就呼叫,否則可能會出現orm註冊model衝突的情況
func InitConsumer(rg *HandlerRegist, mysql string) *nsq.Consumer {
    // 註冊資料庫
    _, err := orm.GetDB("default_nsq")
    if err != nil {
        fmt.Println("orm.RegisterDataBase = ", mysql)
        err = orm.RegisterDataBase("default_nsq", "mysql", mysql, 10, 10)
    }
    if err != nil {
    panic("InitConsumer註冊資料庫失敗:" + err.Error())
    }

    // 消費者配置,這裡使用基本配置
    cfg := nsq.NewConfig()
    cfg.LookupdPollInterval = time.Second                // 設定重連時間
    cfg.AuthSecret = ""                                  // nsq認證金鑰,暫時不需要
    c, err := nsq.NewConsumer(rg.topic, rg.channel, cfg) // 新建一個消費者
    if err != nil {
        panic(err)
    }

    // 設定日誌級別
    c.SetLogger(log.New(os.Stdout, "nsq consumer:", 0), nsq.LogLevelInfo)

    // 獲取消費者ip
    ip, err := getLocalIp()
    if err != nil {
        panic("nsq NewNsqConsumer:" + err.Error())
    }
    rg.ip = ip

    // 新增消費者介面
    c.AddHandler(rg)
    // 建立一個nsqd連線
    if err := c.ConnectToNSQD(rg.nsqd); err != nil {
        panic(err)
    }
    // 建立多個nsqd連線
    if err := c.ConnectToNSQDs([]string{"address1", "address2"}); err != nil {
        panic(err)
    }
    // 建立NSQLookupd連線
    if err := c.ConnectToNSQLookupd(rg.nsqd); err != nil {
        panic(err)
    }
    return c
}

// c.Stop()

// 處理訊息
func (rg *HandlerRegist) HandleMessage(message *nsq.Message) error {
	go rg.ConsumeLog(message) 
	return rg.h.HandleMessage(message)
}

// 新增日誌 (需要編寫models)
func (rg *HandlerRegist) ConsumeLog(message *nsq.Message) (int64, error) {
	log := &models.NsqConsumeLog{}
	log.ConsumerIp = rg.ip
	log.NsqdUrl = rg.nsqd
	log.Topic = rg.topic
	log.Channel = rg.channel
	log.Message = string(message.Body)
	log.MessageId = getMessageId(message.Body)
	return models.AddNsqConsumeLog(log)
}

func getMessageId(body []byte) string {
	m := map[string]interface{}{}
	json.Unmarshal(body, &m)
	if v, ok := m["nsq_msg_uuid"].(string); ok {
		return v
	}
	return ""
}
 

• IP

func getLocalIp() (string, error) {
    addrs, err := net.InterfaceAddrs()

    if err != nil {
        fmt.Println("獲取ip地址失敗")
        return "", err
    }
    var ip []string
    for _, address := range addrs {
        // 檢查ip地址判斷是否迴環地址
        if ipnet, ok := address.(*net.IPNet); ok && !ipnet.IP.IsLoopback() {
            if ipnet.IP.To4() != nil {
                ip = append(ip, ipnet.IP.String())
            }
        }
    }
    return strings.Join(ip, ";"), nil
}