Skip to main content

编写代码

本文带你从零开始，将你的应用接入 KnotLink 协议。我们会实现一个完整的消息提醒功能：其他程序调用接口，传入消息内容，弹出通知并回复"OK"。

选择你使用的语言：

选择 SDK​

纯 Python 应用使用 knotlink，适合 CLI 工具、后台服务、无 GUI 场景。基于回调函数，API 简洁直观。

pip install knotlink

PyQt5 桌面应用请使用 PyQt5 选项卡。

一、开放接口（Open Socket）—— 让你的程序能被调用​

请求-响应模式：你的程序作为回答者（Responser），接收其他程序的调用请求，处理后将结果返回。

场景​

我们要实现的功能很直观：

其他程序发来一条消息文本 → 我们弹出通知 → 回复 "OK" 表示收到

1.1 导入库​

from knotlink import OpenSocketResponser

就这么简单。OpenSocketResponser 是一个现成的类，封装了与 守护进程的全部通信细节。

1.2 创建回答者并设置回调​

# 用你的 AppID 和功能 ID 创建回答者
responser = OpenSocketResponser(
    APPID="com.example.msgreminder",  # 你的应用唯一标识（倒置域名格式）
    OpenSocketID="show"               # 这个接口的功能标识（描述性名称）
)

APPID 推荐使用倒置域名格式（如 com.example.msgreminder），确保在整个 KnotLink 网络中唯一。OpenSocketID 是一个描述性的功能名称（如 show、search、control），在单个应用内唯一即可。

旧格式兼容

早期项目使用 0x 开头的 8 位十六进制数字（如 APPID="0x00000014"），KnotLink 继续支持这种格式。如果你维护的是旧项目，无需迁移；新项目推荐使用倒置域名格式，可读性更好。详见 核心概念 — 标识符命名规范。

接下来，告诉回答者"收到请求时怎么处理"——设置回调函数：

def handle_request(data: str) -> str:
    """收到请求时的处理逻辑"""
    print(f"收到消息：{data}")
    show_notification(data)   # 弹出通知（你需要自己实现这个函数）
    return "OK"               # 返回值会发送回调用方

responser.set_RecvFunc(handle_request)

回调函数的签名是 (data: str) -> str：

  • 入参 data：调用方发来的原始数据（字符串格式）
  • 返回值：将自动发送回调用方作为响应
回调在后台线程中执行

set_RecvFunc 的回调在守护线程中被调用，不会阻塞你的主程序。如果你的处理逻辑涉及 UI 操作（如弹出窗口），请确保线程安全。

1.3 用 KLUDF 解析请求数据​

调用方发来的 data 是原始字符串。为了让数据交换有章可循，KnotLink 推荐使用 KLUDF 键值对格式：

msgContext=测试消息;type=popup;duration=3000

用 KLKVMap 解析：

from knotlink import KLKVMap

def handle_request(data: str) -> str:
    # 解析键值对格式的请求数据
    kv = KLKVMap()
    kv.deserialize(data)

    message = kv.get("msgContext", "默认消息")   # 获取消息内容
    msg_type = kv.get("type", "popup")           # 获取消息类型
    duration = kv.get("duration", "3000")        # 获取显示时长（毫秒）

    print(f"弹出通知：{message}，类型：{msg_type}，持续 {duration}ms")
    show_notification(message, msg_type, int(duration))

    return "OK"

KLKVMap 提供三个核心方法：

方法说明示例
deserialize(str)将 key1=value1;key2=value2 字符串解析为字典kv.deserialize("name=张三;age=18")
serialize()将字典序列化为键值对字符串kv.serialize() → "name=张三;age=18"
get(key, default)安全取值，键不存在时返回默认值kv.get("name", "未知")

关于 KLUDF 支持的全部三种格式（键值对 / JSON / CLI），详见 消息体规范。

一个接口，多重功能

单个 OpenSocketResponser 可以处理多种逻辑功能——通过请求参数中的 cmd、action 等字段路由到不同操作。例如：

def handle_request(data: str) -> str:
    kv = KLKVMap()
    kv.deserialize(data)
    cmd = kv.get("cmd", "")

    if cmd == "BACKUP":
        return do_backup(kv.get("folder", ""))
    elif cmd == "RESTORE":
        return do_restore(kv.get("folder", ""))
    else:
        return "未知指令"

在 funclist.json 中可以将 BACKUP 和 RESTORE 声明为两个独立功能条目，共用同一个 openSocketID，通过 static 类型的 cmd 参数区分。详见 填写功能清单。

1.4 完整示例：命令行消息提醒​

下面是一个可直接运行的完整示例。它启动后持续监听，任何程序都能向它发送消息：

"""
message_reminder.py — KnotLink 消息提醒服务
启动后监听 com.example.msgreminder/show，接收消息并打印到控制台。
"""

import time
from knotlink import OpenSocketResponser, KLKVMap


def handle_request(data: str) -> str:
    """处理收到的请求"""
    kv = KLKVMap()
    kv.deserialize(data)

    message = kv.get("msgContext", "（无内容）")
    sender = kv.get("sender", "未知来源")

    # 在实际项目中，这里可以弹出 GUI 通知窗口
    print(f"\n📩 收到来自 [{sender}] 的消息：")
    print(f"   {message}")
    print(f"   已回复：OK")

    return "OK"


def main():
    print("🔗 消息提醒服务启动中...")
    print("   等待其他程序调用...（按 Ctrl+C 退出）\n")

    # 创建回答者，绑定到指定的 AppID 和 OpenSocketID
    responser = OpenSocketResponser(
        APPID="com.example.msgreminder",
        OpenSocketID="show"
    )
    responser.set_RecvFunc(handle_request)

    try:
        # 保持程序运行，持续监听
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        print("\n👋 消息提醒服务已退出")


if __name__ == "__main__":
    main()

1.5 调试：用 SDK 自带的测试工具验证​

写好回答者后，不需要等别人来调用——用 OpenSocketQuerier 写一个简单的测试脚本：

"""
test_reminder.py — 测试消息提醒接口
"""
from knotlink import OpenSocketQuerier, KLKVMap

# 构建请求数据（键值对格式）
kv = KLKVMap()
kv["msgContext"] = "这是一条测试消息"
kv["sender"] = "测试脚本"

# 创建查询者，向目标接口发送请求
querier = OpenSocketQuerier(
    APPID="com.example.msgreminder",
    OpenSocketID="show"
)

# 同步查询：发送请求并等待响应
response = querier.query(kv.serialize())
print(f"响应：{response}")

测试步骤：

  1. 先启动 message_reminder.py
  2. 再运行 test_reminder.py
  3. 观察 message_reminder.py 的控制台输出，确认收到消息
  4. 观察 test_reminder.py 的控制台输出，确认收到 "OK" 响应

二、发送信号（Signal）—— 向外部广播事件​

发布-订阅模式：你的程序作为信号发送者（SignalSender），向 KnotLink 网络广播事件。任何订阅了该信号的程序都能收到。

场景​

我们要实现：当某个事件发生时（比如定时器触发、文件变更），向外部广播一条通知信号，携带事件相关信息。

2.1 导入库​

from knotlink import SignalSender

2.2 创建信号发送者​

# 创建发送者，同时配置 AppID 和 SignalID
sender = SignalSender(
    APPID="com.example.msgreminder",
    SignalID="messageReceived"
)

也可以在创建后再配置：

sender = SignalSender()
sender.set_config(
    APPID="com.example.msgreminder",
    SignalID="messageReceived"
)

2.3 用 KLUDF 构建信号数据​

信号的携带数据同样推荐用 KLUDF 键值对格式：

from knotlink import KLKVMap

# 构建信号数据
kv = KLKVMap()
kv["event"] = "timer_fired"
kv["message"] = "定时任务已触发"
kv["timestamp"] = "2026-07-05 14:30:00"

# 序列化为字符串
signal_data = kv.serialize()
# → "event=timer_fired;message=定时任务已触发;timestamp=2026-07-05 14:30:00"

2.4 发送信号​

一行代码：

sender.emitt(signal_data)

emitt() 是异步的——调用后立即返回，不会阻塞你的程序。订阅者会在后台收到信号。

2.5 完整示例：定时广播​

"""
timer_broadcast.py — 每 5 秒广播一次当前时间
"""

import time
from datetime import datetime
from knotlink import SignalSender, KLKVMap


def main():
    sender = SignalSender(
        APPID="com.example.msgreminder",
        SignalID="messageReceived"
    )

    print("🔗 定时广播服务启动，每 5 秒发送一次信号...")
    print("   （按 Ctrl+C 退出）\n")

    count = 0
    try:
        while True:
            count += 1
            now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")

            # 构建信号数据
            kv = KLKVMap()
            kv["event"] = "heartbeat"
            kv["count"] = str(count)
            kv["timestamp"] = now

            # 发送信号
            sender.emitt(kv.serialize())
            print(f"📡 第 {count} 次广播 → {kv.serialize()}")

            time.sleep(5)
    except KeyboardInterrupt:
        print(f"\n👋 共广播 {count} 次，服务已退出")


if __name__ == "__main__":
    main()

2.6 调试：订阅信号验证发送​

用 SignalSubscriber 订阅同一个信号，验证发送端是否正常工作：

"""
test_subscriber.py — 订阅信号，验证广播
"""
import time
from knotlink import SignalSubscriber, KLKVMap


def on_signal(data: str):
    """收到信号时的处理"""
    kv = KLKVMap()
    kv.deserialize(data)
    event = kv.get("event", "未知事件")
    count = kv.get("count", "?")
    timestamp = kv.get("timestamp", "?")
    print(f"📩 收到信号 [{event}] 第 {count} 次 @ {timestamp}")


def main():
    print("🔗 信号订阅者启动，等待广播...（按 Ctrl+C 退出）\n")

    subscriber = SignalSubscriber(
        APPID="com.example.msgreminder",
        SignalID="messageReceived"
    )
    subscriber.set_RecvFunc(on_signal)

    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        print("\n👋 订阅者已退出")


if __name__ == "__main__":
    main()

测试步骤：

  1. 先启动 test_subscriber.py
  2. 再运行 timer_broadcast.py
  3. 观察订阅者控制台，应每 5 秒收到一条信号

三、同一程序同时提供接口 + 发送信号​

在实际项目中，一个程序往往既要开放接口（让别人调用自己），又要发送信号（向外部广播事件）。二者可以共存：

"""
combined_service.py — 同时提供接口和发送信号
"""

import time
from knotlink import OpenSocketResponser, SignalSender, KLKVMap


def handle_request(data: str) -> str:
    """处理接口调用"""
    kv = KLKVMap()
    kv.deserialize(data)
    message = kv.get("msgContext", "")

    # 处理请求的同时，广播一条信号通知其他程序
    notify_kv = KLKVMap()
    notify_kv["event"] = "message_received"
    notify_kv["content"] = message
    signal_sender.emitt(notify_kv.serialize())

    print(f"处理请求并广播信号：{message}")
    return "OK"


# 全局初始化
signal_sender = SignalSender(
    APPID="com.example.msgreminder",
    SignalID="messageReceived"
)
responser = OpenSocketResponser(
    APPID="com.example.msgreminder",
    OpenSocketID="show"
)
responser.set_RecvFunc(handle_request)


def main():
    print("🔗 组合服务启动（接口 + 信号）...")
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        print("\n👋 服务已退出")


if __name__ == "__main__":
    main()

四、进阶：使用 JSON 格式​

当数据层级复杂时，KLUDF 同样支持 JSON 格式。直接使用 Python 内置的 json 库即可：

import json
from knotlink import OpenSocketResponser

def handle_request(data: str) -> str:
    # 尝试解析为 JSON
    try:
        obj = json.loads(data)
        message = obj.get("msgContext", "")
        options = obj.get("options", {})
    except json.JSONDecodeError:
        # 回退到键值对解析
        from knotlink import KLKVMap
        kv = KLKVMap()
        kv.deserialize(data)
        message = kv.get("msgContext", "")

    print(f"收到消息：{message}")
    return json.dumps({"status": "ok", "received": message})

responser = OpenSocketResponser("com.example.msgreminder", "show")
responser.set_RecvFunc(handle_request)

对于大多数场景，键值对格式足够且更轻量。当需要嵌套对象、数组等复杂结构时再选用 JSON。

关于消息格式的完整规范与选型建议，参见 消息体规范。

五、关键注意事项​

5.1 阻塞与并发​

OpenSocketResponser 的回调在后台守护线程中执行，不会阻塞主线程。但如果你的回调执行时间很长（如超过 30 秒），调用方可能超时。建议：

  • 耗时操作交给后台任务队列，回调中快速返回确认
  • 对于需要异步处理后再回复的场景，先用 "ACCEPTED" 占位，处理完成后再通过 Signal 通知结果

5.2 标识符唯一性​

KnotLink 支持两种标识符风格：

  • 新格式（推荐）：appID 采用倒置域名（如 com.yourcompany.yourapp），openSocketID / signalID 使用描述性名称（如 search、messageReceived）
  • 旧格式（兼容）：全部使用 0x 开头的 8 位十六进制数字（如 0x00000014）

两种格式均可在同一网络中混用。新项目建议采用新格式以获得更好的可读性。无论哪种格式，appID + openSocketID / signalID 的组合必须全局唯一。

5.3 守护进程依赖​

所有通信都依赖 守护进程。如果守护进程未启动，OpenSocketResponser 和 SignalSender 的初始化将失败（无法连接 127.0.0.1:6378 等端口）。

在生产环境中，建议在初始化时捕获连接异常并给出友好提示。

六、集成到你的项目​

上面的完整示例是独立演示程序。实际使用时，你需要把 KnotLink 嵌入你已有的项目。只需三步：

① 安装依赖

pip install knotlink

② 在你现有的初始化代码中加上几行

# 你的项目已有的 main() 或初始化函数
def main():
    init_your_business_logic()
    init_your_ui()

    # +++ 加上这三行 +++
    from knotlink import OpenSocketResponser
    responser = OpenSocketResponser("com.yourcompany.yourapp", "yourFunction")
    responser.set_RecvFunc(your_handler)  # 对接你已有的业务逻辑

    your_event_loop()  # 你原来的主循环

③ 写一个 handler 桥接到你的业务代码

def your_handler(data: str) -> str:
    # data 是 KnotLink 传进来的请求
    # 调用你项目中已有的函数处理
    result = your_existing_business_logic(data)
    return result  # 返回给调用方

不需要重构项目结构，不需要引入新线程。OpenSocketResponser 和 SignalSender 在后台自动处理通信，你的代码保持原样。

上面所有代码示例都是这个模式——把它们当模板，替换成你自己的 APPID、函数名和业务逻辑即可。

下一步​

  • 代码写好后，需要填写功能清单（funclist.json），让工具链和其他开发者能发现你的接口 → 填写功能清单
  • 准备好发布？填写插件自述文件并打包 → 打包与分发