装饰器

API 包路径: mcdreforged.api.decorator

@new_thread

@mcdreforged.api.decorator.new_thread.new_thread[源代码]

这是一个让你的函数多线程异步执行的一行代码解决方案。当使用这个装饰器时，函数将在一个新的守护线程中执行

这个装饰器只将函数的返回值改为创建的 Thread 对象。除了返回值，它还保留了被装饰的函数的所有签名，所以你可以安全地使用被装饰的函数，就像它们没有被装饰一样

这也是旧的 MCDR 0.x 插件的简单的兼容升级方法

被装饰的函数的返回值被修改为用于执行这个函数的 thread 对象

被修饰的函数拥有着 1 个额外的属性：

original 属性：储存着原始的未被修饰的函数对象

例子：

>>> import time

>>> @new_thread('My Plugin Thread')
... def do_something(text: str):
...     time.sleep(1)
...     print(threading.current_thread().name)
>>> callable(do_something.original)
True
>>> t = do_something('foo')
>>> isinstance(t, FunctionThread)
True
>>> t.join()
My Plugin Thread

参数:: arg – 一个 str，表示线程的名字。建议总是使用这个参数来指定线程名，因此当你通过 server.logger 输出某些内容时，一个有意义的线程名称，而非一个无意义的 Thread-3，将被显示

class mcdreforged.api.decorator.new_thread.FunctionThread(target, name, args, kwargs)[源代码]

一个 Thread 的子类，用于在装饰器 new_thread() 中包裹一个同步的函数调用

get_return_value(block: bool = False, timeout: float | None = None)[源代码]

获取原函数的返回值

如果调用原函数时抛出了异常，则那个异常将在这里被重新抛出

例子：

>>> import time
>>> @new_thread
... def do_something(text: str):
...     time.sleep(1)
...     return text

>>> do_something('task').get_return_value(block=True)
'task'

参数:

block – 是否应该在获取返回值前 join 此线程，以确保函数调用结束
timeout – 线程 join 的最大超时时间

抛出:

RuntimeError – 若在获取返回值时线程仍然存活。这可能是因为线程仍在运行且 block=False，或者线程 join 操作超时

返回:

原函数的返回值

@event_listener

@mcdreforged.api.decorator.event_listener.event_listener(event: PluginEvent | str, *, priority: int | None = None)[源代码]

这个装饰器是用于在不使用 register_event_listener() 的前提下注册自定义事件监听器

它接受单个 str 或 PluginEvent 作为参数，代表着你打算监听的事件，然后它将会将被装饰的函数作为对应监听器的回调函数

强烈建议你仅在插件的入口点使用此装饰器，来保证它可以正确地工作，且在正确的时间注册事件监听器

例子：

@event_listener(MCDRPluginEvents.GENERAL_INFO)
def my_on_info(server, info):
    server.logger.info('on info in my own listener')

上述例子等价于:

def on_load(server, old):
    server.register_event_listener(MCDRPluginEvents.GENERAL_INFO, my_on_info)

参数:

event – 由于注册事件监听器的事件

关键字参数:

priority – 可选参数，事件监听器的优先级

抛出:

TypeError – 如果给定的 event 非法
RuntimeError – 如果获取 PluginServerInterface 失败（见 as_plugin_server_interface() 以了解详情）

@spam_proof

@mcdreforged.api.decorator.spam_proof.spam_proof(arg=None, *, lock_class=<function RLock>, skip_callback: ~typing.Callable | None = None)[源代码]

用一个锁来保护被装饰的函数，避免它被多个线程同时调用

在“多重同时调用”发生时，只有第一个函数调用能被成功执行，其他的函数调用将被跳过

被装饰的函数的返回值被修改为一个 bool，代表这次函数调用是否正常地执行了

被修饰的函数拥有着 2 个额外的属性：

original 属性：储存着原始的未被修饰的函数对象
lock 属性：储存着防重复调用逻辑中使用的锁对象

例子：

@spam_proof
def some_work(value):
    # doing some important logics
    foo = value

上述例子等价于:

lock = threading.RLock()

def some_work(value) -> bool:
    acquired = lock.acquire(blocking=False)
    if acquired:
        try:
            # doing some not thread-safe logics
            foo = value
        finally:
            lock.release()
    return acquired

关键字参数:

lock_class – 锁的类型。它可以是 threading.Lock 或者 threading.RLock (默认值)
skip_callback – （可选参数）一个回调函数，将在在被装饰的函数调用被跳过时被调，参数为被装饰函数的所有参数

关键字参数 skip_callback 例子：

>>> def my_callback(value):
...     print('skip', value)

>>> @spam_proof(skip_callback=my_callback)
... def some_work(value):
...     event.wait()

>>> def threaded_invoke():
...     print(some_work(0.1))  # invocation normal

>>> from threading import Thread, Event
>>> t, event = Thread(target=threaded_invoke), Event()
>>> t.start()
>>> some_work(123)  # invocation skipped
skip 123
False
>>> _ = event.set(), t.join()
True

在 v2.5.0 版本加入.

在 v2.7.0 版本加入: 添加 skip_callback 关键字参数