go - 如何同步常量写入和周期性读取和更新

Question

定义问题:

我们有这个物联网设备，每个设备都会向我们发送有关汽车位置的日志。我们要计算汽车在线行驶的距离!因此，当日志出现时(将其放入队列等之后)，我们会这样做:

type Delta struct {
    DeviceId string
    time     int64
    Distance float64
}
var LastLogs = make(map[string]FullLog)
var Distances = make(map[string]Delta)


func addLastLog(l FullLog) {
    LastLogs[l.DeviceID] = l
}
func AddToLogPerDay(l FullLog) {
    //mutex.Lock()
    if val, ok := LastLogs[l.DeviceID]; ok {
        if distance, exist := Distances[l.DeviceID]; exist {
            x := computingDistance(val, l)
            Distances[l.DeviceID] = Delta{
                DeviceId: l.DeviceID,
                time:     distance.time + 1,
                Distance: distance.Distance + x,
            }
        } else {
            Distances[l.DeviceID] = Delta{
                DeviceId: l.DeviceID,
                time:     1,
                Distance: 0,
            }
        }
    }
    addLastLog(l)

}

它基本上使用效用函数计算距离!所以在 Distances每个设备 ID 都映射到一定距离!现在这里是问题开始的地方:虽然这个距离被添加到 Distances map ，我想要一个 go 例程将这些数据放入数据库，但由于有很多设备和许多日志等，因此对每个日志进行此查询并不是一个好主意。所以我需要每 5 秒执行一次，这意味着每 5 秒尝试清空添加到 map 的所有最后距离的列表。我写了这个函数:

func UpdateLogPerDayTable() {
    for {
        for _, distance := range Distances {
            logs := model.HourPerDay{}
            result := services.CarDBProvider.DB.Table(model.HourPerDay{}.TableName()).
                Where("created_at >?  AND device_id = ?", getCurrentData(), distance.DeviceId).
                Find(&logs)
            if result.Error != nil && !result.RecordNotFound() {
                log.Infof("Something went wrong while checking the log: %v", result.Error)
            } else {
                if !result.RecordNotFound() {
                    logs.CountDistance = distance.Distance

                    logs.CountSecond = distance.time

                    err := services.CarDBProvider.DB.Model(&logs).
                        Update(map[string]interface{}{
                            "count_second":   logs.CountSecond,
                            "count_distance": logs.CountDistance,
                        })
                    if err.Error != nil {
                        log.Infof("Something went wrong while updating the log: %v", err.Error)
                    }

                } else if result.RecordNotFound() {
                    dayLog := model.HourPerDay{
                        Model:         gorm.Model{},
                        DeviceId:      distance.DeviceId,
                        CountSecond:   int64(distance.time),
                        CountDistance: distance.Distance,
                    }
                    err := services.CarDBProvider.DB.Create(&dayLog)
                    if err.Error != nil {
                        log.Infof("Something went wrong while adding the log: %v", err.Error)
                    }
                }
            }
        }
        time.Sleep(time.Second * 5)
    }
}

它被称为 go utlis.UpdateLogPerDayTable()在另一个去例行公事上。但是这里有很多问题:

我不知道如何保护 Distances所以当我在另一个例程中添加它时，我在其他地方阅读它，一切都很好!(问题是我想使用 go channel 并且不知道该怎么做)

对于这个问题，我如何安排任务？

可能我会添加一个 redis 来存储所有在线设备，这样我就可以更快地进行选择查询并更新实际的数据库。还为 redis 添加过期时间，因此如果设备在一段时间内没有发送和数据，它就会消失!我应该把这段代码放在哪里？

对不起，如果我的解释还不够，但我真的需要一些帮助。专门用于代码实现

Answer 1

Go 有一个非常酷的模式，使用 for / select通过多个 channel 。这允许您使用超时和最大记录大小来批量写入距离。使用这种模式需要使用 channel 。

首先是将您的距离建模为 channel :

distances := make(chan Delta)

然后你跟踪当前批处理

var deltas []Delta

然后

ticker := time.NewTicker(time.Second * 5)

var deltas []Delta

for {
  select {
     case <-ticker.C:
        // 5 seconds up flush to db
        // reset deltas
     case d := <-distances:
        deltas = append(deltas, d)
        if len(deltas) >= maxDeltasPerFlush {
           // flush
           // reset deltas
        }
  }
}

I don't know how to secure Distances so when I add it in another routine I read it somewhere else ,every thing is ok!(The problem is that I want to use go channels and don't have any idea how to do it)

如果您打算保留 map 并共享内存，您需要使用 mutual exclusion (mutex) 来保护它。在 go 例程之间同步访问。使用 channel允许您将副本发送到 channel ，从而无需跨 Delta 对象进行同步。根据您的架构，您还可以创建一个由 channel 连接的 go 例程管道，这可以使它只有一个 go 例程 ( monitor go routine ) 正在访问 Delta ，也消除了同步的需要。

How can I schedule tasks in go for this problem?

使用 channel 作为传递方式的原语 Deltas去不同的套路:)

Probably I will add a redis to store all the devices that or online so I could do the select query faster and just update the actual database. also add an expire time for redis so if a device didn't send and data for some time, it vanishes! where should I put this code?

这取决于您完成的架构。你可以写一个 decorator对于选择操作，它将首先检查 redis，然后转到数据库。此功能的客户端不必知道这一点。写操作可以以相同的方式完成:写入持久存储，然后使用缓存值和过期时间写回 redis。使用装饰器，客户端不需要知道这一点，他们只需执行读取和写入，缓存逻辑将在装饰器内部实现。有很多方法可以做到这一点，这在很大程度上取决于您的实现在哪里解决。