go - 使用 channel 捕获 Goroutine 的输出和错误

Question

我有一个调用函数的 for 循环 runCommand()它在交换机上运行远程命令并打印输出。该函数在每次迭代时在 goroutine 中调用，我使用的是 sync.Waitgroup同步 goroutines。现在，我需要一种方法来捕获 runCommand() 的输出和任何错误作用于 channel 。我已经阅读了很多文章并观看了很多关于将 channel 与 goroutines 一起使用的视频，但这是我第一次编写并发应用程序，我似乎无法理解这个想法。

基本上，我的程序从命令行获取主机名列表，然后异步连接到每个主机，在其上运行配置命令，并打印输出。如果出现错误，我的程序可以继续配置其余主机。

我将如何以惯用的方式将每次调用的输出或错误发送到 runCommand()到一个 channel 然后接收打印输出或错误？

这是我的代码:

package main

import (
    "fmt"
    "golang.org/x/crypto/ssh"
    "os"
    "time"
    "sync"
)

func main() {
    hosts := os.Args[1:]
    clientConf := configureClient("user", "password")

    var wg sync.WaitGroup
    for _, host := range hosts {
        wg.Add(1)
        go runCommand(host, &clientConf, &wg)
    }
    wg.Wait()

    fmt.Println("Configuration complete!")
}

// Run a remote command
func runCommand(host string, config *ssh.ClientConfig, wg *sync.WaitGroup) {
    defer wg.Done()
    // Connect to the client
    client, err := ssh.Dial("tcp", host+":22", config)
    if err != nil {
        fmt.Println(err)
        return
    }
    defer client.Close()
    // Create a session
    session, err := client.NewSession()
    if err != nil {
        fmt.Println(err)
        return
    }
    defer session.Close()
    // Get the session output
    output, err := session.Output("show lldp ne")
    if err != nil {
        fmt.Println(err)
        return
    }
    fmt.Print(string(output))
    fmt.Printf("Connection to %s closed.\n", host)
}

// Set up client configuration
func configureClient(user, password string) ssh.ClientConfig {
    var sshConf ssh.Config
    sshConf.SetDefaults()
    // Append supported ciphers
    sshConf.Ciphers = append(sshConf.Ciphers, "aes128-cbc", "aes256-cbc", "3des-cbc", "des-cbc", "aes192-cbc")
    // Create client config
    clientConf := &ssh.ClientConfig{
        Config:          sshConf,
        User:            user,
        Auth:            []ssh.AuthMethod{ssh.Password(password)},
        HostKeyCallback: ssh.InsecureIgnoreHostKey(),
        Timeout:         time.Second * 5,
    }
    return *clientConf
}

编辑: 我按照建议摆脱了 Waitgroup，现在我需要通过在打印输出和打印 Connection to <host> closed. 之前打印主机名来跟踪哪个输出属于哪个主机。 goroutine 完成时的消息。例如:

$ go run main.go host1[,host2[,...]]
Connecting to <host1>
[Output]
...
[Error]
Connection to <host1> closed.

Connecting to <host2>
...
Connection to <host2> closed.

Configuration complete!

我知道上面不一定会处理 host1和 host2按顺序，但我需要分别在输出/错误之前和之后为连接和关闭消息打印正确的主机值。我尝试推迟在 runCommand() 中打印结束消息功能，但消息在输出/错误之前打印出来。在每次 goroutine 调用后在 for 循环中打印结束消息也无法按预期工作。

更新代码:

package main

import (
    "fmt"
    "golang.org/x/crypto/ssh"
    "os"
    "time"
)

type CmdResult struct {
    Host string
    Output string
    Err error
}

func main() {
    start := time.Now()

    hosts := os.Args[1:]
    clientConf := configureClient("user", "password")
    results := make(chan CmdResult)

    for _, host := range hosts {
        go runCommand(host, &clientConf, results)
    }
    for i := 0; i < len(hosts); i++ {
        output := <- results
        fmt.Println(output.Host)
        if output.Output != "" {
            fmt.Printf("%s\n", output.Output)
        }
        if output.Err != nil {
            fmt.Printf("Error: %v\n", output.Err)
        }
    }
    fmt.Printf("Configuration complete! [%s]\n", time.Since(start).String())
}

// Run a remote command
func runCommand(host string, config *ssh.ClientConfig, ch chan CmdResult) {
    // This is printing before the output/error(s).
    // Does the same when moved to the bottom of this function.
    defer fmt.Printf("Connection to %s closed.\n", host)

    // Connect to the client
    client, err := ssh.Dial("tcp", host+":22", config)
    if err != nil {
        ch <- CmdResult{host, "", err}
        return
    }
    defer client.Close()
    // Create a session
    session, err := client.NewSession()
    if err != nil {
        ch <- CmdResult{host, "", err}
        return
    }
    defer session.Close()
    // Get the session output
    output, err := session.Output("show lldp ne")
    if err != nil {
        ch <- CmdResult{host, "", err}
        return
    }
    ch <- CmdResult{host, string(output), nil}
}

// Set up client configuration
func configureClient(user, password string) ssh.ClientConfig {
    var sshConf ssh.Config
    sshConf.SetDefaults()
    // Append supported ciphers
    sshConf.Ciphers = append(sshConf.Ciphers, "aes128-cbc", "aes256-cbc", "3des-cbc", "des-cbc", "aes192-cbc")
    // Create client config
    clientConf := &ssh.ClientConfig{
        Config:          sshConf,
        User:            user,
        Auth:            []ssh.AuthMethod{ssh.Password(password)},
        HostKeyCallback: ssh.InsecureIgnoreHostKey(),
        Timeout:         time.Second * 5,
    }
    return *clientConf
}

Answer 1

如果您使用无缓冲 channel ，您实际上不需要 sync.WaitGroup，因为您可以为将在 channel 上发送的每个 goroutine 调用 channel 上的接收操作符一次。每个接收操作都将阻塞，直到发送语句准备就绪，从而导致与 WaitGroup 相同的行为。

要做到这一点，请将 runCommand 更改为在所有条件下都在函数退出之前恰好执行一次 send 语句。

首先，创建一个类型以通过 channel 发送:

type CommandResult struct {
    Output      string
    Err         error
}

并编辑您的 main() {...} 以在 channel 上执行接收操作的次数与将发送到 channel 的 goroutine 的数量相同:

func main() {
    ch := make(chan CommandResult) // initialize an unbuffered channel
    // rest of your setup
    for _, host := range hosts {
        go runCommand(host, &clientConf, ch) // pass in the channel
    }
    for x := 0; x < len(hosts); x++ {
        fmt.Println(<-ch) // this will block until one is ready to send
    }

并编辑您的 runCommand 函数以接受 channel ，删除对 WaitGroup 的引用，并在所有条件下只执行一次发送:

func runCommand(host string, config *ssh.ClientConfig, ch chan CommandResult) {
    // do stuff that generates output, err; then when ready to exit function:
    ch <- CommandResult{output, err}
}

---

编辑:问题更新为标准输出消息顺序要求

I'd like to get nicely formatted output that ignores the order of events

在这种情况下，从 runCommand 中删除所有打印消息，您将把所有输出放入您在 channel 上传递的元素中，以便将它们组合在一起。编辑 CommandResult 类型以包含您要组织的其他字段，例如:

type CommandResult struct {
    Host        string
    Output      string
    Err         error
}

如果您不需要对结果进行排序，您可以继续打印收到的数据，例如

for x := 0; x < len(hosts); x++ {
    r := <-ch
    fmt.Printf("Host: %s----\nOutput: %s\n", r.Host, r.Output)
    if r.Err != nil {
        fmt.Printf("Error: %s\n", r.Err)
    }
}

如果您确实需要对结果进行排序，那么在您的 main goroutine 中，将在 channel 上接收到的元素添加到一个 slice 中:

    ...
    results := make([]CommandResult, 0, len(hosts))
    for x := 0; x < len(hosts); x++ {
        results = append(results, <-ch) // this will block until one is ready to send
    }

然后你可以使用the sort package在 Go 标准库中对结果进行排序以进行打印。例如，您可以按主机的字母顺序对它们进行排序。或者您可以将结果放入以主机字符串作为键而不是 slice 的映射中，以允许您按原始主机列表的顺序打印。