forked from bot/app
1.3 KiB
1.3 KiB
title | order | icon | category |
---|---|---|---|
liteyuki.core.manager | 1 | laptop-code | API |
class ChannelDeliver
def __init__(self, active: Channel[Any], passive: Channel[Any], channel_deliver_active: Channel[Channel[Any]], channel_deliver_passive: Channel[tuple[str, dict]]) -> None
class ProcessManager
进程管理器
def __init__(self, lifespan: 'Lifespan') -> None
def start(self, name: str) -> None
开启后自动监控进程,并添加到进程字典中
Args:
name:
Returns:
def start_all(self) -> None
启动所有进程
def add_target(self, name: str, target: TARGET_FUNC, args: tuple, kwargs: Any) -> None
添加进程
Args:
name: 进程名,用于获取和唯一标识
target: 进程函数
args: 进程函数参数
kwargs: 进程函数关键字参数,通常会默认传入chan_active和chan_passive
def join_all(self) -> None
def terminate(self, name: str) -> None
终止进程并从进程字典中删除
Args:
name:
Returns:
def terminate_all(self) -> None
def is_process_alive(self, name: str) -> bool
检查进程是否存活
Args:
name:
Returns: