Python 인메모리 캐시(수명 연장)
나는 동일한 프로세스를 실행하는 여러 스레드를 가지고 있는데, 이 스레드들은 앞으로 n초 동안 무언가가 작업되어서는 안 된다는 것을 서로에게 알릴 수 있어야 합니다. 하지만 작업을 수행한다면 세상이 끝나는 것이 아닙니다.
제 목표는 문자열과 TTL을 캐시에 전달하고 캐시에 있는 모든 문자열을 목록으로 가져올 수 있는 것입니다.캐시는 메모리에 저장될 수 있으며 TTL은 20초 이내가 될 것입니다.
이것이 어떻게 이루어질 수 있는지에 대한 제안이 있는 사람이 있습니까?
라이브러리를 에 3 하지 에 를 할 의 할 의 를 를 ttl_hash=None 변수는 " hash"라고 은 영향을 주는 입니다. 는 "에 한 " 라고 입니다 " 한 은 " 을 의 입니다 을 은 한 한 라고 의 lru_cache.
예를 들어,
from functools import lru_cache
import time
@lru_cache()
def my_expensive_function(a, b, ttl_hash=None):
del ttl_hash # to emphasize we don't use it and to shut pylint up
return a + b # horrible CPU load...
def get_ttl_hash(seconds=3600):
"""Return the same value withing `seconds` time period"""
return round(time.time() / seconds)
# somewhere in your code...
res = my_expensive_function(2, 2, ttl_hash=get_ttl_hash())
# cache will be updated once in an hour
2 만약 당신이 3, 2.7을 사용하고 있다면, python 3을 사용하고 있다면,ExpiringDict승인된 답변에 언급된 것은 현재 만료된 상태입니다.github repo에 대한 마지막 커밋은 2017년 6월 17일이었고 Python 3.5와 함께 작동하지 않는 공개된 문제가 있습니다.
2020년 9월 1일 현재, 더 최근에 유지된 프로젝트 캐시 툴이 있습니다.
pip install cachetools
from cachetools import TTLCache
cache = TTLCache(maxsize=10, ttl=360)
cache['apple'] = 'top dog'
...
>>> cache['apple']
'top dog'
... after 360 seconds...
>>> cache['apple']
KeyError exception raised
ttl초 단위로 살 수 있는 시간입니다.
만료되는 인메모리 캐시와 관련하여, 일반적으로 이를 수행하는 일반적인 설계 패턴은 사전이 아니라 함수 또는 방법 데코레이터를 통해 수행됩니다.캐시 사전은 백그라운드에서 관리됩니다.이와 같이 데코레이터가 아닌 사전을 사용하는 사용자의 답변을 어느 정도 보완합니다.
장식가는 비슷한 일을 많이 하지만 살 수 있는 시간을 가지고 일을 합니다.
import cachetools.func
@cachetools.func.ttl_cache(maxsize=128, ttl=10 * 60)
def example_function(key):
return get_expensively_computed_value(key)
class ExampleClass:
EXP = 2
@classmethod
@cachetools.func.ttl_cache()
def example_classmethod(cls, i):
return i * cls.EXP
@staticmethod
@cachetools.func.ttl_cache()
def example_staticmethod(i):
return i * 3
모듈을 사용할 수 있습니다.
은 입니다.
ExpiringDict클래스(class)는 캐싱 목적으로 자동 만료 값을 가진 정렬된 사전입니다.
대해 에서 은 하지 를 합니다 에 a 를 합니다 에서 은 에 Lock.
좀 것은 , 관계가 , 된 Δ Δ Δ Δ Δ Δ Δ Δ Δ Δ Δ Δ Δ Δ Δ Δ Δ Δ Δ Δ Δ Δ Δ Δ Δ Δ Δ Δ Δ Δ Δ Δ Δ Δ Δ Δ Δ Δ Δ Δ Δ Δ Δ Δ Δ Δ Δ Δ Δ Δ Δ Δ Δ Δ Δ Δfunctools.lru_cache(이 글을 쓴 후 하비에르의 비슷한 대답을 알아챘지만, 이것은 장고가 필요하지 않기 때문에 어쨌든 글을 올렸다고 생각했습니다.)
import functools
import time
def time_cache(max_age, maxsize=128, typed=False):
"""Least-recently-used cache decorator with time-based cache invalidation.
Args:
max_age: Time to live for cached results (in seconds).
maxsize: Maximum cache size (see `functools.lru_cache`).
typed: Cache on distinct input types (see `functools.lru_cache`).
"""
def _decorator(fn):
@functools.lru_cache(maxsize=maxsize, typed=typed)
def _new(*args, __time_salt, **kwargs):
return fn(*args, **kwargs)
@functools.wraps(fn)
def _wrapped(*args, **kwargs):
return _new(*args, **kwargs, __time_salt=int(time.time() / max_age))
return _wrapped
return _decorator
그리고 그 용도:
@time_cache(10)
def expensive(a: int):
"""An expensive function."""
time.sleep(1 + a)
print("Starting...")
expensive(1)
print("Again...")
expensive(1)
print("Done")
은 NB 을 합니다.time.time그리고 모든 주의사항들이 함께 나옵니다.사용할 수 있습니다.time.monotonic대신에 가능한 경우/적절한 경우
저는 @iutinvg의 아이디어가 정말 마음에 듭니다. 단지 조금 더 나아가고 싶었습니다. 통과하기 위해 알아야 하는 것과 분리합니다.ttl모든 기능을 다 사용할 수 있게 하고 장식품으로 만들어서 생각할 필요가 없어요.이 ㅇdjango,py3 어떤 의존성도 하고 요 해 도 해 요 도 하고 .
import time
from django.utils.functional import lazy
from functools import lru_cache, partial, update_wrapper
def lru_cache_time(seconds, maxsize=None):
"""
Adds time aware caching to lru_cache
"""
def wrapper(func):
# Lazy function that makes sure the lru_cache() invalidate after X secs
ttl_hash = lazy(lambda: round(time.time() / seconds), int)()
@lru_cache(maxsize)
def time_aware(__ttl, *args, **kwargs):
"""
Main wrapper, note that the first argument ttl is not passed down.
This is because no function should bother to know this that
this is here.
"""
def wrapping(*args, **kwargs):
return func(*args, **kwargs)
return wrapping(*args, **kwargs)
return update_wrapper(partial(time_aware, ttl_hash), func)
return wrapper
를 .maxsize이다 ㅇNone, 하지만 당신은 정말로 여기에 그렇지 않은 값을 추가해야 합니다.None. 이로 인해 메모리 누수가 발생합니다(python 인스턴스는 프로그램이 사라질 때까지 메모리를 제거하지 않고 영원히 메모리를 사용합니다).
효과가 있음을 증명합니다(예시 적용).
@lru_cache_time(seconds=10, maxsize=128)
def meaning_of_life():
"""
This message should show up if you call help().
"""
print('this better only show up once!')
return 42
@lru_cache_time(seconds=10, maxsize=128)
def multiply(a, b):
"""
This message should show up if you call help().
"""
print('this better only show up once!')
return a * b
# This is a test, prints a `.` for every second, there should be 10s
# between each "this better only show up once!" *2 because of the two functions.
for _ in range(20):
meaning_of_life()
multiply(50, 99991)
print('.')
time.sleep(1)
또 하나의 해결책
어떻게 돼가요?
- 은 을 하여 은 을 사용하여 캐시됩니다.
@functools.lru_cache여의maxsize그리고.typed매개 변수들. Result를 사용하여 함수의 반환 합니다.time.monotonic() + ttl.- 을 Δ"" Δ"합니다" Δ""에 대해 확인합니다.
time.monotonic()현재 시간이 "사망" 시간을 초과하는 경우, 새로운 "사망" 시간으로 반환 값을 다시 계산합니다.
코드를 보여주십시오.
from functools import lru_cache, wraps
from time import monotonic
def lru_cache_with_ttl(maxsize=128, typed=False, ttl=60):
"""Least-recently used cache with time-to-live (ttl) limit."""
class Result:
__slots__ = ('value', 'death')
def __init__(self, value, death):
self.value = value
self.death = death
def decorator(func):
@lru_cache(maxsize=maxsize, typed=typed)
def cached_func(*args, **kwargs):
value = func(*args, **kwargs)
death = monotonic() + ttl
return Result(value, death)
@wraps(func)
def wrapper(*args, **kwargs):
result = cached_func(*args, **kwargs)
if result.death < monotonic():
result.value = func(*args, **kwargs)
result.death = monotonic() + ttl
return result.value
wrapper.cache_clear = cached_func.cache_clear
return wrapper
return decorator
사용법은?
# Recalculate cached results after 5 seconds.
@lru_cache_with_ttl(ttl=5)
def expensive_function(a, b):
return a + b
혜택들
- 짧고 검토하기 쉬우며 PyPI를 설치할 필요가 없습니다.Python 표준 라이브러리인 3.7+에만 의존합니다.
- 귀찮지 않습니다
ttl=10모든 호출 사이트에 필요한 매개 변수입니다. - 모든 항목을 동시에 제거하지 않습니다.
- 키/값 쌍은 실제로 지정된 TTL 값에 대해 살아 있습니다.
- 고유한 의 만/ 하나당 키/값 쌍 합니다.
(*args, **kwargs)항목이 만료된 경우에도. - 장식가로 활동합니다(하비에르 버지 답변과 루이스 벨처 답변에 쿠도스).
- 실이 안전한가요?
- python.org 에서 제공하는 CPython의 C-최적화를 통해 얻을 수 있는 이점과 PyPy와 호환됩니다.
합격한 답은 #2, #3, #4, #5, #6에 실패합니다.
단점
만료된 항목을 사전에 제거하지 않습니다.만료된 항목은 캐시가 최대 크기에 도달한 경우에만 제거됩니다.크기에 is대지할우예에가(우예할xse x: maxsize isNone가 발생하지 .), 가 하지 하지 가
고유한 Δ Δ Δ Δ Δ Δ Δ Δ Δ Δ Δ Δ Δ Δ Δ Δ Δ Δ Δ Δ Δ Δ Δ Δ Δ ΔΔ Δ Δ Δ Δ(*args, **kwargs)캐시된 함수에 주어집니다.따라서 만약 단지 10개의 다른 파라미터 조합이 있다면, 캐시는 최대 10개의 항목만 가질 것입니다.
"시간에 민감한 해시" 및 "시간 염" 솔루션은 동일한 키(다른 시간 해시/염)를 가진 여러 개의 키/값 캐시 항목이 캐시에 남아 있기 때문에 훨씬 더 좋지 않습니다.
타사 패키지를 사용하지 않으려면 사용자 지정 timed_lru_cache 데코레이터를 추가할 수 있습니다. 이 장식은 lru_cache 데코레이터를 기반으로 합니다.
다음은 20초 수명과 최대 크기 128입니다.개별 항목이 아닌 전체 캐시는 20초 후에 만료됩니다.
from datetime import datetime, timedelta
from functools import lru_cache, wraps
def timed_lru_cache(seconds: int = 20, maxsize: int = 128):
def wrapper_cache(func):
func = lru_cache(maxsize=maxsize)(func)
func.lifetime = timedelta(seconds=seconds)
func.expiration = datetime.utcnow() + func.lifetime
@wraps(func)
def wrapped_func(*args, **kwargs):
if datetime.utcnow() >= func.expiration:
func.cache_clear()
func.expiration = datetime.utcnow() + func.lifetime
return func(*args, **kwargs)
return wrapped_func
return wrapper_cache
그럼 그냥 추가해주세요.@timed_lru_cache()당신의 기능 이상이면 당신은 가도 좋을 것입니다.
@timed_lru_cache()
def my_function():
# code goes here...
그런 거?
from time import time, sleep
import itertools
from threading import Thread, RLock
import signal
class CacheEntry():
def __init__(self, string, ttl=20):
self.string = string
self.expires_at = time() + ttl
self._expired = False
def expired(self):
if self._expired is False:
return (self.expires_at < time())
else:
return self._expired
class CacheList():
def __init__(self):
self.entries = []
self.lock = RLock()
def add_entry(self, string, ttl=20):
with self.lock:
self.entries.append(CacheEntry(string, ttl))
def read_entries(self):
with self.lock:
self.entries = list(itertools.dropwhile(lambda x:x.expired(), self.entries))
return self.entries
def read_entries(name, slp, cachelist):
while True:
print "{}: {}".format(name, ",".join(map(lambda x:x.string, cachelist.read_entries())))
sleep(slp)
def add_entries(name, ttl, cachelist):
s = 'A'
while True:
cachelist.add_entry(s, ttl)
print("Added ({}): {}".format(name, s))
sleep(1)
s += 'A'
if __name__ == "__main__":
signal.signal(signal.SIGINT, signal.SIG_DFL)
cl = CacheList()
print_threads = []
print_threads.append(Thread(None, read_entries, args=('t1', 1, cl)))
# print_threads.append(Thread(None, read_entries, args=('t2', 2, cl)))
# print_threads.append(Thread(None, read_entries, args=('t3', 3, cl)))
adder_thread = Thread(None, add_entries, args=('a1', 2, cl))
adder_thread.start()
for t in print_threads:
t.start()
for t in print_threads:
t.join()
adder_thread.join()
저는 @iutinvg 솔루션이 간단해서 정말 마음에 들었습니다.하지만 저는 캐시해야 할 모든 기능에 추가적인 논쟁을 넣고 싶지 않습니다.그래서 Lewis와 Javiers의 답변에 영감을 받아 장식가가 가장 좋을 것이라고 생각했습니다.하지만 (하비에르처럼) 제3자 라이브러리를 사용하고 싶지 않았고 Lewis 솔루션을 개선할 수 있다고 생각했습니다.이것이 제가 생각해낸 것입니다.
import time
from functools import lru_cache
def ttl_lru_cache(seconds_to_live: int, maxsize: int = 128):
"""
Time aware lru caching
"""
def wrapper(func):
@lru_cache(maxsize)
def inner(__ttl, *args, **kwargs):
# Note that __ttl is not passed down to func,
# as it's only used to trigger cache miss after some time
return func(*args, **kwargs)
return lambda *args, **kwargs: inner(time.time() // seconds_to_live, *args, **kwargs)
return wrapper
의 솔루션은 하여 더 바닥 은 를 를 하여 ( )//) 에 캐스팅이 필요하지 않습니다.
사용.
@ttl_lru_cache(seconds_to_live=10)
def expensive(a: int):
"""An expensive function."""
time.sleep(1 + a)
print("Starting...")
expensive(1)
print("Again...")
expensive(1)
print("Done")
이로 : 하면 로 하지 를 하지 로 하면 를 maxsize=None는 시간이 에 따라
이것이 제 구현입니다.
from functools import wraps
from time import time
def ttl_cache(ttl: int):
def wrapper(func):
cached = None
cached_at = time()
@wraps(func)
def _decorator(*args, **kwargs):
nonlocal cached, cached_at
now = time()
if (now - cached_at) >= ttl or cached is None:
cached_at = time()
cached = func(*args, **kwargs)
return cached
return _decorator
return wrapper
사용.
@ttl_cache(5)
def work():
return randint(0, 10)
while True:
print(work())
sleep(1)
변이 가능 매핑, Ordered Dict 및 default Dict(목록)이 있는 dicttl로 이동할 수도 있습니다.
각 키가 30초인 일반적인 딕트를 초기화합니다.
data = {'a': 1, 'b': 2}
dict_ttl = DictTTL(30, data)
오더드 딕트
data = {'a': 1, 'b': 2}
dict_ttl = OrderedDictTTL(30, data)
defaultDict(목록)
dict_ttl = DefaultDictTTL(30)
data = {'a': [10, 20], 'b': [1, 2]}
[dict_ttl.append_values(k, v) for k, v in data.items()]
누군가가 파이썬 패키지에 넣으려고 작업을 했습니다. https://github.com/vpaliy/lru-expiring-cache 를 참조.
글쎄요, 저는 다른 답변들(문제를 제대로 다루지 않는 답변들)로 인해 오해를 받았기 때문에, 이것이 최선의 도구가 아닐 수도 있습니다.여전히,
from lru import LruCache
cache = LruCache(maxsize=10, concurrent=True)
def producer(key: str, value = True, TTL = 20):
cache.add(key = key, value = value, expires = TTL)
def consumer():
remaining_items = cache.items()
# Alternatively, iterate over available items until you find one not in the cache
return remaining_items
producer("1", TTL = 1)
producer("5", TTL = 3)
print(consumer()) ## 1, 5
time.sleep(2)
print(consumer()) ## 5
time.sleep(2)
print(consumer()) ## nothing
은 도, 을 합니다.('Concurrent', 'True')동시 모드로 실행될 때 엔트리가 됩니다.
언급URL : https://stackoverflow.com/questions/31771286/python-in-memory-cache-with-time-to-live
'programing' 카테고리의 다른 글
| ORDER BY 절을 사용하거나 사용하지 않는 파티션에 대한 분석 카운트 (0) | 2023.09.13 |
|---|---|
| #로 재미있는 표기법 (0) | 2023.09.13 |
| 스크롤뷰 터치 핸들링 내의 수평 스크롤뷰 (0) | 2023.09.13 |
| 이 jQuery ajax 클릭 이벤트가 여러 번 발생하는 이유는 무엇입니까? (0) | 2023.09.13 |
| WordPress에서의 콜백 함수로서의 클래스 전달방법 (0) | 2023.09.13 |