Как написать функции, которые выполняются в goroutines для отправки и получения сообщений?

Question

Мне нужно написать код на стороне клиента, который получает сообщения в виде строк с сервера, а также принимает данные из консоли для завершения сообщения на сервер.Обе эти операции должны выполняться одновременно друг с другом.Я написал код, который выполняет эти операции, но последовательно не одновременно.

Вот как выглядит мой текущий код:

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('\n')
        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('\n')
        if err != nil {
            log.Fatalln(err)
        }
        fmt.Println("Received: %s", buff2)
    }
}

buff получает входящие сообщения от сервера, а затем buff2 принимает входящие сообщенияиз консоли, но для того, чтобы снова получать входящие сообщения, buff2 нуждается в некотором вводе.Я знаю, что это можно сделать с помощью каналов, мьютекс-блокировок и т. Д., Но из-за недостатка понимания основ у меня возникли проблемы с их использованием.

Я думаю, что реальный код должен выглядеть примерно так:

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
    } ()
}

Как сделать ввод и вывод как две отдельные программы?

Answer 1

Запустите один из циклов непосредственно в SocketClient.Запустите другой цикл в новой процедуре, запущенной 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('\n')
            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('\n')
        if err != nil {
            return err
        }
        if _, err := conn.Write(buff); err != nil {
            return err
        }
    }
}

. Используйте метод bufio.Reader ReadBytes вместо ReadString, чтобы избежать ненужных преобразований между []byte и string..

Answer 2

Я написал сообщение в блоге о параллелизме Go.https://marcofranssen.nl/concurrency-in-go/

также использовал некоторые концепции в этом сообщении в блоге.

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

, что также позволяет мне перехватывать ввод из командной строки, как сигналы для безопасногозавершите работу сервера.

Вы можете использовать аналогичный подход, чтобы взять свой командный ввод и использовать его.

Основная идея состоит в том, чтобы использовать каналы для связи между вашими процедурами go.

Answer 3

Вы можете создать такую ​​функцию, которая возвращает канал.Он также запускает новую подпрограмму go, которая будет записывать в возвращаемый канал.

Это позволяет потребителям вызывать этот метод и ожидать значения в возвращенном канале.

Я также обновилподпись метода для использования *bufio.Scanner, так как это более эффективный метод разбиения на новые строки.

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
}

Теперь мы можем использовать этот метод ReadFully в 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\n", v)
        }
    }()

    wg.Wait()
}

Обратите внимание, что этот пример полностью исключает отмены и т. Д. В общем, вы никогда не должны запускать процедуру go без способа ее остановить.

Ознакомьтесь с этой замечательной статьей о конвейерах.https://blog.golang.org/pipelines