tgoop.com/python_job_interview/1207
Create:
Last Update:
Last Update:
👾 Задача из собеседования по Python
Уровень: middle-senior
Условие:
Реализуйте потокобезопасный кэш с TTL и политикой вытеснения LRU. Кэш должен:
1️⃣ Автоматически удалять записи по истечении TTL.
2️⃣ При достижении максимального размера вытеснять реже всего используемые элементы.
3️⃣ Гарантировать корректную работу в многопоточной среде.
4️⃣ Поддерживать методы: get(key), set(key, value, ttl), delete(key), clear().
5️⃣ Опционально: реализовать ленивое удаление просроченных записей.
Решение:
import time
import threading
from collections import OrderedDict
class TTLLRUCache:
def __init__(self, maxsize=1024):
self.maxsize = maxsize
self._cache = OrderedDict()
self._lock = threading.RLock()
self._expiry_times = {}
def get(self, key):
with self._lock:
if key not in self._cache:
return None
# Ленивое удаление просроченного ключа
if self._is_expired(key):
self._delete(key)
return None
# Обновляем порядок использования (LRU)
self._cache.move_to_end(key)
return self._cache[key]
def set(self, key, value, ttl=None):
with self._lock:
# Вытеснение старых записей при достижении лимита
if len(self._cache) >= self.maxsize:
self._evict()
self._cache[key] = value
self._cache.move_to_end(key)
self._expiry_times[key] = time.time() + ttl if ttl else None
def delete(self, key):
with self._lock:
self._delete(key)
def _delete(self, key):
self._cache.pop(key, None)
self._expiry_times.pop(key, None)
def clear(self):
with self._lock:
self._cache.clear()
self._expiry_times.clear()
def _is_expired(self, key):
expiry = self._expiry_times.get(key)
return expiry is not None and expiry < time.time()
def _evict(self):
# Сначала удаляем просроченные ключи
expired_keys = [k for k in self._cache if self._is_expired(k)]
for k in expired_keys:
self._delete(k)
# Если после этого кэш всё ещё полон, применяем LRU
if len(self._cache) >= self.maxsize:
self._cache.popitem(last=False)
def __contains__(self, key):
with self._lock:
if key not in self._cache:
return False
if self._is_expired(key):
self._delete(key)
return False
return True
Пояснение:
1. Потокобезопасность:
Используется
threading.RLock для всех операций, изменяющих состояние кэша. Это позволяет рекурсивные блокировки из одного потока.2. TTL:
Время истечения хранится в отдельном словаре
_expiry_times. При обращении к ключу проверяется его актуальность.3. LRU-политика:
OrderedDict автоматически поддерживает порядок использования элементов. Метод move_to_end() обновляет позицию при каждом обращении, а popitem(last=False) удаляет самый старый элемент.4. Гибкое удаление:
— Ленивое (при обращении к ключу)
— Активное (при добавлении новых элементов через
_evict())5. Оптимизация:
— Сначала удаляются просроченные ключи, и только потом применяется LRU.
— Сложность операций: O(1) для get/set (в среднем случае).
@python_job_interview
BY Python вопросы с собеседований
Share with your friend now:
tgoop.com/python_job_interview/1207