nsqd New函数

深藏阁楼爱情的钟 2023-02-28 15:27 32阅读 0赞

func New(opts *Options) (*NSQD, error) {
        
    	var err error
    
    	dataPath := opts.DataPath	//从datapath读路径
    	if opts.DataPath == "" {
        
    		cwd, _ := os.Getwd()//没有配置默认在当前目录
    		dataPath = cwd
    	}
    	if opts.Logger == nil {
        
    		opts.Logger = log.New(os.Stderr, opts.LogPrefix, log.Ldate|log.Ltime|log.Lmicroseconds)//默认log配置
    	}
    
    	n := &NSQD{
        //初始化 map和chan
    		startTime:            time.Now(),
    		topicMap:             make(map[string]*Topic),
    		clients:              make(map[int64]Client),
    		exitChan:             make(chan int),
    		notifyChan:           make(chan interface{
        }),
    		optsNotificationChan: make(chan struct{
        }, 1),
    		dl:                   dirlock.New(dataPath),//初始化目录锁
    	}
    	//就是封装了http Client+ Client连接超时+Client请求超时，没有tlsConfig配置，毕竟只是http
    	httpcli := http_api.NewClient(nil, opts.HTTPClientConnectTimeout, opts.HTTPClientRequestTimeout)
    	n.ci = clusterinfo.New(n.logf, httpcli)//封装了log+http client
    
    	n.lookupPeers.Store([]*lookupPeer{
        })//atomic.Value 存储类型
    
    	n.swapOpts(opts)//n.opts 也是atomic.Value,存储
    	n.errValue.Store(errStore{
        })//errValue 同上
    
    	err = n.dl.Lock()
    	if err != nil {
        
    		return nil, fmt.Errorf("failed to lock data-path: %v", err)
    	}
    
    	if opts.MaxDeflateLevel < 1 || opts.MaxDeflateLevel > 9 {
        
    		return nil, errors.New("--max-deflate-level must be [1,9]")
    	}
    
    	if opts.ID < 0 || opts.ID >= 1024 {
        //ID经过md5后对1024取余 后保证了id在[0,1024)
    		return nil, errors.New("--node-id must be [0,1024)")
    	}
    
    	if opts.StatsdPrefix != "" {
        //statsd的前缀
    		var port string
    		_, port, err = net.SplitHostPort(opts.HTTPAddress)//取出httpAddress中port
    		if err != nil {
        
    			return nil, fmt.Errorf("failed to parse HTTP address (%s) - %s", opts.HTTPAddress, err)
    		}
    		//将broadcastaddress和port合成一个address，替换.成_ 生成statsdHostKey
    		statsdHostKey := statsd.HostKey(net.JoinHostPort(opts.BroadcastAddress, port))
    		prefixWithHost := strings.Replace(opts.StatsdPrefix, "%s", statsdHostKey, -1)//替换statsdPrefix中的%s 替换为statsdHostKey
    		if prefixWithHost[len(prefixWithHost)-1] != '.' {
        
    			prefixWithHost += "."
    		}
    		opts.StatsdPrefix = prefixWithHost
    	}
    
    	if opts.TLSClientAuthPolicy != "" && opts.TLSRequired == TLSNotRequired {
        //判断tlsclient 权限证书。以及是否需要tls
    		opts.TLSRequired = TLSRequired
    	}
    
    	tlsConfig, err := buildTLSConfig(opts)//生成TLS配置。
    	if err != nil {
        
    		return nil, fmt.Errorf("failed to build TLS config - %s", err)
    	}
    	if tlsConfig == nil && opts.TLSRequired != TLSNotRequired {
        
    		return nil, errors.New("cannot require TLS client connections without TLS key and cert")
    	}
    	n.tlsConfig = tlsConfig
    
    	for _, v := range opts.E2EProcessingLatencyPercentiles {
        
    		if v <= 0 || v > 1 {
        
    			return nil, fmt.Errorf("invalid E2E processing latency percentile: %v", v)
    		}
    	}
    
    	n.logf(LOG_INFO, version.String("nsqd"))
    	n.logf(LOG_INFO, "ID: %d", opts.ID)
    
    	n.tcpServer = &tcpServer{
        }
    	n.tcpListener, err = net.Listen("tcp", opts.TCPAddress)//监听TcpServer Listen
    	if err != nil {
        
    		return nil, fmt.Errorf("listen (%s) failed - %s", opts.TCPAddress, err)
    	}
    	n.httpListener, err = net.Listen("tcp", opts.HTTPAddress)//监听HTTPServer Listen
    	if err != nil {
        
    		return nil, fmt.Errorf("listen (%s) failed - %s", opts.HTTPAddress, err)
    	}
    	if n.tlsConfig != nil && opts.HTTPSAddress != "" {
        
    		n.httpsListener, err = tls.Listen("tcp", opts.HTTPSAddress, n.tlsConfig)//监听HTTPSServer Listen
    		if err != nil {
        
    			return nil, fmt.Errorf("listen (%s) failed - %s", opts.HTTPSAddress, err)
    		}
    	}
    
    	return n, nil
    }