// Week03/blog/01/01.go
package main
 
import (
	"log"
	"net/http"
	_ "net/http/pprof"
)
 
// 初始化函數(shù)
func setup() {
	// 這里面有一些初始化的操作
}
 
// 入口函數(shù)
func main() {
	setup()
 
	// 主服務(wù)
	server()
 
	// for debug
	pprof()
 
	select {}
}
 
// http api server
func server() {
	go func() {
		mux := http.NewServeMux()
		mux.HandleFunc("/ping", func(w http.ResponseWriter, r *http.Request) {
			w.Write([]byte("pong"))
		})
 
		// 主服務(wù)
		if err := http.ListenAndServe(":8080", mux); err != nil {
			log.Panicf("http server err: %+v", err)
			return
		}
	}()
}
 
// 輔助服務(wù)，用來debug性能測試
func pprof() {
	// 輔助服務(wù)，監(jiān)聽了其他端口，這里是 pprof 服務(wù)，用于 debug
	go http.ListenAndServe(":8081", nil)
}

package main
 
import (
    "log"
    "net/http"
    _ "net/http/pprof"
)
 
func setup(){
    // 初始化操作
}
 
 
func main(){
    
    setup()
    
    // for debug
    go pprof()
    
    // 主服務(wù)，http api
    go server()
    
    select{}
}
 
 
func server(){
    
    mux ：= http.NewServerMux()
    mux.HandleFunc("ping", func(w http.ResponseWriter, r * http.Request){
        w.Write([]byte("pong"))
    }
    
    // 主服務(wù)
    if err := http.ListerAndServer(":8080",mux); err != nil{
        log.panic("http server launch error: %v", err)
        return
    }
    
}
 
func pprof(){
    // 輔助服務(wù) 監(jiān)聽其他端口，這里是pprof服務(wù)，擁有debug
    http.ListerAndServer(":8081",nil)
}

package main
 
import (
	"log"
	"net/http"
	_ "net/http/pprof"
)
 
func setup() {
	// 這里面有一些初始化的操作
}
 
func main() {
	setup()
 
	// for debug
	go pprof()
 
	// 主服務(wù), http本來就是一個阻塞的服務(wù)
	server()
}
 
func server() {
	mux := http.NewServeMux()
	mux.HandleFunc("/ping", func(w http.ResponseWriter, r *http.Request) {
		w.Write([]byte("pong"))
	})
 
	// 主服務(wù)
	if err := http.ListenAndServe(":8080", mux); err != nil {
		log.Panicf("http server err: %+v", err)
		return
	}
}
 
func pprof() {
	// 輔助服務(wù)，監(jiān)聽了其他端口，這里是 pprof 服務(wù)，用于 debug
	http.ListenAndServe(":8081", nil)
}

package main
 
import (
	"context"
	"fmt"
	"log"
	"net/http"
	_ "net/http/pprof"
	"time"
)
 
func setup() {
	// 這里面有一些初始化的操作
}
 
func main() {
	setup()
 
	// 用于監(jiān)聽服務(wù)退出, 這里使用了兩個 goroutine，所以 cap 為2
	done := make(chan error, 2)
 
	// 無緩沖的通道，用于控制服務(wù)退出，傳入同一個 stop，做到只要有一個服務(wù)退出了那么另外一個服務(wù)也會隨之退出
	stop := make(chan struct{}, 0)
 
	// for debug
	go func() {
		//  pprof 傳遞一個 channel
		fmt.Println("pprof start...")
		done <- pprof(stop)
		fmt.Printf("err1:%v\n", done)
 
	}()
 
	// 主服務(wù)
	go func() {
		fmt.Println("app start...")
		done <- app(stop)
		fmt.Printf("err2:%v\n", done)
	}()
 
	// stopped 用于判斷當(dāng)前 stop 的狀態(tài)
	var stopped bool
 
	// 這里循環(huán)讀取 done 這個 channel
	// 只要有一個退出了，我們就關(guān)閉 stop channel
	for i := 0; i < cap(done); i++ {
 
		// 對于有緩沖的chan, chan中無值會一直處于阻塞狀態(tài)
	    // 對于app 服務(wù)會一直阻塞狀態(tài)，不會有 數(shù)據(jù)寫入到done 通道，只有在5s后，模擬的 pprof會有err寫入chan,此時才會觸發(fā)以下邏輯
		if err := <-done; err != nil {
			log.Printf("server exit err: %+v", err)
		}
 
		if !stopped {
			stopped = true
			// 通過關(guān)閉 無緩沖的channel 來通知所有的 讀 stop相關(guān)的goroutine退出
			close(stop)
		}
	}
}
 
// http 服務(wù)
func app(stop <-chan struct{}) error {
	mux := http.NewServeMux()
	mux.HandleFunc("/ping", func(w http.ResponseWriter, r *http.Request) {
		w.Write([]byte("pong"))
	})
 
	return server(mux, ":8080", stop)
}
 
func pprof(stop <-chan struct{}) error {
	// 注意這里主要是為了模擬服務(wù)意外退出，用于驗證一個服務(wù)退出，其他服務(wù)同時退出的場景
	// 因為這里沒有返回err， 所以done chan中無法接收到值， 主程序中會一直阻塞住
	go func() {
		server(http.DefaultServeMux, ":8081", stop)
	}()
 
	time.Sleep(5 * time.Second)
	// 模擬出錯
	return fmt.Errorf("mock pprof exit")
}
 
// 啟動一個服務(wù)
func server(handler http.Handler, addr string, stop <-chan struct{}) error {
 
	s := http.Server{
		Handler: handler,
		Addr:    addr,
	}
 
	// 這個 goroutine 控制退出，因為 stop channel 只要close或者是寫入數(shù)據(jù)，這里就會退出
	go func() {
		// 無緩沖channel等待，寫入或者關(guān)閉
		<-stop
		log.Printf("server will exiting, addr: %s", addr)
		// 此時 httpApi 服務(wù)就會優(yōu)雅的退出
		s.Shutdown(context.Background())
	}()
    
	// 沒有觸發(fā)異常的話，會一直處于阻塞
	return s.ListenAndServe()
}

package main
 
 
import (
    "log"
    _ "net/http/pprof"
    "net/http"
    
)
 
func setup() {
	// 這里面有一些初始化的操作
	log.Print("服務(wù)啟動初始化...")
}
 
func main() {
	setup()
 
	// for debug
	go pprof()
 
	// 主服務(wù), http本來就是一個阻塞的服務(wù)
	server()
}
 
func server() {
	mux := http.NewServeMux()
	mux.HandleFunc("/ping", func(w http.ResponseWriter, r *http.Request) {
		w.Write([]byte("pong"))
	})
	
	mux.HandleFunc("/leak", LeakHandle)
 
	// 主服務(wù)
	if err := http.ListenAndServe(":8080", mux); err != nil {
		log.Panicf("http server err: %+v", err)
		return
	}
}
 
func pprof() {
	// 輔助服務(wù)，監(jiān)聽了其他端口，這里是 pprof 服務(wù)，用于 debug
	http.ListenAndServe(":8081", nil)
}
 
func LeakHandle(w http.ResponseWriter, r *http.Request) {
	ch := make(chan bool, 0)
	go func() {
		fmt.Println("異步任務(wù)做一些操作")
		<-ch
	}()
 
	w.Write([]byte("will leak"))
}

// Reporter 埋點服務(wù)上報
type Reporter struct {
}
 
var reporter Reporter
 
// 模擬耗時
func (r Reporter) report(data string) {
	time.Sleep(time.Second)
	fmt.Printf("report: %s\n", data)
}
 
mux.HandleFunc("/ping", func(w http.ResponseWriter, r *http.Request) {
    // 在請求中異步調(diào)用
    // 這里并沒有滿足一致性
    go reporter.report("ping pong")
    fmt.Println("ping")
    w.Write([]byte("pong"))
})

$ go tun demo.go
ping
^C signal:interrupt

// 埋點上報
package main
 
import (
	"context"
	"fmt"
	"log"
	"net/http"
	"sync"
)
 
// Reporter 埋點服務(wù)上報
type Reporter struct {
	worker   int
	messages chan string
	wg       sync.WaitGroup
	closed   chan struct{}
	once     sync.Once
}
 
// NewReporter NewReporter
func NewReporter(worker, buffer int) *Reporter {
	return &Reporter{
		worker:   worker,
		messages: make(chan string, buffer),
		closed:   make(chan struct{}),
	}
}
 
// 執(zhí)行上報
func (r *Reporter) Run(stop <-chan struct{}) {
	// 用于執(zhí)行錯誤
	go func() {
		// 沒有錯誤時
		<-stop
		fmt.Println("stop...")
		r.shutdown()
	}()
 
	for i := 0; i < r.worker; i++ {
		r.wg.Add(1)
 
		go func() {
			defer r.wg.Done()
			for {
				select {
				case <-r.closed:
					return
				case msg := <-r.messages:
					fmt.Printf("report: %s\n", msg)
				}
			}
		}()
	}
 
	r.wg.Wait()
	fmt.Println("report workers exit...")
}
 
// 這里不必關(guān)閉 messages
// 因為 closed 關(guān)閉之后，發(fā)送端會直接丟棄數(shù)據(jù)不再發(fā)送
// Run 方法中的消費者也會退出
// Run 方法會隨之退出
func (r *Reporter) shutdown() {
	r.once.Do(func() { close(r.closed) })
}
 
// 模擬耗時
func (r *Reporter) Report(data string) {
	// 這個是為了及早退出
	// 并且為了避免我們消費者能力很強，發(fā)送者這邊一直不阻塞，可能還會一直寫數(shù)據(jù)
	select {
	case <-r.closed:
		fmt.Printf("reporter is closed, data will be discarded: %s \n", data)
	default:
	}
 
	select {
	case <-r.closed:
		fmt.Printf("reporter is closed, data will be discarded: %s \n", data)
	case r.messages <- data:
	}
}
 
func setup3() {
	// 初始化一些操作
	fmt.Println("程序啟動...")
}
 
func main() {
	setup3()
 
	// 用于監(jiān)聽服務(wù)完成時退出
	done := make(chan error, 3)
 
	// 實例化一個 reporter
	reporter := NewReporter(2, 100)
 
	// 用于控制服務(wù)退出，傳入同一個 stop，做到只要有一個服務(wù)退出了那么另外一個服務(wù)也會隨之退出
	stop := make(chan struct{}, 0)
 
	// for debug
	go func() {
		done <- pprof3(stop)
	}()
 
	// http主服務(wù)
	go func() {
		done <- app3(reporter, stop)
	}()
 
	// 上報服務(wù)，接收一個監(jiān)控停止的 chan
	go func() {
		reporter.Run(stop)
		done <- nil
	}()
 
	// 這里循環(huán)讀取 done 這個 channel
	// 只要有一個退出了，我們就關(guān)閉 stop channel
	for i := 0; i < cap(done); i++ {
 
		// 對于有緩沖的chan, chan中無值會一直處于阻塞狀態(tài)
		// 對于app 服務(wù)會一直阻塞狀態(tài)，不會有 數(shù)據(jù)寫入到done 通道，只有在5s后，模擬的 pprof會有err寫入chan,此時才會觸發(fā)以下邏輯
		if err := <-done; err != nil {
			log.Printf("server exit err: %+v", err)
		}
		// 通過關(guān)閉 無緩沖的channel 來通知所有的 讀 stop相關(guān)的goroutine退出
		close(stop)
	}
}
 
func pprof3(stop <-chan struct{}) error {
 
	// 輔助服務(wù)，監(jiān)聽了其他端口，這里是 pprof 服務(wù)，用于 debug
	err := server3(http.DefaultServeMux, ":8081", stop)
	return err
}
 
func app3(report *Reporter, stop <-chan struct{}) error {
 
	mux := http.NewServeMux()
	mux.HandleFunc("/ping", func(w http.ResponseWriter, r *http.Request) {
		// 在請求中異步調(diào)用
		// 這里并沒有滿足一致性
		go report.Report("ping pong")
		fmt.Println("ping")
 
		_, err := w.Write([]byte("pong"))
		if err != nil {
			log.Println("response err")
		}
	})
 
	return server3(mux, ":8080", stop)
}
 
// 啟動一個服務(wù)
func server3(handler http.Handler, addr string, stop <-chan struct{}) error {
 
	s := http.Server{
		Handler: handler,
		Addr:    addr,
	}
 
	// 這個 goroutine 控制退出，因為 stop channel 只要close 或者是寫入數(shù)據(jù)，這里就會退出
	go func() {
		// 無緩沖channel等待，寫入或者關(guān)閉
		<-stop
		log.Printf("server will exiting, addr: %s", addr)
		// 此時 httpApi 服務(wù)就會優(yōu)雅的退出
		err := s.Shutdown(context.Background())
		if err != nil {
			log.Printf("server exiting occur error, %s", err.Error())
		}
	}()
 
	// 沒有觸發(fā)異常的化，會一直處于阻塞
	return s.ListenAndServe()
}

// DeferRecover defer recover from panic.
func DeferRecover(tag string, handlePanic func(error)) func() {
	return func() {
		if err := recover(); err != nil {
			log.Errorf("%s, recover from: %v\n%s\n", tag, err, debug.Stack())
			if handlePanic != nil {
				handlePanic(fmt.Errorf("%v", err))
			}
		}
	}
}
 
// WithRecover recover from panic.
func WithRecover(tag string, f func(), handlePanic func(error)) {
	defer DeferRecover(tag, handlePanic)()
 
	f()
}
 
// Go is a wrapper of goroutine with recover.
func Go(name string, f func(), handlePanic func(error)) {
	go WithRecover(fmt.Sprintf("goroutine %s", name), f, handlePanic)
}