在这种情况下如何正确循环缓冲通道？

Posted 2023-03-14

【中文标题】在这种情况下如何正确循环缓冲通道？

【英文标题】：How to properly loop through buffered channel in this case?

【发布时间】：2015-10-12 21:30:48

【问题描述】：

我正在尝试使用 stdlib 制作某种端口扫描器。这更像是一个练习，所以请不要评论所涉及的逻辑。

看下面的代码：

package main

import (
    "flag"
    "fmt"
    "net"
    "time"
    "strings"
    "strconv"
    "log"
    "sync"
)

var commonPorts = map[int]string 
    21: "ftp",
    22: "sftp",
    80: "http",
    110: "pop3",
    143: "imap",
    443: "https",
    631: "ipp",
    993: "imaps",
    995: "pop3s",


type OP struct 
    mu sync.Mutex
    ports []string


func (o *OP) SafeAdd(port string) 
    o.mu.Lock()
    defer o.mu.Unlock()
    o.ports = append(o.ports, port)



func worker(host string, port int) string 
    address := fmt.Sprintf("%s:%d", host, port)

    conn, err := net.DialTimeout("tcp", address, time.Second * 3)
    if err != nil 
        return ""; // is offline, cannot connect
    
    conn.Close()

    stringI := strconv.Itoa(port)
    if name, ok := commonPorts[port]; ok 
        stringI += fmt.Sprintf("(%s)", name)
    

    return stringI


func processWithChannels(host string) <-chan string
    openPort := make(chan string, 1000)
    var wg sync.WaitGroup
    for i := 1; i <= 65535; i++ 
        wg.Add(1)
        go func(openPort chan string, host string, i int) 
            defer wg.Done()
            port := worker(host, i)
            if port != "" 
                openPort <- port
            
        (openPort, host, i)
    
    wg.Wait()
    close(openPort)
    return openPort


func main() 
    var host = flag.String("host", "127.0.0.1", "please insert the host")
    var pType = flag.Int("type", 2, "please insert the type")

    flag.Parse()
    fmt.Printf("Scanning: %s...\n", *host)

    if _, err := net.LookupHost(*host); err != nil 
        log.Fatal(err)
    

    openPorts := &OPports: []string;

    if *pType == 1 

        ports := processWithChannels(*host);
        for port := range ports 
            openPorts.SafeAdd(port)
        

     else 

        var wg sync.WaitGroup
        for i := 1; i <= 65535; i++ 
            wg.Add(1)
            go func(o *OP, host string, i int)
                defer wg.Done()
                if port := worker(host, i); port != "" 
                    o.SafeAdd(port)
                
            (openPorts, *host, i)
        
        wg.Wait()

    

    if len(openPorts.ports) > 0 
        fmt.Printf("Following ports are opened: %s\n", strings.Join(openPorts.ports, ", "))
     else 
        fmt.Printf("No open port on the host: %s!\n", *host)
    

有两种启动扫描的方法，一种是使用缓冲通道，另一种是使用 sync.GroupWait 并在所有扫描完成后退出。

在我看来，在这种情况下，使用 sync.GroupWait 比使用缓冲通道并循环遍历它直到它为空更有意义。但是，在这里使用缓冲通道，我看不到一种方法来检测通道上没有其他内容，我应该从 for 循环中退出，除非使用另一个 sync.WaitGroup 块。

我想我的问题是，如果我只想使用缓冲通道解决方案，我该如何正确实现它，以便我知道处理何时完成，以便我可以继续执行其余代码？ （请不要建议超时）。

这也是这两种类型的小基准，以防有人感兴趣：

MacBook-Pro:PortScanner c$ time ./PortScanner -host yahoo.com -type 1
Scanning: yahoo.com...
Following ports are opened: 80(http), 143(imap), 110(pop3), 995(pop3s), 993(imaps)

real    0m4.620s
user    0m1.193s
sys     0m1.284s
MacBook-Pro:PortScanner c$ time ./PortScanner -host yahoo.com -type 2
Scanning: yahoo.com...
Following ports are opened: 110(pop3), 80(http), 143(imap), 995(pop3s), 993(imaps)

real    0m4.055s
user    0m1.051s
sys     0m0.946s

【问题讨论】：

您的渠道解决方案不起作用怎么办？如果通道中没有更多项目并关闭，范围循环将退出。

@JimB 如果我不在那里使用sync.WaitGroup，它会因为某种原因挂起......

@JimB - 我的意思是来自 processWithChannels 函数的 WaitGroup。如果我删除它，那么它会挂起。或者您的意思是这是正确的方法，因为无法知道所有这些过程何时完成并且通道何时关闭？

【参考方案1】：

如果您需要将超过 1000 个项目放入频道，则对 processWithChannels 的调用将挂起。如果您要使用缓冲通道来保存所有值直到处理，则必须有足够的容量来接受所有值。

如果您要将所有值收集到单个切片中，则没有理由使用通道，您的第二个解决方案就可以了。

如果您想尽快将端口“流式传输”回来，那么您需要介于两种解决方案之间的一些东西

ports := make(chan string)

var wg sync.WaitGroup
for i := 1; i <= 65535; i++ 
    wg.Add(1)
    go func(i int) 
        defer wg.Done()
        if port := worker(*host, i); port != "" 
            ports <- port
        
    (i)


go func() 
    wg.Wait()
    close(ports)
()

for port := range ports 
    fmt.Println("PORT:", port)

但是，这可能会遇到问题，例如当您同时拨打所有 65535 端口时缺少开放端口。以下是使用一组工作人员同时拨号的一种可能模式：

ports := make(chan string)
toScan := make(chan int)
var wg sync.WaitGroup

// make 100 workers for dialing
for i := 0; i < 100; i++ 
    wg.Add(1)
    go func() 
        defer wg.Done()
        for p := range toScan 
            ports <- worker(*host, p)
        
    ()


// close our receiving ports channel once all workers are done
go func() 
    wg.Wait()
    close(ports)
()

// feed the ports to the worker pool
go func() 
    for i := 1; i <= 65535; i++ 
        toScan <- i
    
    // signal the workers to stop
    close(toScan)
()

for port := range ports 
    if port != "" 
        fmt.Println("PORT:", port)
    

【讨论】：

谢谢。我想到在这里使用缓冲通道的唯一原因是，我认为它会将 1000 个项目发送到通道中，然后通道将阻塞，直到其中的项目被处理为新的项目腾出空间，直到没有其他东西剩下，这样，减慢了速度并且不立即发出 65535 请求。这有意义吗？

@Twisted1919：我明白你的意思，但这不起作用，因为我引用的原因，以及你打电话给worker的事实在你发送在频道上。如果您想限制并发拨号的数量，请使用一组 goroutines（无论如何这会更有效）。我举个例子。

我明白了，好吧，在这种情况下，我认为这个例子会有很大帮助:)

派生出 100 名工人来完成实际工作的部分让世界变得与众不同。感谢您抽出宝贵时间。