如何实现即时写入和定期读取的数据同步

从现在开始，我们要努力学习啦！今天我给大家带来《如何实现即时写入和定期读取的数据同步》，感兴趣的朋友请继续看下去吧！下文中的内容我们主要会涉及到等等知识点，如果在阅读本文过程中有遇到不清楚的地方，欢迎留言呀！我们一起讨论，一起学习！

定义问题：

我们有这个物联网设备，每个设备都会向我们发送有关汽车位置的日志。我们想要计算汽车在线行驶的距离！因此，当日志到来时（将其放入队列等之后），我们会这样做：

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 秒尝试清空添加到地图的所有最后距离的列表。我写了这个函数：

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 例程中被称为 go utlis.updatelogperdaytable() 。但是这里存在很多问题：

1. 我不知道如何保护 distances，所以当我将它添加到另一个例程中时，我在其他地方阅读它，一切都很好！（问题是我想使用 go 通道，但不知道如何使用）去做）
2. 如何在 go 中安排任务来解决这个问题？
3. 我可能会添加一个 redis 来存储所有设备或在线设备，这样我就可以更快地执行选择查询并更新实际数据库。还为 redis 添加过期时间，因此如果设备一段时间没有发送数据，它就会消失！我应该把这段代码放在哪里？

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

解决方案

go 有一个非常酷的模式，在多个通道上使用 for / select。这允许您使用超时和最大记录大小来批量进行距离写入。使用此模式需要使用通道。

第一件事是将距离建模为通道：

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
        }
  }
}

我不知道如何确保距离，所以当我将其添加到另一个中时 我在其他地方读过它，一切都很好！（问题是 我想使用 go 频道，但不知道该怎么做）

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

如何在 go 中安排任务来解决这个问题？

使用通道作为如何将 deltas 传递给不同 go 例程的原语:)

可能我会添加一个redis来存储所有设备或在线设备 我可以更快地执行选择查询，只需更新实际的 数据库。还为 redis 添加过期时间，以便设备未发送时 和数据一段时间后就消失了！我应该把这段代码放在哪里？

这取决于您完成的架构。你可以为 select 操作写一个 decorator，它会先检查 redis 然后再去数据库。该函数的客户端不必知道这一点。写入操作可以以相同的方式完成：写入持久存储，然后将缓存值和过期时间写回到 redis。使用装饰器，客户端不需要知道这一点，他们只需执行读取和写入，并且缓存逻辑将在装饰器内部实现。有很多方法可以实现这一点，这在很大程度上取决于您的实施方案。

今天关于《如何实现即时写入和定期读取的数据同步》的内容介绍就到此结束，如果有什么疑问或者建议，可以在golang学习网公众号下多多回复交流；文中若有不正之处，也希望回复留言以告知！