NSQ--nsqd初始化啓動

啓動的main函數在nsq/apps/nsqd/nsqd.go裏.

func main() {   
    prg := &program{}   
    if err := svc.Run(prg, syscall.SIGINT, syscall.SIGTERM); err != nil {      
        log.Fatal(err)   
    }
}


使用了svc的庫,程序安全退出、服務控制兩個方面。
核心在於系統信號獲取、Go Concurrency Patterns、以及基本的代碼封裝。
import "github.com/judwhite/go-svc/svc"
複製代碼

初始化處理

func (p *program) Init(env svc.Environment) error 複製代碼

啓動處理

func (p *program) Start() error 複製代碼

結束處理

func (p *program) Start() error複製代碼

主要看下Start():

func (p *program) Start() error { 
    //調用nsqFlagset和Parse進行命令行參數集初始化  
    opts := nsqd.NewOptions()   
    flagSet := nsqdFlagSet(opts)   
    flagSet.Parse(os.Args[1:])   
    rand.Seed(time.Now().UTC().UnixNano())   
    if flagSet.Lookup("version").Value.(flag.Getter).Get().(bool) {      
        fmt.Println(version.String("nsqd"))      
        os.Exit(0)   
    }   

    //判斷config參數是否存在，若存在的話還需進行配置文件的讀取
    var cfg config   
    configFile := flagSet.Lookup("config").Value.String()   
    if configFile != "" {      
        _, err := toml.DecodeFile(configFile, &cfg)      
        if err != nil {         
        log.Fatalf("ERROR: failed to load config file %s - %s", configFile, err.Error())      
        }   
    }  
    cfg.Validate()  
    //配置文件檢查經過後，建立默認配置opts，並於命令行參數和配置文件進行合併。
    options.Resolve(opts, flagSet, cfg) 
    //nsgd的代碼在nsg/nsgd/nsgd.go的裏面
    //建立一個nsgd  
    nsqd := nsqd.New(opts)   
    err := nsqd.LoadMetadata()   
    if err != nil {      
        log.Fatalf("ERROR: %s", err.Error())   
    }   
    err = nsqd.PersistMetadata()   
    if err != nil {      
        log.Fatalf("ERROR: failed to persist metadata - %s", err.Error())   
    }   
    nsqd.Main()   
    p.nsqd = nsqd   
    return nil
}複製代碼

LoadMetadata(),負責加載保存在文件中的Topic出來:

func (n *NSQD) LoadMetadata() error {   
    //使用atomic包中的方法來保證方法執行前和執行後isLoading值的改變
    atomic.StoreInt32(&n.isLoading, 1)   
    defer atomic.StoreInt32(&n.isLoading, 0)
    //元數據以json格式保存在nsqd可執行文件目錄下的nsqd.%d.dat中。其中%d爲表明該程序的ID，     //經過在啓動時的命令行worker-id或者配置文件中的id指定。默認ID是經過對主機名散列後得到。     //所以保證了同一臺機器每次啓動的ID相同。   
    fn := newMetadataFile(n.getOpts())   
    data, err := readOrEmpty(fn)   
    if err != nil {      
        return err   
    }   
    if data == nil {      
        return nil // fresh start   
    }   
    var m meta   
    err = json.Unmarshal(data, &m)   
    if err != nil {      
        return fmt.Errorf("failed to parse metadata in %s - %s", fn, err)   
    }   
    for _, t := range m.Topics {      
        //檢查topic名稱是否合法（長度在1-64之間，知足正則表達式^[\.a-zA-Z0-9_-]+(#ephemeral)?$） 
        //，若不合法則忽略      
        if !protocol.IsValidTopicName(t.Name) {         
            n.logf(LOG_WARN, "skipping creation of invalid topic %s", t.Name)         
            continue      
        }      
        //使用GetTopic函數經過名字得到topic對象      
        topic := n.GetTopic(t.Name)      
        //判斷當前topic對象是否處於暫停狀態，是的話調用Pause函數暫停topic      
        if t.Paused {         
            topic.Pause()      
        }     
         //獲取當前topic下全部的channel，而且遍歷channel，執行的操做與topic基本一致      
        for _, c := range t.Channels {         
            //檢查channel名稱是否合法（長度在1-64之間，知足正則表達式^[\.a-zA-Z0-9_-]+(#ephemeral)?$） 
            //，若不合法則忽略         
            if !protocol.IsValidChannelName(c.Name) {            
                n.logf(LOG_WARN, "skipping creation of invalid channel %s", c.Name)            
                continue         
            }         
            //使用GetChannel函數經過名字得到channel對象         
            channel := topic.GetChannel(c.Name)         
            //判斷當前channel對象是否處於暫停狀態，是的話調用Pause函數暫停channel         
            if c.Paused {            
                channel.Pause()         
            }      
        }      
        topic.Start()   
    }   
    return nil
}複製代碼

PersistMetadata()將當前的topic和channel信息寫入nsqd.%d.dat文件中:

func (n *NSQD) PersistMetadata() error {   
    // persist metadata about what topics/channels we have, across restarts   
    fileName := newMetadataFile(n.getOpts())   
    n.logf(LOG_INFO, "NSQ: persisting topic/channel metadata to %s", fileName)   
    js := make(map[string]interface{})   
    topics := []interface{}{}   
    //將topicMap轉成map[string]interface{}的map,轉成Json   
    for _, topic := range n.topicMap {      
        if topic.ephemeral {         
        continue      
        }      
        topicData := make(map[string]interface{})      
        topicData["name"] = topic.name      
        topicData["paused"] = topic.IsPaused()      
        channels := []interface{}{}      
        topic.Lock()      
        for _, channel := range topic.channelMap {         
           channel.Lock()         
            if channel.ephemeral {            
                channel.Unlock()            
                continue         
            }         
            channelData := make(map[string]interface{})         
            channelData["name"] = channel.name         
            channelData["paused"] = channel.IsPaused()         
            channels = append(channels, channelData)         
            channel.Unlock()      
        }      
        topic.Unlock()      
        topicData["channels"] = channels      
        topics = append(topics, topicData)   
    }   
    js["version"] = version.Binary   
    js["topics"] = topics   
    data, err := json.Marshal(&js)   
    if err != nil {      
        return err   
    }   
    //寫入文件時先建立擴展名爲tmp的臨時文件   
    tmpFileName := fmt.Sprintf("%s.%d.tmp", fileName, rand.Int())   
    //將topic和channel列表json序列化後寫回文件中   
    err = writeSyncFile(tmpFileName, data)   
    if err != nil {      
        return err   
    }   
    // 寫入內容後並保存後再調用atomicRename函數將tmp文件重命名爲nsqd.%d.dat。   
    err = os.Rename(tmpFileName, fileName)   
    if err != nil {      
        return err   
    }   
    // technically should fsync DataPath here   
    return nil
}複製代碼

參考至https://www.cnblogs.com/zhangboyu/p/7457061.html