nsq源码阅读(1)-启动和优雅退出

func main() {
	prg := &program{}
	// svc 框架的Run 方法启动一个service
	if err := svc.Run(prg, syscall.SIGINT, syscall.SIGTERM); err != nil {
		log.Fatal(err)
	}
}

// svc 框架, 启动 service, 也可以理解为 daemon
//
// 方法会一直阻塞, 直到 指定的信号量到来
// 如果指定信号为空, 默认是 syscall.SIGINT and syscall.SIGTERM
func Run(service Service, sig ...os.Signal) error {
	env := environment{}

	//先调用Init 初始化
	if err := service.Init(env); err != nil {
		return err
	}

	//再调用Start
	//其实有时候根本分不太清楚哪里是init哪里是start
	//作为一个框架, 当然要做的优雅一点
	if err := service.Start(); err != nil {
		return err
	}

	//准备信号量处理
	if len(sig) == 0 {
		sig = []os.Signal{syscall.SIGINT, syscall.SIGTERM}
	}

	signalChan := make(chan os.Signal, 1)
	signalNotify(signalChan, sig...)

	//整个程序会阻塞在这里, 等待系统信号量
	<-signalChan

	// 当信号来到, 调用stop 方法停掉
	return service.Stop()
}

func (p *program) Init(env svc.Environment) error {
	//svc 封装了不同操作系统的Deamon服务进程相关的东西
	//TODO window deamon的不同点
	if env.IsWindowsService() {
		dir := filepath.Dir(os.Args[0])
		return os.Chdir(dir)
	}
	return nil
}

func (p *program) Start() error {
	// 配置参数解析
	flagSet.Parse(os.Args[1:])

	if *showVersion {
		fmt.Println(version.String("nsqlookupd"))
		os.Exit(0)
	}

	var cfg map[string]interface{}
	if *config != "" {
		_, err := toml.DecodeFile(*config, &cfg)
		if err != nil {
			log.Fatalf("ERROR: failed to load config file %s - %s", *config, err.Error())
		}
	}

	opts := nsqlookupd.NewOptions()
	options.Resolve(opts, flagSet, cfg)
	daemon := nsqlookupd.New(opts)

	// 这里是主要的启动方法
	daemon.Main()
	p.nsqlookupd = daemon
	return nil
}

func (p *program) Stop() error {
	if p.nsqlookupd != nil {
		p.nsqlookupd.Exit()
	}
	return nil
}

func main() {
	p.nsqlookupd.Main()
	<-signalChan
	p.nsqlookupd.Exit()
}

func (l *NSQLookupd) Main() {
	ctx := &Context{l}

	tcpListener, err := net.Listen("tcp", l.opts.TCPAddress)
	if err != nil {
		l.logf("FATAL: listen (%s) failed - %s", l.opts.TCPAddress, err)
		os.Exit(1)
	}
	// 这里为什么要加锁?
	// 当前场景不好理解, 换个场景:
	//if l.playerList.count() > 0 {
	//	如果这里l 不 lock的话, 多线程 执行了 l.playerList = anotherPlayerList
	//	然后再 执行delete 就不对了
	//	l.playerList.delete(n)
	//}
	l.Lock()
	l.tcpListener = tcpListener
	l.Unlock()
	tcpServer := &tcpServer{ctx: ctx} // 这个就是handler

	//起一个tcp处理进程, 主要处理 client 和nsq server的连接
	l.waitGroup.Wrap(func() {
		protocol.TCPServer(tcpListener, tcpServer, l.opts.Logger)
	})

	//nsqlookup 还支持http操作, 所以再起一个线程
	httpListener, err := net.Listen("tcp", l.opts.HTTPAddress)
	if err != nil {
		l.logf("FATAL: listen (%s) failed - %s", l.opts.HTTPAddress, err)
		os.Exit(1)
	}
	l.Lock()
	l.httpListener = httpListener
	l.Unlock()
	httpServer := newHTTPServer(ctx)
	l.waitGroup.Wrap(func() {
		http_api.Serve(httpListener, httpServer, "HTTP", l.opts.Logger)
	})
}

type WaitGroupWrapper struct {
	sync.WaitGroup
}

func (w *WaitGroupWrapper) Wrap(cb func()) {
	w.Add(1) // +1
	// 起一个线程, 就是go 的 goroutine
	go func() {
		cb() //执行任务
		w.Done() // -1
	}()
}

w.Add(1)
go func() {
	protocol.TCPServer(tcpListener, tcpServer, l.opts.Logger)
	w.Done()
}()
w.Wait()  //Main 线程就会阻塞在这里; 等待TCPServer 结束时调用w.Done()后才退出

// ==================== 注意, 这里的 TCPServer 运行在一个新的 goroutine 中
// 这个TCPServer 也是一个可以复用的比较不错的 tcp 程序 accept 模式
func TCPServer(listener net.Listener, handler TCPHandler, l app.Logger) {
	l.Output(2, fmt.Sprintf("TCP: listening on %s", listener.Addr()))

	for {
		clientConn, err := listener.Accept()
		if err != nil {
			if nerr, ok := err.(net.Error); ok && nerr.Temporary() {
				l.Output(2, fmt.Sprintf("NOTICE: temporary Accept() failure - %s", err))
				runtime.Gosched()       // 是临时的错误, 暂停一下继续
				continue
			}
			// theres no direct way to detect this error because it is not exposed
			if !strings.Contains(err.Error(), "use of closed network connection") {
				l.Output(2, fmt.Sprintf("ERROR: listener.Accept() - %s", err))
			}
			break
		}
		go handler.Handle(clientConn)
	}

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

// ==================== 注意, 这里 是 Main 所在线程
func (l *NSQLookupd) Exit() {
	if l.tcpListener != nil {
		//Close 之后, listener.Accept() 就会返回error,
		//从而退出循环, TCPServer 线程退出
		l.tcpListener.Close()
	}

	if l.httpListener != nil {
		l.httpListener.Close()
	}
	l.waitGroup.Wait()
}

func (l *NSQLookupd) Main() {
	w.Add(1)
	go func() {
		//=== TCPServer 线程 start
		for {
			clientConn, err := listener.Accept()
			if err != nil {
				break
			}
			go handler.Handle(clientConn)
		}
		w.Done()
		//=== TCPServer 线程 end
	}()
}

func (l *NSQLookupd) Exit() {
	l.tcpListener.Close()
	l.waitGroup.Wait()
}

func main() {
	// 告诉wg 准备要启动一个goroutine
	waitGroup.Add(1)
	go func() {
		for {
			clientConn, err := listener.Accept()
			if err != nil {
				break
			}
			go handler.Handle(clientConn)
		}

		waitGroup.Done()
	}()

	// 当退出信号到达
	<-signalChan

	// 关闭listener
	tcpListener.Close()

	//检查并等待 所有的 goroutine 都结束, 而不是强制退出
	waitGroup.Wait()
}

func main() {
	// 告诉wg 准备要启动一个goroutine
	waitGroup.Add(1)
	go func() {
		for {
			clientConn, err := listener.Accept()
			if err != nil {
				break
			}
			go handler.Handle(clientConn)
		}

		waitGroup.Done()
	}()

	// 当退出信号到达
	<-signalChan

	// 关闭listener
	// tcp 服务线程也会因为accept 返回error 而退出
	tcpListener.Close()

	//检查并等待 所有的 goroutine 都结束, 而不是强制退出
	waitGroup.Wait()
}