我正在緩存系統(tǒng)的數(shù)據(jù)庫，我遇到了這個問題：我們有一些遠(yuǎn)程設(shè)備每秒向服務(wù)器發(fā)送日志！由于我們有很多這樣的設(shè)備，我們不能只將所有數(shù)據(jù)存儲在一個普通的數(shù)據(jù)庫表中。我們實際上需要快速訪問每個設(shè)備發(fā)送的最后 5 分鐘的日志。日志在 json 中，但我們將它們映射到 Go 結(jié)構(gòu)。每個日志都有一些數(shù)據(jù)，例如： type Log struct {
DeviceID string
CompanyID uint
DeviceTime time.Time
Latitude float64
Longitude float64
Altitude float64
SpeedOTG float32}我們正在使用redis將數(shù)據(jù)存儲在 ram 上。我的問題：任務(wù)是將每個設(shè)備最近 5 分鐘的所有日志存儲在一個列表中。我也應(yīng)該在提供緩存鍵時檢索這些數(shù)據(jù)。我必須做一些事情，比如隊列。當(dāng)它滿了（它包含最后 5 分鐘的日志）時，每個新條目都會進(jìn)入，舊條目應(yīng)該出去。但是，如果列表仍然不包含最后 5 分鐘，那么什么都不會發(fā)生！所有計算都是基于上面定義的DeviceTime 字段完成的。起初我考慮了一個具有確定大?。? * 60）日志的列表來存儲，但沒有被授權(quán)者每秒都有一個新日志！
1 回答
慕尼黑8549860
TA貢獻(xiàn)1818條經(jīng)驗 獲得超11個贊
將您的日志存儲在Redis Streams中。要插入，請使用XADD命令。[轉(zhuǎn)到 Redis 示例]。
您可以將所有設(shè)備存儲在一個流中，或者每個設(shè)備使用一個流。
在這里，假設(shè)每個設(shè)備一個流并保留一組設(shè)備 ID，您將插入：
SADD devices myDeviceID
XADD logs:myDeviceID * DeviceID myDeviceID CompanyID myCompanyID DeviceTime "myDeviceTime"
... SpeedOTG "mySpeedOTG"
該設(shè)置devices用于保留設(shè)備 ID 列表，以防您想要檢索所有設(shè)備的數(shù)據(jù)并且您不想依賴保持它在運行（客戶端）或使用 SCAN。要檢索所有設(shè)備日志，您將使用類似于我在此 RedisCache 實現(xiàn)中使用哪種數(shù)據(jù)類型中描述的模式？, Lua 腳本示例在最后。
如果字段壓縮，流的好處之一：當(dāng)您保持字段一致（相同的字段以相同的順序）時，Redis 通過不在每個條目上重復(fù)字段名稱來進(jìn)行優(yōu)化。
通過使用*，您可以讓 Redis 將時間戳設(shè)置為條目 ID。只要您確定它會嚴(yán)格增加，您就可以使用自己的時間戳作為 ID。您不能添加到 ID 低于最后添加的 ID 的流中。
您可以使用XTRIM將列表減少到 300 個條目，但正如您所說，您不能依賴每秒一致的 XADD。
要將其精確到最后 5 分鐘，您可以使用：
XRANGE logs:myDeviceID - + COUNT 1
這將檢索第一個（最舊的）條目。您將其與新添加的最新條目的時間戳進(jìn)行比較，如果超過 5 分鐘，則將其刪除XDEL。
您重復(fù)直到最早的條目不超過 5 分鐘。
當(dāng)然，做這個邏輯客戶端意味著很多往返，影響你的表現(xiàn)。所以我建議你使用Lua 腳本來做 Redis-server-side。您可以每次使用 EVAL 傳遞腳本。但最好是加載一次腳本然后使用EVALSHA。
劇本：
redis.call('SADD', KEYS[1], ARGV[2])
local latestID = redis.call('XADD', 'logs:'..ARGV[2], '*', 'DeviceID', ARGV[2],
'OtherFields', ARGV[3])
local latestTime = tonumber(string.sub(latestID, 1, string.find(latestID, '-') - 1))
local oldestID = redis.call('XRANGE', 'logs:'..ARGV[2], '-', '+', 'COUNT', '1')
oldestID = oldestID[1][1]
local oldestTime = tonumber(string.sub(oldestID, 1, string.find(oldestID, '-') - 1))
local maxTime = tonumber(ARGV[1])
while (latestTime - oldestTime) > maxTime do
redis.call('XDEL', 'logs:'..ARGV[2], oldestID)
oldestID = redis.call('XRANGE', 'logs:'..ARGV[2], '-', '+', 'COUNT', '1')
oldestID = oldestID[1][1]
oldestTime = tonumber(string.sub(oldestID, 1, string.find(oldestID, '-') - 1))
end
return { 'Added: '..latestID, 'Device: '..ARGV[2], 'Stream Key: logs:'..ARGV[2],
'Length: '..redis.call('XLEN', 'logs:'..ARGV[2]) }
參數(shù)：
EVALSHA <shaId> 1 devices 300000 myDeviceID ...
^ARGV[3...]: the rest of fields
^ARGV[2]: DeviceID
^ARGV[1]: logging time desired (ms), 5 minutes = 300000
^KEYS[1]: the set key and prefix to device log keys
^numkeys
使用流，您可以獲得很多功能，例如按時間戳劃分的范圍或消費者組。
但最重要的是，為了讓您的日志不受時間限制，請使用 Lua 腳本。您也可以使用列表來執(zhí)行此操作，將編碼的日志條目存儲在一個字符串中：timestamp:valuesSerialized并使用與 Lua 類似的方法在一個原子操作中添加檢查修剪。
- 1 回答
- 0 關(guān)注
- 108 瀏覽
添加回答
舉報
0/150
提交
取消