Go并发编程之goroutine使用正确方法

目录

• 1. 对创建的gorouting负载

• 1.1 不要创建一个你不知道何时退出的 goroutine
• 1.2 不要帮别人做选择
• 1.3 不要作为一个旁观者
• 1.4 不要创建不知道什么时候退出的 goroutine
• 1.5 不要创建都无法退出的 goroutine
• 1.6 确保创建出的goroutine工作已经完成

// Week03/blog/01/01.go
package main
import (
	"log"
	"net/http"
	_ "net/http/pprof"
)
// 初始化函数
func setup() {
	// 这里面有一些初始化的操作
}
// 入口函数
func main() {
	setup()
	// 主服务
	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"))
		})
		// 主服务
		if err := http.ListenAndServe(":8080", mux); err != nil {
			log.Panicf("http server err: %+v", err)
			return
		}
	}()
}
// 辅助服务，用来debug性能测试
func pprof() {
	// 辅助服务，监听了其他端口，这里是 pprof 服务，用于 debug
	go http.ListenAndServe(":8081", nil)
}

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

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

package main
import (
	"context"
	"fmt"
	"log"
	"net/http"
	_ "net/http/pprof"
	"time"
)
func setup() {
	// 这里面有一些初始化的操作
}
func main() {
	setup()
	// 用于监听服务退出, 这里使用了两个 goroutine，所以 cap 为2
	done := make(chan error, 2)
	// 无缓冲的通道，用于控制服务退出，传入同一个 stop，做到只要有一个服务退出了那么另外一个服务也会随之退出
	stop := make(chan struct{}, 0)
	// for debug
	go func() {
		//  pprof 传递一个 channel
		fmt.Println("pprof start...")
		done <- pprof(stop)
		fmt.Printf("err1:%v\n", done)
	}()
	// 主服务
	go func() {
		fmt.Println("app start...")
		done <- app(stop)
		fmt.Printf("err2:%v\n", done)
	}()
	// stopped 用于判断当前 stop 的状态
	var stopped bool
	// 这里循环读取 done 这个 channel
	// 只要有一个退出了，我们就关闭 stop channel
	for i := 0; i < cap(done); i++ {
		// 对于有缓冲的chan, chan中无值会一直处于阻塞状态
	    // 对于app 服务会一直阻塞状态，不会有 数据写入到done 通道，只有在5s后，模拟的 pprof会有err写入chan,此时才会触发以下逻辑
		if err := <-done; err != nil {
			log.Printf("server exit err: %+v", err)
		}
		if !stopped {
			stopped = true
			// 通过关闭 无缓冲的channel 来通知所有的 读 stop相关的goroutine退出
			close(stop)
		}
	}
}
// http 服务
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 {
	// 注意这里主要是为了模拟服务意外退出，用于验证一个服务退出，其他服务同时退出的场景
	// 因为这里没有返回err， 所以done chan中无法接收到值， 主程序中会一直阻塞住
	go func() {
		server(http.DefaultServeMux, ":8081", stop)
	}()
	time.Sleep(5 * time.Second)
	// 模拟出错
	return fmt.Errorf("mock pprof exit")
}
// 启动一个服务
func server(handler http.Handler, addr string, stop <-chan struct{}) error {
	s := http.Server{
		Handler: handler,
		Addr:    addr,
	}
	// 这个 goroutine 控制退出，因为 stop channel 只要close或者是写入数据，这里就会退出
	go func() {
		// 无缓冲channel等待，写入或者关闭
		<-stop
		log.Printf("server will exiting, addr: %s", addr)
		// 此时 httpApi 服务就会优雅的退出
		s.Shutdown(context.Background())
	}()
	// 没有触发异常的话，会一直处于阻塞
	return s.ListenAndServe()
}

package main
import (
    "log"
    _ "net/http/pprof"
    "net/http"
)
func setup() {
	// 这里面有一些初始化的操作
	log.Print("服务启动初始化...")
}
func main() {
	setup()
	// for debug
	go pprof()
	// 主服务, http本来就是一个阻塞的服务
	server()
}
func server() {
	mux := http.NewServeMux()
	mux.HandleFunc("/ping", func(w http.ResponseWriter, r *http.Request) {
		w.Write([]byte("pong"))
	})
	mux.HandleFunc("/leak", LeakHandle)
	// 主服务
	if err := http.ListenAndServe(":8080", mux); err != nil {
		log.Panicf("http server err: %+v", err)
		return
	}
}
func pprof() {
	// 辅助服务，监听了其他端口，这里是 pprof 服务，用于 debug
	http.ListenAndServe(":8081", nil)
}
func LeakHandle(w http.ResponseWriter, r *http.Request) {
	ch := make(chan bool, 0)
	go func() {
		fmt.Println("异步任务做一些操作")
		<-ch
	}()
	w.Write([]byte("will leak"))
}

// Reporter 埋点服务上报
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) {
    // 在请求中异步调用
    // 这里并没有满足一致性
    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 埋点服务上报
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{}),
	}
}
// 执行上报
func (r *Reporter) Run(stop <-chan struct{}) {
	// 用于执行错误
	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...")
}
// 这里不必关闭 messages
// 因为 closed 关闭之后，发送端会直接丢弃数据不再发送
// Run 方法中的消费者也会退出
// Run 方法会随之退出
func (r *Reporter) shutdown() {
	r.once.Do(func() { close(r.closed) })
}
// 模拟耗时
func (r *Reporter) Report(data string) {
	// 这个是为了及早退出
	// 并且为了避免我们消费者能力很强，发送者这边一直不阻塞，可能还会一直写数据
	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()
	// 用于监听服务完成时退出
	done := make(chan error, 3)
	// 实例化一个 reporter
	reporter := NewReporter(2, 100)
	// 用于控制服务退出，传入同一个 stop，做到只要有一个服务退出了那么另外一个服务也会随之退出
	stop := make(chan struct{}, 0)
	// for debug
	go func() {
		done <- pprof3(stop)
	}()
	// http主服务
	go func() {
		done <- app3(reporter, stop)
	}()
	// 上报服务，接收一个监控停止的 chan
	go func() {
		reporter.Run(stop)
		done <- nil
	}()
	// 这里循环读取 done 这个 channel
	// 只要有一个退出了，我们就关闭 stop channel
	for i := 0; i < cap(done); i++ {
		// 对于有缓冲的chan, chan中无值会一直处于阻塞状态
		// 对于app 服务会一直阻塞状态，不会有 数据写入到done 通道，只有在5s后，模拟的 pprof会有err写入chan,此时才会触发以下逻辑
		if err := <-done; err != nil {
			log.Printf("server exit err: %+v", err)
		}
		// 通过关闭 无缓冲的channel 来通知所有的 读 stop相关的goroutine退出
		close(stop)
	}
}
func pprof3(stop <-chan struct{}) error {
	// 辅助服务，监听了其他端口，这里是 pprof 服务，用于 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) {
		// 在请求中异步调用
		// 这里并没有满足一致性
		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)
}
// 启动一个服务
func server3(handler http.Handler, addr string, stop <-chan struct{}) error {
	s := http.Server{
		Handler: handler,
		Addr:    addr,
	}
	// 这个 goroutine 控制退出，因为 stop channel 只要close 或者是写入数据，这里就会退出
	go func() {
		// 无缓冲channel等待，写入或者关闭
		<-stop
		log.Printf("server will exiting, addr: %s", addr)
		// 此时 httpApi 服务就会优雅的退出
		err := s.Shutdown(context.Background())
		if err != nil {
			log.Printf("server exiting occur error, %s", err.Error())
		}
	}()
	// 没有触发异常的化，会一直处于阻塞
	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)
}