NSQ 源码阅读(二) NSQD

buptztd · · 1098 次点击 · · 开始浏览

这是一个创建于的文章，其中的信息可能已经有所发展或是发生改变。

第一次，站长亲自招 Gopher 了>>>

未完成

前言

NSQD是 nsq 的主要逻辑部分，请参考官方文档。我们直接看代码。

入口函数

main 函数位于

github.com/nsqio/nsq/apps/nsqd/nsqd.go

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

其中svc.Run 是一个service wrapper，来自于go-svc。它传入Service 接口，并且做了四件事情Init, Start, NotifySignal和 Stop:

// Run runs your Service.
//
// Run will block until one of the signals specified in sig is received.
// If sig is empty syscall.SIGINT and syscall.SIGTERM are used by default.
func Run(service Service, sig ...os.Signal) error {
    env := environment{}
    if err := service.Init(env); err != nil {
        return err
    }

    if err := service.Start(); err != nil {
        return err
    }

    if len(sig) == 0 {
        sig = []os.Signal{syscall.SIGINT, syscall.SIGTERM}
    }

    signalChan := make(chan os.Signal, 1)
    signalNotify(signalChan, sig...)
    <-signalChan

    return service.Stop()
}

让我们来看下program 对于Service 这个接口的实现

func (p *program) Init(env svc.Environment) error {
    if env.IsWindowsService() {
        dir := filepath.Dir(os.Args[0])
        return os.Chdir(dir)
    }
    return nil
}
func (p *program) Start() error {
    //ztd: 初始化选项
    opts := nsqd.NewOptions()
    //ztd: 设置接受的参数
    flagSet := nsqdFlagSet(opts)
    flagSet.Parse(os.Args[1:])

    rand.Seed(time.Now().UTC().UnixNano())
    //ztd: 如果仅仅查询版本信息
    if flagSet.Lookup("version").Value.(flag.Getter).Get().(bool) {
        fmt.Println(version.String("nsqd"))
        os.Exit(0)
    }
    //ztd: 如果指定了conf 文件
    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()
    //ztd: 合并命令行和配置文件中的配置
    options.Resolve(opts, flagSet, cfg)
    nsqd := nsqd.New(opts)
    //ztd: metadata 中存储了topic和 channel 信息
    err := nsqd.LoadMetadata()
    if err != nil {
        log.Fatalf("ERROR: %s", err.Error())
    }
    //ztd: 不是很明白为什么读完了马上又写
    err = nsqd.PersistMetadata()
    if err != nil {
        log.Fatalf("ERROR: failed to persist metadata - %s", err.Error())
    }
    nsqd.Main()

    p.nsqd = nsqd
    return nil
}

func (p *program) Stop() error {
    if p.nsqd != nil {
        p.nsqd.Exit()
    }
    return nil
}