代码之家 › 专栏 › 技术社区 › zemirco

在带取消的UDP消息上启动多个计时器

goroutine udp go

zemirco · 技术社区 · 6 年前

我在听UDP上的消息。我们有这样的装置。他们还说下一次发布公告的时间。如果没有发生这种情况,我们假设一个设备已经不见了。

我想列出当前网络中的设备列表。我想添加新设备并删除那些我没有听说过的设备。

这是我到目前为止得到的。

1)我有一个内存数据库,可以存放所有设备。

func NewDB() *DB {
    return &DB{
        table: make(map[string]Announcement),
    }
}

type DB struct {
    mutex sync.Mutex
    table map[string]Announcement
}

func (db *DB) Set(ip string, ann Announcement) {
    db.mutex.Lock()
    defer db.mutex.Unlock()
    db.table[ip] = ann
}

func (db *DB) Delete(ip string) {
    db.mutex.Lock()
    defer db.mutex.Unlock()
    delete(db.table, ip)
}

func (db *DB) Snapshot() map[string]Announcement {
    db.mutex.Lock()
    defer db.mutex.Unlock()
    return db.table
}

2)我有一个web服务器,它将这个数据库服务于我的JavaScript前端

http.HandleFunc("/json", func(w http.ResponseWriter, r *http.Request) {
    w.Header().Set("Content-Type", "application/json")
    json.NewEncoder(w).Encode(db.Snapshot())
})

// start server
go func() {
    log.Fatal(http.ListenAndServe(":8085", nil))
}()

3)最后我在监听UDP消息。每当一个新设备被添加到数据库中时,我也会用提供的超时值创建一个新的计时器(这里我只是将其设置为10秒)。当一个新的消息到达时,我检查一个现有的计时器,当它存在时停止它,如果它不再发送消息,重新启动它来清除设备。

但是我不是真的工作。这个 AfterFunc 被称为常去之路。尽管设备仍在网络中,但它已从我的 db . 有什么想法吗?

// some global variable
var (
    timers = map[string]*time.Timer{}
)

for {
    // create new buffer
    b := make([]byte, 1500)

    // read message from udp into buffer
    n, src, err := conn.ReadFromUDP(b)
    if err != nil {
        panic(err)
    }

    // convert raw json bytes to struct
    var ann Announcement
    if err := json.Unmarshal(b[:n], &ann); err != nil {
        panic(err)
    }

    // add announcement to db
    ip := src.IP.String()
    db.Set(ip, ann)

    // check for existing timer
    timer, ok := timers[ip]
    if ok {
        log.Println("stopping timer", ip)
        // stop existing timer
        timer.Stop()
    }

    // start new timer for device
    timer = time.AfterFunc(time.Second*10, func() {
        log.Println("time after func", ip)
        delete(timers, ip)
        db.Delete(ip)
    })

    // store timer in timers db
    timers[ip] = timer

    time.Sleep(250 * time.Millisecond)
}

1 回复 | 直到 6 年前

MichalG 6 年前

我认为你的问题可能与 ip 您捕获的变量 AfterFunc func .

timer = time.AfterFunc(time.Second*10, func() {
        log.Println("time after func", ip)
        delete(timers, ip)
        db.Delete(ip)
    })

从该代码中删除 知识产权 将使用的值调用 知识产权 变量目前这个计时器过期了。因此,如果同时您从另一个具有不同IP的设备接收到数据包,则此数据包将被删除。

正在发生的事情的示例:

第二个1 :IP为1.2.3.4的设备发送UDP数据包。 ip = 1.2.3.4 , 后函数 调用,10秒计时器启动
第二个3 :IP为4.5.6.7的设备发送UDP数据包。现在 ip = 4.5.6.7 , 后函数 调用,10秒计时器启动
第二个10 :删除当前值的函数 知识产权 变量被调用,设备 4.5.6.7 已删除
第二个13 :第二个计时器超时,我们尝试删除 4.5.6.7节 再一次

因此,IP设备 1.2.3.4 永远不会被删除。

可以通过创建一个函数来修复此问题,该函数接受一个参数并返回当前值为参数的func()。

timer = time.AfterFunc(time.Second*10, func(ip string) func() {
    return func() {
        log.Println("time after func", ip)
        delete(timers, ip)
        db.Delete(ip)
    }   
}(ip))

更简单的工作示例:

package main

import (
    "fmt"
    "time"
)

func main() {
    fmt.Println("Capturing value of i at the moment of execution of func()")
    for i := 0; i < 5; i++ {

        afterFuncTimer := time.AfterFunc(time.Second*2, func() {
            fmt.Printf("AfterFunc() with %v\n", i)
        })

        defer afterFuncTimer.Stop()
    }

    time.Sleep(5 * time.Second)

    fmt.Println("Capturing value of i from the loop")
    for i := 0; i < 5; i++ {

        afterFuncTimer := time.AfterFunc(time.Second*2, func(i int) func() {
            return func() {
                fmt.Printf("AfterFunc() with %v\n", i)
            }
        }(i))

        defer afterFuncTimer.Stop()
    }

    time.Sleep(5 * time.Second)

}

在游乐场上跑步: https://play.golang.org/p/bGWzTaWe3ZU