nsqd启动加载历史数据LoadMetadata

迷南。 2023-02-26 08:24 3阅读 0赞

### LoadMetadata ###

func (n *NSQD) LoadMetadata() error {
        
    	atomic.StoreInt32(&n.isLoading, 1)//函数开始 原子写入n.isLoading为1
    	defer atomic.StoreInt32(&n.isLoading, 0)//函数退出时 原子写入 n.isLoading为0
    	//我从其他goroutine中读n.isLoading如果为1 则表示nsqd正在读取历史数据。
    	fn := newMetadataFile(n.getOpts())//从n的opt里面拿出data-path 历史数据的文件目录地址
    
    	data, err := readOrEmpty(fn)//读取历史数据
    	if err != nil {
        
    		return err
    	}
    	if data == nil {
        
    		return nil // fresh start
    	}
    
    	var m meta
    	err = json.Unmarshal(data, &m)//json 反序列化
    	if err != nil {
        
    		return fmt.Errorf("failed to parse metadata in %s - %s", fn, err)
    	}
    
    	for _, t := range m.Topics {
        //遍历topic
    		if !protocol.IsValidTopicName(t.Name) {
        //判断是否有效topic name
    			n.logf(LOG_WARN, "skipping creation of invalid topic %s", t.Name)
    			continue
    		}
    		topic := n.GetTopic(t.Name)
    		if t.Paused {
        
    			topic.Pause()
    		}
    		for _, c := range t.Channels {
        
    			if !protocol.IsValidChannelName(c.Name) {
        
    				n.logf(LOG_WARN, "skipping creation of invalid channel %s", c.Name)
    				continue
    			}
    			channel := topic.GetChannel(c.Name)
    			if c.Paused {
        
    				channel.Pause()
    			}
    		}
    		topic.Start()
    	}
    	return nil
    }

type meta struct {
        
    	Topics []struct {
        
    		Name     string `json:"name"`
    		Paused   bool   `json:"paused"`
    		Channels []struct {
        
    			Name   string `json:"name"`
    			Paused bool   `json:"paused"`
    		} `json:"channels"`
    	} `json:"topics"`
    }

IsValidTopicName  
IsValidChannelName  
这两个函数都是利用正则进行匹配

`^[\.a-zA-Z0-9_-]+(#ephemeral)?$`（使用 PERL 语法）
    ^ ：匹配开始位置。
    [] - 字符集匹配，匹配方括号中定义的字符集之一
    + - 至少一次匹配（一次或更多次匹配）。

#### GetTopic ####

func (n *NSQD) GetTopic(topicName string) *Topic {
        
    	// 大多数情况下，我们都认为我们已经有了topic，这里进行二次尝试读取，第一次用读锁，第二次用写锁
    	n.RLock()
    	t, ok := n.topicMap[topicName]
    	n.RUnlock()
    	if ok {
        
    		return t
    	}
    
    	n.Lock()//写锁===============================
    
    	t, ok = n.topicMap[topicName]
    	if ok {
        
    		n.Unlock()//这里只是分支退出不能遗忘下面的释放锁
    		return t
    	}
    	//若内存中不存在topic 新建一个topic
    	deleteCallback := func(t *Topic) {
        //设置删除已存在的topic的回调函数
    		n.DeleteExistingTopic(t.name)
    	}
    	t = NewTopic(topicName, &context{
        n}, deleteCallback)
    	n.topicMap[topicName] = t
    
    	n.Unlock() //写锁=======================================锁的粒度到这里
    
    	n.logf(LOG_INFO, "TOPIC(%s): created", t.name)
    	// topic is created but messagePump not yet started
    	//如果正在读取数据状态中，topic不调用start函数启用
    	// if loading metadata at startup, no lookupd connections yet, topic started after load
    	if atomic.LoadInt32(&n.isLoading) == 1 {
        
    		return t
    	}
    	//此后的代码就不是loadmetadata用到的了
    	// if using lookupd, make a blocking call to get the topics, and immediately create them.
    	// this makes sure that any message received is buffered to the right channels
    	lookupdHTTPAddrs := n.lookupdHTTPAddrs()//生成lookupdHTTPAddrs地址
    	if len(lookupdHTTPAddrs) > 0 {
        
    		channelNames, err := n.ci.GetLookupdTopicChannels(t.name, lookupdHTTPAddrs)//从lookupd，根据topic name查询所属的channelname，开了goroutine 查询所有的lookupd，获取所有的channel name 去重，排序
    		if err != nil {
        
    			n.logf(LOG_WARN, "failed to query nsqlookupd for channels to pre-create for topic %s - %s", t.name, err)
    		}
    		for _, channelName := range channelNames {
        
    			if strings.HasSuffix(channelName, "#ephemeral") {
        
    				continue // do not create ephemeral channel with no consumer client
    			}
    			t.GetChannel(channelName)//判断内存中存在否，否则创建一个。更新chan的状态
    		}
    	} else if len(n.getOpts().NSQLookupdTCPAddresses) > 0 {
        
    		n.logf(LOG_ERROR, "no available nsqlookupd to query for channels to pre-create for topic %s", t.name)
    	}
    
    	// now that all channels are added, start topic messagePump
    	t.Start()//startChan向 写入值
    	return t
    }

#### GetChannel ####

func (t *Topic) GetChannel(channelName string) *Channel {
        
    	t.Lock()
    	channel, isNew := t.getOrCreateChannel(channelName)
    	t.Unlock()
    
    	if isNew {
        
    		// update messagePump state
    		select {
        
    		case t.channelUpdateChan <- 1:
    		case <-t.exitChan:
    		}
    	}
    
    	return channel
    }