nsqlookupd

README

nsqlookupd

nsqlookupd is the daemon that manages topology metadata and serves client requests to discover the location of topics at runtime.

Read the docs

Documentation

Index

• type ClientV1
  • func NewClientV1(conn net.Conn) *ClientV1
  • func (c *ClientV1) String() string
• type Context
• type LookupProtocolV1
  • func (p *LookupProtocolV1) Exec(client *ClientV1, reader *bufio.Reader, params []string) ([]byte, error)
  • func (p *LookupProtocolV1) IDENTIFY(client *ClientV1, reader *bufio.Reader, params []string) ([]byte, error)
  • func (p *LookupProtocolV1) IOLoop(conn net.Conn) error
  • func (p *LookupProtocolV1) PING(client *ClientV1, params []string) ([]byte, error)
  • func (p *LookupProtocolV1) REGISTER(client *ClientV1, reader *bufio.Reader, params []string) ([]byte, error)
  • func (p *LookupProtocolV1) UNREGISTER(client *ClientV1, reader *bufio.Reader, params []string) ([]byte, error)
• type NSQLookupd
  • func New(opts *Options) *NSQLookupd
  • func (l *NSQLookupd) Exit()
  • func (l *NSQLookupd) Main()
  • func (l *NSQLookupd) RealHTTPAddr() *net.TCPAddr
  • func (l *NSQLookupd) RealTCPAddr() *net.TCPAddr
• type Options
  • func NewOptions() *Options
• type PeerInfo
• type Producer
  • func (p *Producer) IsTombstoned(lifetime time.Duration) bool
  • func (p *Producer) String() string
  • func (p *Producer) Tombstone()
• type Producers
  • func (pp Producers) FilterByActive(inactivityTimeout time.Duration, tombstoneLifetime time.Duration) Producers
  • func (pp Producers) PeerInfo() []*PeerInfo
• type Registration
  • func (k Registration) IsMatch(category string, key string, subkey string) bool
• type RegistrationDB
  • func NewRegistrationDB() *RegistrationDB
  • func (r *RegistrationDB) AddProducer(k Registration, p *Producer) bool
  • func (r *RegistrationDB) AddRegistration(k Registration)
  • func (r *RegistrationDB) FindProducers(category string, key string, subkey string) Producers
  • func (r *RegistrationDB) FindRegistrations(category string, key string, subkey string) Registrations
  • func (r *RegistrationDB) LookupRegistrations(id string) Registrations
  • func (r *RegistrationDB) RemoveProducer(k Registration, id string) (bool, int)
  • func (r *RegistrationDB) RemoveRegistration(k Registration)
• type Registrations
  • func (rr Registrations) Filter(category string, key string, subkey string) Registrations
  • func (rr Registrations) Keys() []string
  • func (rr Registrations) SubKeys() []string

Types

type ClientV1

type ClientV1 struct {
	net.Conn // 组合net.Conn接口
	// contains filtered or unexported fields
}

func NewClientV1

func NewClientV1(conn net.Conn) *ClientV1

初始化一个ClientV

func (*ClientV1) String

func (c *ClientV1) String() string

实现String接口

type Context

type Context struct {
	// contains filtered or unexported fields
}

type LookupProtocolV1

type LookupProtocolV1 struct {
	// contains filtered or unexported fields
}

func (*LookupProtocolV1) Exec

func (p *LookupProtocolV1) Exec(client *ClientV1, reader *bufio.Reader, params []string) ([]byte, error)

这个方法就是执行相应的命令动作 有 PING IDENTIFY REGISTER UNREGISTER 这四个类型

func (*LookupProtocolV1) IDENTIFY

func (p *LookupProtocolV1) IDENTIFY(client *ClientV1, reader *bufio.Reader, params []string) ([]byte, error)

func (*LookupProtocolV1) IOLoop

func (p *LookupProtocolV1) IOLoop(conn net.Conn) error

协议的主要处理函数

func (*LookupProtocolV1) PING

func (p *LookupProtocolV1) PING(client *ClientV1, params []string) ([]byte, error)

PING 用于client中的心跳处理命令

func (*LookupProtocolV1) REGISTER

func (p *LookupProtocolV1) REGISTER(client *ClientV1, reader *bufio.Reader, params []string) ([]byte, error)

具体协议报文如下 REGISTER topic1\n 这个只创建一个名字为topic1的topic 或如下报文 REGISTER topic1 channel1\n 这个只创建一个名字为topic1的topic 且topic1下面创建一个名字为channel1的channel

func (*LookupProtocolV1) UNREGISTER

func (p *LookupProtocolV1) UNREGISTER(client *ClientV1, reader *bufio.Reader, params []string) ([]byte, error)

type NSQLookupd

type NSQLookupd struct {
	sync.RWMutex // 读写锁

	DB *RegistrationDB // product 注册数据库 具体分析后面章节再讲
	// contains filtered or unexported fields
}

func New

func New(opts *Options) *NSQLookupd

初始化NSQLookupd实例

func (*NSQLookupd) Exit

func (l *NSQLookupd) Exit()

func (*NSQLookupd) Main

func (l *NSQLookupd) Main()

func (*NSQLookupd) RealHTTPAddr

func (l *NSQLookupd) RealHTTPAddr() *net.TCPAddr

func (*NSQLookupd) RealTCPAddr

func (l *NSQLookupd) RealTCPAddr() *net.TCPAddr

type Options

type Options struct {
	Verbose bool `flag:"verbose"` // enable verbose logging

	TCPAddress       string `flag:"tcp-address"`       //	TCP地址服务
	HTTPAddress      string `flag:"http-address"`      // 	HTTP地址服务
	BroadcastAddress string `flag:"broadcast-address"` // 	address of this lookupd node, (default to the OS hostname) (default "PROSNAKES.local")

	// 如果生产者最近一次ping距离现在的时间在InactiveProducerTimeout内，还是会被保存在活动队列中（FilterByActive中获取）
	InactiveProducerTimeout time.Duration `flag:"inactive-producer-timeout"` // duration of time a producer will remain in the active list since its last ping
	// 被标记为墓碑状态的生产者，如果registration存在，那么在这个时间内依然会存在
	TombstoneLifetime time.Duration `flag:"tombstone-lifetime"` // duration of time a producer will remain tombstoned if registration remains

	Logger logger
}

func NewOptions

func NewOptions() *Options

type PeerInfo

type PeerInfo struct {
	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"`
	// contains filtered or unexported fields
}

端点信息存储

type Producer

type Producer struct {
	// contains filtered or unexported fields
}

生产者定义

func (*Producer) IsTombstoned

func (p *Producer) IsTombstoned(lifetime time.Duration) bool

被标记为墓碑状态,同时距标记时间小于lifetime值。

func (*Producer) String

func (p *Producer) String() string

格式化输出信息

func (*Producer) Tombstone

func (p *Producer) Tombstone()

将Producer标记为墓碑状态（需要被删除）

type Producers

type Producers []*Producer

func (Producers) FilterByActive

func (pp Producers) FilterByActive(inactivityTimeout time.Duration, tombstoneLifetime time.Duration) Producers

获取所有可用的Producer

func (Producers) PeerInfo

func (pp Producers) PeerInfo() []*PeerInfo

获取Producers中所有的PeerInfo

type Registration

type Registration struct {
	Category string
	Key      string
	SubKey   string
}

func (Registration) IsMatch

func (k Registration) IsMatch(category string, key string, subkey string) bool

是否符合模糊匹配的规则

type RegistrationDB

type RegistrationDB struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

生产者和Registration关系的维护 同一个Registration对应多个生存者

func NewRegistrationDB

func NewRegistrationDB() *RegistrationDB

创建RegistrationDB对象

func (*RegistrationDB) AddProducer

func (r *RegistrationDB) AddProducer(k Registration, p *Producer) bool

add a producer to a registration 将一个Producer添加到指定的Registration里

func (*RegistrationDB) AddRegistration

func (r *RegistrationDB) AddRegistration(k Registration)

add a registration key 添加一个registration的key

func (*RegistrationDB) FindProducers

func (r *RegistrationDB) FindProducers(category string, key string, subkey string) Producers

根据category key subkey查找所有的Producer

func (*RegistrationDB) FindRegistrations

func (r *RegistrationDB) FindRegistrations(category string, key string, subkey string) Registrations

查找Registrations

func (*RegistrationDB) LookupRegistrations

func (r *RegistrationDB) LookupRegistrations(id string) Registrations

根据producer.peerInfo.id查找所属的registration(可能多个)

func (*RegistrationDB) RemoveProducer

func (r *RegistrationDB) RemoveProducer(k Registration, id string) (bool, int)

remove a producer from a registration 从registration里删除一个Producer

func (*RegistrationDB) RemoveRegistration

func (r *RegistrationDB) RemoveRegistration(k Registration)

remove a Registration and all it's producers 删除Registration和它对应的Producers

type Registrations

type Registrations []Registration

func (Registrations) Filter

func (rr Registrations) Filter(category string, key string, subkey string) Registrations

过滤获取所有与输入参数匹配的Registration

func (Registrations) Keys

func (rr Registrations) Keys() []string

获取MAP中所有Registration的key

func (Registrations) SubKeys

func (rr Registrations) SubKeys() []string

获取MAP中所有Registration的subkey