go语言nsq源码解读五 nsqlookupd源码registration_db.go

本篇将讲解registration_db.go文件。
package nsqlookupd

import (
    "fmt"
    "sync"
    "time"
)

//定义了类型RegistrationDB，按字面意思：注册数据库，可理解为保存注册信息的数据库。
type RegistrationDB struct {
    //嵌入了RWMutex，所以在后面在方法中才可以调用Lock和Unlock方法
    sync.RWMutex
    //一个MAP，以Registration为键，Producers为值
    //看下面的代码中Producers的定义，它是Producer的slice，从中可看出Registration和Producer是一对多的关系
    registrationMap map[Registration]Producers
}

//定义类型Registration，
type Registration struct {
    Category string
    Key      string
    SubKey   string
}

//定义类型Registrations为Registration的slice
type Registrations []Registration

//定义类型PeerInfo
type PeerInfo struct {
    id               string
    RemoteAddress    string `json:"remote_address"`
    Hostname         string `json:"hostname"`
    BroadcastAddress string `json:"broadcast_address"`
    TcpPort          int    `json:"tcp_port"`
    HttpPort         int    `json:"http_port"`
    Version          string `json:"version"`
    lastUpdate       time.Time
}

//定义类型Producer
type Producer struct {
    //PeerInfo的指针
    peerInfo *PeerInfo
    //是否要被移除（tombstoned）
    tombstoned bool
    //移除时间
    tombstonedAt time.Time
}

//定义类型Producers为Producer的slice
type Producers []*Producer

//Producer的String方法
func (p *Producer) String() string {
    return fmt.Sprintf("%s [%d, %d]", p.peerInfo.BroadcastAddress, p.peerInfo.TcpPort, p.peerInfo.HttpPort)
}

//本方法将Producer标记为墓碑状态
func (p *Producer) Tombstone() {
    p.tombstoned = true
    p.tombstonedAt = time.Now()
}

//被标记为墓碑状态,同时距标记时间小于lifetime值。
//比如：在0分0秒时调用了上一个函数Tombstone(),在0分1秒时调用函数IsTombstoned(5)，返回结果为true
func (p *Producer) IsTombstoned(lifetime time.Duration) bool {
    return p.tombstoned && time.Now().Sub(p.tombstonedAt) < lifetime
}

//新建RegistrationDB类型的变量
func NewRegistrationDB() *RegistrationDB {
    return &RegistrationDB{
        //make一个map
        registrationMap: make(map[Registration]Producers),
    }
}

//添加一个registration的key，只是把map的key设置了，value为一个空的Producers slice
func (r *RegistrationDB) AddRegistration(k Registration) {
    r.Lock()
    defer r.Unlock()
    _, ok := r.registrationMap[k]
    if !ok {
        r.registrationMap[k] = make(Producers, 0)
    }
}

//将一个Producer添加到指定的Registration里
func (r *RegistrationDB) AddProducer(k Registration, p *Producer) bool {
    r.Lock()
    defer r.Unlock()
    //producers是一个slice
    producers := r.registrationMap[k]

    //遍历producers，看这个要添加的Producer是否已经存在了
    found := false
    for _, producer := range producers {
        //通过producer.peerInfo.id来判断是否为同一个Producer
        if producer.peerInfo.id == p.peerInfo.id {
            found = true
        }
    }

    //只有要添加的Producer不存在时，才添加到Registration里
    if found == false {
        r.registrationMap[k] = append(producers, p)
    }
    return !found
}

// 根据producer.peerInfo.id从registration里删除一个Producer
func (r *RegistrationDB) RemoveProducer(k Registration, id string) (bool, int) {
    r.Lock()
    defer r.Unlock()

    //map中不存在key为k的Registration记录，所以也就无需删除了。
    producers, ok := r.registrationMap[k]
    if !ok {
        return false, 0
    }
    removed := false
    //创建空的Producers slice，所有不需要删除的Producer都移动这个slice里，并在移除完成后重新赋值给Registration
    //注意学习这种从slice中移除一个元素的方式
    cleaned := make(Producers, 0)
    for _, producer := range producers {
        //id不相同，不是要删除的producer，移到cleaned slice里
        if producer.peerInfo.id != id {
            cleaned = append(cleaned, producer)
        } else {
            removed = true
        }
    }
    // Note: this leaves keys in the DB even if they have empty lists
    //重新赋值，这确保了即使cleaned是一个空slice,键值k仍然会在RegistrationDB中存在
    r.registrationMap[k] = cleaned

    //返回两个值，是否删除，新slice的长度
    return removed, len(cleaned)
}

// 删除Registration和它对应的Producers
func (r *RegistrationDB) RemoveRegistration(k Registration) {
    r.Lock()
    defer r.Unlock()
    delete(r.registrationMap, k)
}

//查找Registrations，可以看到，传入参数的三个变量与Registration类型里的三个值是对应的
func (r *RegistrationDB) FindRegistrations(category string, key string, subkey string) Registrations {
    r.RLock()
    defer r.RUnlock()
    results := make(Registrations, 0)
    for k := range r.registrationMap {
        //找出registrationMap中所有category，key，subkey与入参相同的Registration
        //IsMatch方法在后面的代码中定义
        if !k.IsMatch(category, key, subkey) {
            continue
        }
        results = append(results, k)
    }
    return results
}

//根据category key subkey查找所有的Producer
func (r *RegistrationDB) FindProducers(category string, key string, subkey string) Producers {
    r.RLock()
    defer r.RUnlock()
    results := make(Producers, 0)
    //遍历map
    for k, producers := range r.registrationMap {
        if !k.IsMatch(category, key, subkey) {
            continue
        }
        //遍历每个registration下的producers
        for _, producer := range producers {
            found := false
            //判断producer是否已经存在了，如果存在的话，就不添加了
            for _, p := range results {
                if producer.peerInfo.id == p.peerInfo.id {
                    found = true
                }
            }
            if found == false {
                results = append(results, producer)
            }
        }
    }
    return results
}

//根据producer.peerInfo.id查找所属的registration key
func (r *RegistrationDB) LookupRegistrations(id string) Registrations {
    r.RLock()
    defer r.RUnlock()
    results := make(Registrations, 0)
    //遍历map
    for k, producers := range r.registrationMap {
        //遍历每个registration下的producers
        for _, p := range producers {
            if p.peerInfo.id == id {
                results = append(results, k)
                break
            }
        }
    }
    return results
}

//依据Registration类型里的三个变量，判断是否与Registration匹配
func (k Registration) IsMatch(category string, key string, subkey string) bool {
    if category != k.Category {
        return false
    }
    if key != "*" && k.Key != key {
        return false
    }
    if subkey != "*" && k.SubKey != subkey {
        return false
    }
    return true
}

//过滤获取所有与输入参数匹配的Registration
func (rr Registrations) Filter(category string, key string, subkey string) Registrations {
    output := make(Registrations, 0)
    for _, k := range rr {
        if k.IsMatch(category, key, subkey) {
            output = append(output, k)
        }
    }
    return output
}

//获取MAP中所有Registration的key
func (rr Registrations) Keys() []string {
    keys := make([]string, len(rr))
    for i, k := range rr {
        keys[i] = k.Key
    }
    return keys
}

//获取MAP中所有Registration的subkey
func (rr Registrations) SubKeys() []string {
    subkeys := make([]string, len(rr))
    for i, k := range rr {
        subkeys[i] = k.SubKey
    }
    return subkeys
}

//获取所有可用的Producer
func (pp Producers) FilterByActive(inactivityTimeout time.Duration, tombstoneLifetime time.Duration) Producers {
    now := time.Now()
    results := make(Producers, 0)
    for _, p := range pp {
        //满足以下两个判断条件的producer被忽略
        //1 超过了活跃时间，在inactivityTimeout时间内没有与nsqlookupd交互
        //2 被标记为墓碑状态，在tombstoneLifetime时间内标记的producer将被过滤掉
        if now.Sub(p.peerInfo.lastUpdate) > inactivityTimeout || p.IsTombstoned(tombstoneLifetime) {
            continue
        }
        results = append(results, p)
    }
    return results
}

//获取Producers中所有的PeerInfo
func (pp Producers) PeerInfo() []*PeerInfo {
    results := make([]*PeerInfo, 0)
    for _, p := range pp {
        results = append(results, p.peerInfo)
    }
    return results
}
读过上述代码，可总结出，registration_db.go文件用MAP以一对多的形式保存Producer，并提供一系列增、删、改、查的操作封装。同时使用RWMutex做并发控制。