首页

    go net.Listener shutdown from caller with channel

    标签:go

    测试时,需要在每个test case传入不同的参数启动Server,它们都使用同一端口,所以在进入下一个test case前,需要关闭当前test case使用的Server

    channel

    首先想到的是定义一个channel类型的变量

    server.go

    var Quit chan bool
    
    func init() {
        Quit = make(chan bool)
    }
    

    要关闭Server时,向这个channel发送数据

    server_test.go

    Quit <- true
    

    接受方server.go

    func Serve(authType int) {
        l, err := net.Listen("tcp", "127.0.0.1:2000")
        ...
        defer func() {
            log.Println("listener close")
            l.Close()
        }()
    
        for {
            conn, _ := l.Accept()
            select {
            case <-Quit:
                log.Println("receive quit")
                return
            default:
                go ServeConn(conn, authType)
            }
        }
    }
    

    如果

    • 从channel收到数据,就return退出select语句,也退出了for语句,然后进入defer,关闭listener
    • 没收到数据,goroutine处理连接

    dead lock

    执行server_test.go,发现代码执行到Quit <- true就卡住了:

    • server_test.go没有结束当前test case
    • server.go也没有从channel收到数据

    net.Listener Accept()

    注意Accept()的注释

    Accept waits for and returns the next connection to the listener

    代码运行到conn, _ := l.Accept(),会等待(wait for),直到有新连接,才会运行后面的代码

    channel blocking

    另一方面,上面定义的Quit变量unbuffered channel,发送操作在接收者准备好之前是阻塞的.没有新连接时,接收方会'停留'在conn, _ := l.Accept(),不会进入后面的select语句,做好接收数据的准备,因而server_test.go也会卡住,没有结束当前test case

    goroutine net.Listener Accept()

    若是将Quit变量改为buffered channel,server_test.go发送方便不会阻塞,可以运行下一个test case,但是Server仍然没有接收到数据,因为没有新连接,Server会'停留'在conn, _ := l.Accept(),无法运行后面接收数据的代码.
    由此可以看到,conn, _ :=l.Accept()没有新连接会等待是问题的根源.可以

    1. 使用goroutineconn, _ :=l.Accept()不阻塞后面的从channel接收数据
    go func() {
        conn, err := l.Accept()
    }()
    
    1. 定义一个channel,当新连接过来时,向这个channel发送(通知)
    type accepted struct {
        conn net.Conn
        err  error
    }
    
    ch := make(chan accepted, 1)
    for {
        go func() {
            conn, err := l.Accept()
            ch <- accepted{conn, err}
        }()
        ...
    }
    
    1. 接收端收到新连接过来的通知,处理连接
    for {
        ...
        select {
        case a := <-ch:
            if a.err != nil {
                continue
            }
            go ServeConn(a.conn, authType)
        ...
        }
    }
    

    overall

    server.go

    var Quit chan bool
    
    func init() {
        Quit = make(chan bool)
    }
    
    func Serve(authType int) {
        l, err := net.Listen("tcp", ":2000")
        log.Println("server start")
        if err != nil {
            log.Fatal(err)
        }
        defer func() {
            log.Println("listener close")
            l.Close()
        }()
    
        ch := make(chan accepted, 1)
        for {
            go func() {
                conn, err := l.Accept()
                ch <- accepted{conn, err}
            }()
    
            select {
            case a := <-ch:
                if a.err != nil {
                    continue
                }
                go ServeConn(a.conn, authType)
            case <-Quit:
                log.Println("receive quit")
                return
            }
        }
    }
    

    server_test.go里面每个test结束时

    Quit <- true
    time.Sleep(1000 * time.Millisecond)
    

    就能关闭Server


    不定期更新