如何编写在goroutine中执行的函数以发送和接收消息？

Question

I have to write a client side code that is receiving messages in the form of strings from server as well as taking input from the console to end message to the server. Both these operations should run concurrently with each other. I have written a code which perform these operations but sequentially not concurrently.

Here's what my current code looks like:

func SocketClient() {
    conn, err := net.Dial("tcp", ":9000")
    if err != nil {
        log.Fatal(err)
    }
    defer conn.Close()
    server_reader := bufio.NewReader(conn)
    input_reader := bufio.NewReader(os.Stdin)
    for {
        // for sending messages
        buff, err := input_reader.ReadString('
')
        if err != nil {
            log.Fatalln(err)
        }
        conn.Write([]byte(buff))

        // for receiving messages but this won't run until the above has taken input from console
        buff2, err := server_reader.ReadString('
')
        if err != nil {
            log.Fatalln(err)
        }
        fmt.Println("Received: %s", buff2)
    }
}

buff receives incoming messages from server and then buff2 takes outgoing message input from console but in order to receive the incoming messages again, buff2 needs some input. I know this can be done using channels, mutex locks etc. but because of my lack of understanding of fundamentals I am having problem using them.

I guess the actual code should look something like this:

func SocketClient() {
    conn, err := net.Dial("tcp", ":9000")
    if err != nil {
        log.Fatal(err)
    }
    defer conn.Close()
    go func() {
        // for sending messages
    } ()
    go func() {
        // for receiving messages
    } ()
}

How to make the input and output as two separate goroutines?

Answer 1

Run one of the loops directly in SocketClient. Run the other loop in a new goroutine started by SocketClient.

func SocketClient() error {
    conn, err := net.Dial("tcp", ":9000")
    if err != nil {
        return err
    }
    defer conn.Close()
    server_reader := bufio.NewReader(conn)
    input_reader := bufio.NewReader(os.Stdin)

    go func() {
        defer conn.Close()

        // This loop will break and the goroutine will exit when the
        // SocketClient function executes conn.Close().

        for {
            buff, err := server_reader.ReadBytes('
')
            if err != nil {
                // optional: log.Fatal(err) to cause program to exit without waiting for new input from stdin.
                return
            }
            fmt.Printf("Received: %s", buff)
        }
    }()

    for {
        // for sending messages
        buff, err := input_reader.ReadBytes('
')
        if err != nil {
            return err
        }
        if _, err := conn.Write(buff); err != nil {
            return err
        }
    }
}

Use the bufio.Reader ReadBytes method instead of the ReadString to avoid unnecessary conversions between []byte and string.

Answer 2

You can create a function like this one, that returns a channel. It also starts a new go routine that will write to the returned channel.

This allows consumers to call this method, and expect values back on the returned channel.

I've also updated the method signature to use a *bufio.Scanner as it's a more efficient method of splitting on newlines.

func ReadFully(scanner *bufio.Scanner) <-chan string {
    out := make(chan  string)

    go func() {
        // once all the values have been read, close the channel
        // this tells the consumers that there will not be anymore
        // values and they don't need to keep listening to this channel.
        defer close(out)

        for scanner.Scan() {
            out <- scanner.Text()
        }

        if err := scanner.Err(); err!= nil {
            log.Fatal(err)
        }
    }()

    return out
}

Now we can use this ReadFully method in SocketClient

func SocketClient() {
    conn, err := net.Dial("tcp", ":9000")
    if err != nil {
        log.Fatal(err)
    }
    defer conn.Close()

    serverScan := bufio.NewScanner(conn)
    inputScan := bufio.NewScanner(os.Stdin)


    serverCh := ReadFully(serverScan)
    inputCh := ReadFully(inputScan)


    // create a wait group, this stops the SocketClient 
    // method from exiting until the workers are done.
    var wg sync.WaitGroup
    wg.Add(2)

    go func() {
        defer wg.Done()
        for v := range inputCh {
            _, _ = conn.Write([]byte(v))
        }
    }()

    go func() {
        defer wg.Done()
        for v := range serverCh {
            fmt.Printf("Received: %s
", v)
        }
    }()

    wg.Wait()
}

Note, this example entirely omits cancellations etc. In general you should never start a go routine without a way to stop it.

Check out this great article on pipelines. https://blog.golang.org/pipelines

Answer 3

I wrote a blog post on Go concurrency. https://marcofranssen.nl/concurrency-in-go/

also used some of the concepts in this blog post.

https://marcofranssen.nl/go-webserver-with-gracefull-shutdown/

which also allows me to intercept commandline input like signals to safely shutdown the server.

You might use a similar approach to take your commanding input and user it.

The main idea is to use channels for communicating between your go routines.