nsq源码阅读 nsqlookupd源码五 http.go http_server.go

再来看看关于HTTP的处理逻辑：

httpServer := newHTTPServer(ctx)

新建HTTP实例，跟踪代码，nsqlookupd/http.go：

package nsqlookupd

import (
"fmt"
"net/http"
"net/http/pprof"
"sync/atomic"

"github.com/julienschmidt/httprouter"
"github.com/nsqio/nsq/internal/http_api"
"github.com/nsqio/nsq/internal/protocol"
"github.com/nsqio/nsq/internal/version"
)

type httpServer struct {
ctx *Context
router http.Handler
}

//新建httpServer实例
func newHTTPServer(ctx *Context) *httpServer {
log := http_api.Log(ctx.nsqlookupd.opts.Logger)

router := httprouter.New()
router.HandleMethodNotAllowed = true
router.PanicHandler = http_api.LogPanicHandler(ctx.nsqlookupd.opts.Logger)
router.NotFound = http_api.LogNotFoundHandler(ctx.nsqlookupd.opts.Logger)
router.MethodNotAllowed = http_api.LogMethodNotAllowedHandler(ctx.nsqlookupd.opts.Logger)
s := &httpServer{
ctx: ctx,
router: router,
}

//监听路由
router.Handle("GET", "/ping", http_api.Decorate(s.pingHandler, log, http_api.PlainText))
router.Handle("GET", "/info", http_api.Decorate(s.doInfo, log, http_api.V1))

// v1 negotiate
router.Handle("GET", "/debug", http_api.Decorate(s.doDebug, log, http_api.V1))
router.Handle("GET", "/lookup", http_api.Decorate(s.doLookup, log, http_api.V1))
router.Handle("GET", "/topics", http_api.Decorate(s.doTopics, log, http_api.V1))
router.Handle("GET", "/channels", http_api.Decorate(s.doChannels, log, http_api.V1))
router.Handle("GET", "/nodes", http_api.Decorate(s.doNodes, log, http_api.V1))

// only v1
router.Handle("POST", "/topic/create", http_api.Decorate(s.doCreateTopic, log, http_api.V1))
router.Handle("POST", "/topic/delete", http_api.Decorate(s.doDeleteTopic, log, http_api.V1))
router.Handle("POST", "/channel/create", http_api.Decorate(s.doCreateChannel, log, http_api.V1))
router.Handle("POST", "/channel/delete", http_api.Decorate(s.doDeleteChannel, log, http_api.V1))
router.Handle("POST", "/topic/tombstone", http_api.Decorate(s.doTombstoneTopicProducer, log, http_api.V1))

// debug
router.HandlerFunc("GET", "/debug/pprof", pprof.Index)
router.HandlerFunc("GET", "/debug/pprof/cmdline", pprof.Cmdline)
router.HandlerFunc("GET", "/debug/pprof/symbol", pprof.Symbol)
router.HandlerFunc("POST", "/debug/pprof/symbol", pprof.Symbol)
router.HandlerFunc("GET", "/debug/pprof/profile", pprof.Profile)
router.Handler("GET", "/debug/pprof/heap", pprof.Handler("heap"))
router.Handler("GET", "/debug/pprof/goroutine", pprof.Handler("goroutine"))
router.Handler("GET", "/debug/pprof/block", pprof.Handler("block"))
router.Handler("GET", "/debug/pprof/threadcreate", pprof.Handler("threadcreate"))

return s
}

func (s *httpServer) ServeHTTP(w http.ResponseWriter, req *http.Request) {
s.router.ServeHTTP(w, req)
}

//GET /ping
func (s *httpServer) pingHandler(w http.ResponseWriter, req *http.Request, ps httprouter.Params) (interface{}, error) {
return "OK", nil
}

//GET /info
func (s *httpServer) doInfo(w http.ResponseWriter, req *http.Request, ps httprouter.Params) (interface{}, error) {

/**
*相当于:
* type version struct {
* Version string `json:"version"`
* }
* v := &version {
* Version: version.Binary,
* }
* return v, nil
*/
return struct {
Version string `json:"version"`
}{
Version: version.Binary,
}, nil
}

//获取所有topic
//GET /topics
func (s *httpServer) doTopics(w http.ResponseWriter, req *http.Request, ps httprouter.Params) (interface{}, error) {
//获取所有topic的名称
//详情见nsqlookupd/registration_db.go文件中Registrations相关
topics := s.ctx.nsqlookupd.DB.FindRegistrations("topic", "*", "").Keys()
return map[string]interface{}{
"topics": topics,
}, nil
}

//获取指定topic下的所有channel
//GET /channels
//@param topic - the topic to list channels for
func (s *httpServer) doChannels(w http.ResponseWriter, req *http.Request, ps httprouter.Params) (interface{}, error) {
reqParams, err := http_api.NewReqParams(req)
if err != nil {
return nil, http_api.Err{400, "INVALID_REQUEST"}
}

topicName, err := reqParams.Get("topic")
if err != nil {
return nil, http_api.Err{400, "MISSING_ARG_TOPIC"}
}

channels := s.ctx.nsqlookupd.DB.FindRegistrations("channel", topicName, "*").SubKeys()
return map[string]interface{}{
"channels": channels,
}, nil
}

//获取指定topic下的所有有效producer
//GET /lookup
//@param topic - the topic to list producers for
func (s *httpServer) doLookup(w http.ResponseWriter, req *http.Request, ps httprouter.Params) (interface{}, error) {
reqParams, err := http_api.NewReqParams(req)
if err != nil {
return nil, http_api.Err{400, "INVALID_REQUEST"}
}

topicName, err := reqParams.Get("topic")
if err != nil {
return nil, http_api.Err{400, "MISSING_ARG_TOPIC"}
}

//topic不存在
registration := s.ctx.nsqlookupd.DB.FindRegistrations("topic", topicName, "")
if len(registration) == 0 {
return nil, http_api.Err{404, "TOPIC_NOT_FOUND"}
}

channels := s.ctx.nsqlookupd.DB.FindRegistrations("channel", topicName, "*").SubKeys()
producers := s.ctx.nsqlookupd.DB.FindProducers("topic", topicName, "")
//过滤出有效producer
//详情可查看nsqlookupd/registration_db.go中Producers的FilterByActive()方法
//和nsqlookupd/options.go中的一些默认配置值
producers = producers.FilterByActive(s.ctx.nsqlookupd.opts.InactiveProducerTimeout,
s.ctx.nsqlookupd.opts.TombstoneLifetime)
return map[string]interface{}{
"channels": channels,
"producers": producers.PeerInfo(),
}, nil
}

//POST /topic/create
//@param topic - the topic to created
func (s *httpServer) doCreateTopic(w http.ResponseWriter, req *http.Request, ps httprouter.Params) (interface{}, error) {
reqParams, err := http_api.NewReqParams(req)
if err != nil {
return nil, http_api.Err{400, "INVALID_REQUEST"}
}

topicName, err := reqParams.Get("topic")
if err != nil {
return nil, http_api.Err{400, "MISSING_ARG_TOPIC"}
}

//验证topic名称是否合法
if !protocol.IsValidTopicName(topicName) {
return nil, http_api.Err{400, "INVALID_ARG_TOPIC"}
}

s.ctx.nsqlookupd.logf("DB: adding topic(%s)", topicName)
key := Registration{"topic", topicName, ""}
s.ctx.nsqlookupd.DB.AddRegistration(key)

return nil, nil
}

//POST /topic/delete
//@param topic - the existing topic to delete
func (s *httpServer) doDeleteTopic(w http.ResponseWriter, req *http.Request, ps httprouter.Params) (interface{}, error) {
reqParams, err := http_api.NewReqParams(req)
if err != nil {
return nil, http_api.Err{400, "INVALID_REQUEST"}
}

topicName, err := reqParams.Get("topic")
if err != nil {
return nil, http_api.Err{400, "MISSING_ARG_TOPIC"}
}

//先删除topic下的channel
registrations := s.ctx.nsqlookupd.DB.FindRegistrations("channel", topicName, "*")
for _, registration := range registrations {
s.ctx.nsqlookupd.logf("DB: removing channel(%s) from topic(%s)", registration.SubKey, topicName)
s.ctx.nsqlookupd.DB.RemoveRegistration(registration)
}

//删除topic
registrations = s.ctx.nsqlookupd.DB.FindRegistrations("topic", topicName, "")
for _, registration := range registrations {
s.ctx.nsqlookupd.logf("DB: removing topic(%s)", topicName)
s.ctx.nsqlookupd.DB.RemoveRegistration(registration)
}

return nil, nil
}

//POST /topic/tombstone
//官方解释Tombstones a specific producer of an existing topic
//详情见http://nsq.io/components/nsqlookupd.html#deletion_tombstones
//个人理解为逻辑删除指定topic的一个节点(producer)
//@param topic - the existing topic
//@param node - the producer (nsqd) to tombstone (identified by <broadcast_address>:<http_port>)
func (s *httpServer) doTombstoneTopicProducer(w http.ResponseWriter, req *http.Request, ps httprouter.Params) (interface{
4000
}, error) {
reqParams, err := http_api.NewReqParams(req)
if err != nil {
return nil, http_api.Err{400, "INVALID_REQUEST"}
}

topicName, err := reqParams.Get("topic")
if err != nil {
return nil, http_api.Err{400, "MISSING_ARG_TOPIC"}
}

//node格式 <broadcast_address>:<http_port>
node, err := reqParams.Get("node")
if err != nil {
return nil, http_api.Err{400, "MISSING_ARG_NODE"}
}

s.ctx.nsqlookupd.logf("DB: setting tombstone for producer@%s of topic(%s)", node, topicName)
//根据topic查producer
producers := s.ctx.nsqlookupd.DB.FindProducers("topic", topicName, "")
for _, p := range producers {
thisNode := fmt.Sprintf("%s:%d", p.peerInfo.BroadcastAddress, p.peerInfo.HTTPPort)
//逻辑删除
if thisNode == node {
p.Tombstone()
}
}

return nil, nil
}

//POST /channel/create
//创建topic和channel
//@param topic - the topic to created
//@param channel - the channel to created
func (s *httpServer) doCreateChannel(w http.ResponseWriter, req *http.Request, ps httprouter.Params) (interface{}, error) {
reqParams, err := http_api.NewReqParams(req)
if err != nil {
return nil, http_api.Err{400, "INVALID_REQUEST"}
}

//获取topic和channel，并验证是否是有效参数
topicName, channelName, err := http_api.GetTopicChannelArgs(reqParams)
if err != nil {
return nil, http_api.Err{400, err.Error()}
}

//新增channel
s.ctx.nsqlookupd.logf("DB: adding channel(%s) in topic(%s)", channelName, topicName)
key := Registration{"channel", topicName, channelName}
s.ctx.nsqlookupd.DB.AddRegistration(key)

//新增topic
s.ctx.nsqlookupd.logf("DB: adding topic(%s)", topicName)
key = Registration{"topic", topicName, ""}
s.ctx.nsqlookupd.DB.AddRegistration(key)

return nil, nil
}

//POST /channel/delete
//删除已有topic下的一个channel
//@param topic - the existing topic
//@param channel - the existing channel to delete
func (s *httpServer) doDeleteChannel(w http.ResponseWriter, req *http.Request, ps httprouter.Params) (interface{}, error) {
reqParams, err := http_api.NewReqParams(req)
if err != nil {
return nil, http_api.Err{400, "INVALID_REQUEST"}
}

//获取topic和channel，并验证是否是有效参数
topicName, channelName, err := http_api.GetTopicChannelArgs(reqParams)
if err != nil {
return nil, http_api.Err{400, err.Error()}
}

//验证channel是否存在指定的topic下
registrations := s.ctx.nsqlookupd.DB.FindRegistrations("channel", topicName, channelName)
if len(registrations) == 0 {
return nil, http_api.Err{404, "CHANNEL_NOT_FOUND"}
}

s.ctx.nsqlookupd.logf("DB: removing channel(%s) from topic(%s)", channelName, topicName)
for _, registration := range registrations {
s.ctx.nsqlookupd.DB.RemoveRegistration(registration)
}

return nil, nil
}

//定义节点信息(producer (nsqd))
type node 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"`
Tombstones []bool `json:"tombstones"`
Topics []string `json:"topics"`
}

//GET /nodes
//获取所有已知节点(nsqd)
func (s *httpServer) doNodes(w http.ResponseWriter, req *http.Request, ps httprouter.Params) (interface{}, error) {
// don't filter out tombstoned nodes
producers := s.ctx.nsqlookupd.DB.FindProducers("client", "", "").FilterByActive(
s.ctx.nsqlookupd.opts.InactiveProducerTimeout, 0)
nodes := make([]*node, len(producers))
for i, p := range producers {
topics := s.ctx.nsqlookupd.DB.LookupRegistrations(p.peerInfo.id).Filter("topic", "*", "").Keys()

// for each topic find the producer that matches this peer
// to add tombstone information
tombstones := make([]bool, len(topics))
for j, t := range topics {
topicProducers := s.ctx.nsqlookupd.DB.FindProducers("topic", t, "")
for _, tp := range topicProducers {
if tp.peerInfo == p.peerInfo {
tombstones[j] = tp.IsTombstoned(s.ctx.nsqlookupd.opts.TombstoneLifetime)
}
}
}

nodes[i] = &node{
RemoteAddress: p.peerInfo.RemoteAddress,
Hostname: p.peerInfo.Hostname,
BroadcastAddress: p.peerInfo.BroadcastAddress,
TCPPort: p.peerInfo.TCPPort,
HTTPPort: p.peerInfo.HTTPPort,
Version: p.peerInfo.Version,
Tombstones: tombstones,
Topics: topics,
}
}

return map[string]interface{}{
"producers": nodes,
}, nil
}

//GET /debug
func (s *httpServer) doDebug(w http.ResponseWriter, req *http.Request, ps httprouter.Params) (interface{}, error) {
s.ctx.nsqlookupd.DB.RLock()
defer s.ctx.nsqlookupd.DB.RUnlock()

data := make(map[string][]map[string]interface{})
for r, producers := range s.ctx.nsqlookupd.DB.registrationMap {
key := r.Category + ":" + r.Key + ":" + r.SubKey
for _, p := range producers {
m := map[string]interface{}{
"id": p.peerInfo.id,
"hostname": p.peerInfo.Hostname,
"broadcast_address": p.peerInfo.BroadcastAddress,
"tcp_port": p.peerInfo.TCPPort,
"http_port": p.peerInfo.HTTPPort,
"version": p.peerInfo.Version,
"last_update": atomic.LoadInt64(&p.peerInfo.lastUpdate),
"tombstoned": p.tombstoned,
"tombstoned_at": p.tombstonedAt.UnixNano(),
}
data[key] = append(data[key], m)
}
}

return data, nil
}
http.go定义了一系列http handle

继续往下可看到：

http_api.Serve(httpListener, httpServer, "HTTP", l.opts.Logger)

创建HTTP server，代码位于internal/http_api/http_server.go里：

package http_api

import (
"fmt"
"log"
"net"
"net/http"
"strings"

"github.com/nsqio/nsq/internal/app"
)

type logWriter struct {
app.Logger
}

func (l logWriter) Write(p []byte) (int, error) {
l.Logger.Output(2, string(p))
return len(p), nil
}

//创建HTTP Serve，并注册handler，handler在nsqlookupd/http.go中定义
func Serve(listener net.Listener, handler http.Handler, proto string, l app.Logger) {
l.Output(2, fmt.Sprintf("%s: listening on %s", proto, listener.Addr()))

server := &http.Server{
Handler: handler,
ErrorLog: log.New(logWriter{l}, "", 0),
}
err := server.Serve(listener)
// theres no direct way to detect this error because it is not exposed
if err != nil && !strings.Contains(err.Error(), "use of closed network connection") {
l.Output(2, fmt.Sprintf("ERROR: http.Serve() - %s", err))
}

l.Output(2, fmt.Sprintf("%s: closing %s", proto, listener.Addr()))
}


到这里nsqlookupd模块的代码就看完了，还剩几个小文件就不在这写了。接下来会看nsqd模块的代码。