单次Flush任务
(fullKey string)
| 517 | |
| 518 | // 单次Flush任务 |
| 519 | func (this *MemoryStorage) flushItem(fullKey string) { |
| 520 | sizeString, key, found := strings.Cut(fullKey, "@") |
| 521 | if !found { |
| 522 | return |
| 523 | } |
| 524 | defer func() { |
| 525 | atomic.AddInt64(&this.totalDirtySize, -types.Int64(sizeString)) |
| 526 | }() |
| 527 | |
| 528 | if this.parentStorage == nil { |
| 529 | return |
| 530 | } |
| 531 | var hash = this.hash(key) |
| 532 | |
| 533 | this.locker.RLock() |
| 534 | item, ok := this.valuesMap[hash] |
| 535 | this.locker.RUnlock() |
| 536 | |
| 537 | // 从内存中移除，并确保无论如何都会执行 |
| 538 | defer func() { |
| 539 | _ = this.Delete(key) |
| 540 | }() |
| 541 | |
| 542 | if !ok { |
| 543 | return |
| 544 | } |
| 545 | |
| 546 | if !item.IsDone { |
| 547 | remotelogs.Error("CACHE", "flush items failed: open writer failed: item has not been done") |
| 548 | return |
| 549 | } |
| 550 | if item.IsExpired() { |
| 551 | return |
| 552 | } |
| 553 | |
| 554 | // 检查是否在列表中，防止未加入列表时就开始flush |
| 555 | isInList, _, err := this.list.Exist(types.String(hash)) |
| 556 | if err != nil { |
| 557 | remotelogs.Error("CACHE", "flush items failed: "+err.Error()) |
| 558 | return |
| 559 | } |
| 560 | if !isInList { |
| 561 | for i := 0; i < 1000; i++ { |
| 562 | isInList, _, err = this.list.Exist(types.String(hash)) |
| 563 | if err != nil { |
| 564 | remotelogs.Error("CACHE", "flush items failed: "+err.Error()) |
| 565 | time.Sleep(1 * time.Second) |
| 566 | continue |
| 567 | } |
| 568 | if isInList { |
| 569 | break |
| 570 | } |
| 571 | time.Sleep(1 * time.Millisecond) |
| 572 | } |
| 573 | if !isInList { |
| 574 | // discard |
| 575 | return |
| 576 | } |
no test coverage detected