Go : 在 For 循环中取消上下文

Question

我正在尝试在 Golang 中创建一个 UDP 服务器来监听一个端口，例如。1234. 我有一个客户端向这个服务器发送启动/停止消息。收到消息“start”后，服务器将开始向该客户端发送随机数据，在停止时，服务器将停止向客户端发送。为此，我使用上下文来创建一个 goroutine 来发送数据并在它“停止”时取消它。我得到的错误是该程序适用于一个客户端，但如果我再次启动客户端，则不会再次发送数据。任何帮助，将不胜感激？UDP服务器代码：package mainimport (    "context"    "fmt"    "math/rand"    "net"    "time")func generateMessageToUDP(ctx context.Context, addr *net.UDPAddr) {    // stop writing to UDP    done := false    fmt.Println("Generating message to UDP client", addr)    conn, err := net.DialUDP("udp", nil, addr)    if err != nil {        fmt.Println("Error: ", err)    }    defer func(conn *net.UDPConn) {        err := conn.Close()        if err != nil {            fmt.Println("Error in closing the UDP Connection: ", err)        }    }(conn)    // write to address using UDP connection    go func() {        for i := 0; !done; i++ {            RandomInt := rand.Intn(100)            fmt.Println("Random Int: ", RandomInt)            _, err = conn.Write([]byte(fmt.Sprintf("%d", RandomInt)))            fmt.Println("Sent ", RandomInt, " to ", addr)            time.Sleep(time.Second * 1)        }    }()    <-ctx.Done()    fmt.Println("Stopping writing to UDP client", addr)    done = true}//var addr *net.UDPAddr//var conn *net.UDPConnfunc main() {    fmt.Println("Hi this is a UDP server")    udpServer, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.IPv4(0, 0, 0, 0), Port: 5010})    if err != nil {        fmt.Println("Error: ", err)    }    defer func(udpServer *net.UDPConn) {        err := udpServer.Close()        if err != nil {            fmt.Println("Error in closing the UDP Connection: ", err)        }    }

Answer 1

你必须小心你正在做的事情。

• 避免数据竞争（完成变量由两个不同的例程读取/写入，没有同步机制）https://go.dev/doc/articles/race_detector

• 每次程序开始向新客户端发送消息时，不要制作新的拨号器。这将打开一个新的本地地址并使用它将其发送给客户端。客户端将从另一个地址接收消息，通常应该忽略这些消息，因为它没有启动与该远程的任何交换。

• 不要将客户端生命周期与程序上下文生命周期混淆。在提供的代码中，客户端发送停止消息将触发整个程序的取消功能，它将停止所有客户端。为每个客户端创建一个新的上下文，派生自程序上下文，在收到停止消息时取消相关的客户端上下文。

• UDP conns 由所有客户端共享，不能因为程序正在为客户端提供服务而停止侦听传入的数据包。IE 调用generateMessageToUDP应该在另一个例程中执行。

以下是针对这些评论的修订版。

添加Avar peers map[string]peer以将远程地址与上下文匹配。类型peer定义为struct {stop  func();since time.Time}。在接收到开始消息后，将与派生上下文一起peer添加到 中。然后，新客户端将在不同的例程中提供服务，该例程绑定到新创建的上下文和服务器套接字。收到停止消息后，程序执行查找，然后取消关联的对等上下文并忘记对等。mappctx, pcancel := context.WithCancel(ctx)go generateMessageToUDP(pctx, udpServer, addr)peer, ok := peers[addr.String()]peer.stop(); delete(peers, addr.String())

package main

import (

    "context"

    "fmt"

    "math/rand"

    "net"

    "time"

)

func generateMessageToUDP(ctx context.Context, conn *net.UDPConn, addr *net.UDPAddr) {

    fmt.Println("Generating message to UDP client", addr)

    go func() {

        for i := 0; ; i++ {

            RandomInt := rand.Intn(100)

            d := []byte(fmt.Sprintf("%d", RandomInt))

            conn.WriteTo(d, addr)

            time.Sleep(time.Second * 1)

        }

    }()

    <-ctx.Done()

    fmt.Println("Stopping writing to UDP client", addr)

}

//var addr *net.UDPAddr

//var conn *net.UDPConn

func main() {

    fmt.Println("Hi this is a UDP server")

    udpServer, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.IPv4(0, 0, 0, 0), Port: 5010})

    if err != nil {

        fmt.Println("Error: ", err)

    }

    defer func(udpServer *net.UDPConn) {

        err := udpServer.Close()

        if err != nil {

            fmt.Println("Error in closing the UDP Connection: ", err)

        }

    }(udpServer)

    // create a buffer to read data into

    type peer struct {

        stop  func()

        since time.Time

    }

    peers := map[string]peer{}

    buffer := make([]byte, 1024)

    ctx, cancel := context.WithCancel(context.Background())

    defer cancel()

    for {

        // read the incoming connection into the buffer

        n, addr, err := udpServer.ReadFromUDP(buffer)

        if err != nil {

            fmt.Println("Error: ", err)

        }

        fmt.Println("Received ", string(buffer[0:n]), " from ", addr)

        if string(buffer[0:n]) == "stop" {

            fmt.Println("Stopped listening")

            peer, ok := peers[addr.String()]

            if !ok {

                continue

            }

            peer.stop()

            delete(peers, addr.String())

            continue

        } else if string(buffer[0:n]) == "start" {

            peer, ok := peers[addr.String()]

            if ok {

                continue

            }

            pctx, pcancel := context.WithCancel(ctx)

            peer.stop = pcancel

            peer.since = time.Now()

            peers[addr.String()] = peer

            // send a response back to the client

            _, err = udpServer.WriteToUDP([]byte("Hi, I am a UDP server"), addr)

            if err != nil {

                fmt.Println("Error: ", err)

            }

            // start a routine to generate messages to the client

            go generateMessageToUDP(pctx, udpServer, addr)

        } else if string(buffer[0:n]) == "ping" {

            peer, ok := peers[addr.String()]

            if !ok {

                continue

            }

            peer.since = time.Now()

            peers[addr.String()] = peer

        } else {

            fmt.Println("Unknown command")

        }

        for addr, p := range peers {

            if time.Since(p.since) > time.Minute {

                fmt.Println("Peer timedout")

                p.stop()

                delete(peers, addr)

            }

        }

    }

}

-- go.mod --

module play.ground

-- client.go --

package main

import (

    "fmt"

    "log"

    "net"

    "time"

)

func main() {

    fmt.Println("Hello, I am a client")

    // Create a new client

    localAddr, err := net.ResolveUDPAddr("udp", "127.0.0.1:5011")

    client3, err := net.DialUDP("udp", localAddr, &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1), Port: 5010})

    if err != nil {

        fmt.Println(err)

        return

    }

    defer client3.Close()

    var n int

    n, err = client3.Write([]byte("start"))

    if err != nil {

        fmt.Println(err)

        return

    }

    log.Println(n)

    now := time.Now()

    b := make([]byte, 2048)

    for time.Since(now) < time.Second*10 {

        n, addr, err := client3.ReadFrom(b)

        fmt.Println(n, addr, err)

        if err != nil {

            fmt.Println(err)

            continue

        }

        if addr.String() == "127.0.0.1:5010" {

            m := b[:n]

            fmt.Println("message:", string(m))

        }

    }

    fmt.Println("Sending stop message")

    _, err = client3.Write([]byte("stop"))

    if err != nil {

        fmt.Println(err)

    }

}

在

    go func() {

        for i := 0; ; i++ {

            RandomInt := rand.Intn(100)

            d := []byte(fmt.Sprintf("%d", RandomInt))

            conn.WriteTo(d, addr)

            time.Sleep(time.Second * 1)

        }

    }()

我将在上下文通道上写入缺失的选择作为练习留给读者，以确定例程是否应该退出。

Answer 2

好的，我在服务器上做了一个简单的修改，并在创建上下文之前添加了一个标签 Start，当我取消上下文时，我添加了 goto 标签。这意味着当任务被取消时，它将再次创建上下文并开始执行其工作