Skip to content

# 模块 `liteyuki.comm.storage` ​

共享内存模块。类似于redis，但是更加轻量级并且线程安全

### var `_on_main_subscriber_receive_funcs` ​

- 说明: 主进程订阅者接收函数

- 类型: `dict[str, list[ASYNC_ON_RECEIVE_FUNC]]`

- 默认值: `{}`

### var `_on_sub_subscriber_receive_funcs` ​

- 说明: 子进程订阅者接收函数

- 类型: `dict[str, list[ASYNC_ON_RECEIVE_FUNC]]`

- 默认值: `{}`

### class `KeyValueStore` ​

#### func `__init__(self)` ​

源代码 或 在GitHub上查看
python

    def __init__(self):
        self._store = {}
        self.active_chan = Channel[tuple[str, Optional[dict[str, Any]]]](name='shared_memory-active')
        self.passive_chan = Channel[tuple[str, Optional[dict[str, Any]]]](name='shared_memory-passive')
        self.publish_channel = Channel[tuple[str, Any]](name='shared_memory-publish')
        self.is_main_receive_loop_running = False
        self.is_sub_receive_loop_running = False

#### func `set(self, key: str, value: Any) -> None` ​

说明: 设置键值对

参数:

> - key: 键
> - value: 值

源代码 或 在GitHub上查看
python

    def set(self, key: str, value: Any) -> None:
        if IS_MAIN_PROCESS:
            lock = _get_lock(key)
            with lock:
                self._store[key] = value
        else:
            self.passive_chan.send(('set', {'key': key, 'value': value}))

#### func `get(self, key: str, default: Optional[Any] = None) -> Optional[Any]` ​

说明: 获取键值对

参数:

> - key: 键
> - default: 默认值

返回: Any: 值

源代码 或 在GitHub上查看
python

    def get(self, key: str, default: Optional[Any]=None) -> Optional[Any]:
        if IS_MAIN_PROCESS:
            lock = _get_lock(key)
            with lock:
                return self._store.get(key, default)
        else:
            recv_chan = Channel[Optional[Any]]('recv_chan')
            self.passive_chan.send(('get', {'key': key, 'default': default, 'recv_chan': recv_chan}))
            return recv_chan.receive()

#### func `delete(self, key: str, ignore_key_error: bool = True) -> None` ​

说明: 删除键值对

参数:

> - key: 键
> - ignore_key_error: 是否忽略键不存在的错误

源代码 或 在GitHub上查看
python

    def delete(self, key: str, ignore_key_error: bool=True) -> None:
        if IS_MAIN_PROCESS:
            lock = _get_lock(key)
            with lock:
                if key in self._store:
                    try:
                        del self._store[key]
                        del _locks[key]
                    except KeyError as e:
                        if not ignore_key_error:
                            raise e
        else:
            self.passive_chan.send(('delete', {'key': key}))

#### func `get_all(self) -> dict[str, Any]` ​

说明: 获取所有键值对

返回: dict[str, Any]: 键值对

源代码 或 在GitHub上查看
python

    def get_all(self) -> dict[str, Any]:
        if IS_MAIN_PROCESS:
            return self._store
        else:
            recv_chan = Channel[dict[str, Any]]('recv_chan')
            self.passive_chan.send(('get_all', {'recv_chan': recv_chan}))
            return recv_chan.receive()

#### func `publish(self, channel_: str, data: Any) -> None` ​

说明: 发布消息

参数:

> - channel_: 频道
> - data: 数据

源代码 或 在GitHub上查看
python

    def publish(self, channel_: str, data: Any) -> None:
        self.active_chan.send(('publish', {'channel': channel_, 'data': data}))

#### func `on_subscriber_receive(self, channel_: str) -> Callable[[ON_RECEIVE_FUNC], ON_RECEIVE_FUNC]` ​

说明: 订阅者接收消息时的回调

参数:

> - channel_: 频道

返回: 装饰器

源代码 或 在GitHub上查看
python

    def on_subscriber_receive(self, channel_: str) -> Callable[[ON_RECEIVE_FUNC], ON_RECEIVE_FUNC]:
        if not IS_MAIN_PROCESS:
            raise RuntimeError('Cannot subscribe in sub process.')

        def decorator(func: ON_RECEIVE_FUNC) -> ON_RECEIVE_FUNC:

            async def wrapper(data: Any):
                if is_coroutine_callable(func):
                    await func(data)
                else:
                    func(data)
            if IS_MAIN_PROCESS:
                if channel_ not in _on_main_subscriber_receive_funcs:
                    _on_main_subscriber_receive_funcs[channel_] = []
                _on_main_subscriber_receive_funcs[channel_].append(wrapper)
            else:
                if channel_ not in _on_sub_subscriber_receive_funcs:
                    _on_sub_subscriber_receive_funcs[channel_] = []
                _on_sub_subscriber_receive_funcs[channel_].append(wrapper)
            return wrapper
        return decorator

`@staticmethod`

#### async func `run_subscriber_receive_funcs(channel_: str, data: Any)` ​

说明: 运行订阅者接收函数

参数:

> - channel_: 频道
> - data: 数据

源代码 或 在GitHub上查看
python

    @staticmethod
    async def run_subscriber_receive_funcs(channel_: str, data: Any):
        [asyncio.create_task(func(data)) for func in _on_main_subscriber_receive_funcs[channel_]]

#### async func `start_receive_loop(self)` ​

说明: 启动发布订阅接收器循环，在主进程中运行，若有子进程订阅则推送给子进程

源代码 或 在GitHub上查看
python

    async def start_receive_loop(self):
        if not IS_MAIN_PROCESS:
            raise RuntimeError('Cannot start receive loop in sub process.')
        while True:
            data = await self.active_chan.async_receive()
            if data[0] == 'publish':
                await self.run_subscriber_receive_funcs(data[1]['channel'], data[1]['data'])
                self.publish_channel.send(data)

### class `GlobalKeyValueStore` ​

`@classmethod`

#### func `get_instance(cls)` ​

源代码 或 在GitHub上查看
python

    @classmethod
    def get_instance(cls):
        if cls._instance is None:
            with cls._lock:
                if cls._instance is None:
                    cls._instance = KeyValueStore()
        return cls._instance

#### attr `_instance = None` ​

#### attr `_lock = threading.Lock()` ​

### var `shared_memory` ​

- 说明: 共享内存对象

- 类型: `KeyValueStore`

- 默认值: `GlobalKeyValueStore.get_instance()`

### var `_ref_count` ​

- 说明: import 引用计数, 防止获取空指针

- 默认值: `0`

`@shared_memory.passive_chan.on_receive(lambda d: d[0] == 'get')`

### func `on_get(data: tuple[str, dict[str, Any]])` ​

源代码 或 在GitHub上查看
python

    @shared_memory.passive_chan.on_receive(lambda d: d[0] == 'get')
    def on_get(data: tuple[str, dict[str, Any]]):
        key = data[1]['key']
        default = data[1]['default']
        recv_chan = data[1]['recv_chan']
        recv_chan.send(shared_memory.get(key, default))

`@shared_memory.passive_chan.on_receive(lambda d: d[0] == 'set')`

### func `on_set(data: tuple[str, dict[str, Any]])` ​

源代码 或 在GitHub上查看
python

    @shared_memory.passive_chan.on_receive(lambda d: d[0] == 'set')
    def on_set(data: tuple[str, dict[str, Any]]):
        key = data[1]['key']
        value = data[1]['value']
        shared_memory.set(key, value)

`@shared_memory.passive_chan.on_receive(lambda d: d[0] == 'delete')`

### func `on_delete(data: tuple[str, dict[str, Any]])` ​

源代码 或 在GitHub上查看
python

    @shared_memory.passive_chan.on_receive(lambda d: d[0] == 'delete')
    def on_delete(data: tuple[str, dict[str, Any]]):
        key = data[1]['key']
        shared_memory.delete(key)

`@shared_memory.passive_chan.on_receive(lambda d: d[0] == 'get_all')`

### func `on_get_all(data: tuple[str, dict[str, Any]])` ​

源代码 或 在GitHub上查看
python

    @shared_memory.passive_chan.on_receive(lambda d: d[0] == 'get_all')
    def on_get_all(data: tuple[str, dict[str, Any]]):
        recv_chan = data[1]['recv_chan']
        recv_chan.send(shared_memory.get_all())