Varlink 是一种由 Red Hat 开发的现代化 IPC 协议，其设计灵感来源于 JSON-RPC，但通过强类型接口定义、服务发现和异步事件推送等机制进行了扩展，旨在简化分布式系统中的服务通信、接口契约管理和跨语言交互。

主要的应用场景：

1. systemd：通过 API 替代命令行工具
2. Podman（无守护进程的容器引擎）

核心特性

• 强类型接口定义

  Varlink 使用 IDL（接口定义语言）明确定义服务接口，支持类型检查和服务版本控制：
  示例接口定义（org.example.Service.interface）

  interface org.example.Service {
      method GetTime() -> (time string)
      method ListProcesses() -> (processes list<Process>)
  }
  

• 跨语言支持 Varlink 提供多种语言的 SDK，包括 Python、Go、Rust 等，服务端与客户端可使用不同语言实现。

• 异步通信模型 基于事件驱动的非阻塞 I/O 模型，支持高并发场景。
• 内置服务发现 通过标准 Unix socket 或 TCP 提供服务注册与发现机制。

代理示例 1：查询 Systemd 日志（类似 journalctl）

import varlink

def query_systemd_log():
    try:
        # 连接到 systemd 的 Varlink 接口
        with varlink.Client("unix:/var/run/systemd/varlink.sock") as client:
            # 调用 systemd 的 GetLog 方法
            log_data = client.call("org.freedesktop.systemd1.Manager.GetLog")

            # 解析返回的日志数据
            for entry in log_data.get("entries", []):
                print(f"[{entry['timestamp']}] {entry['MESSAGE']}")

    except varlink.VarlinkError as e:
        print(f"Varlink 调用错误: {e}")
    except Exception as e:
        print(f"其他错误: {e}")

if __name__ == "__main__":
    query_systemd_log()

代码示例 2：服务器端

time-service/
├── interfaces/
│   └── org.example.Service.v1.0.interface
├── server.py
└── client.py

server.py

import asyncio
import varlink
import datetime

class TimeService(varlink.Service):
    @varlink.method(interface='org.example.Service')
    def GetTime(self):
        return {'time': str(datetime.datetime.now())}

async def main():
    service = TimeService()
    await service.listen('unix:/run/time-service.sock', bus='session')

if __name__ == '__main__':
    asyncio.run(main())

client.py

import varlink

try:
    with varlink.Client('unix:/run/time-service.sock') as c:
        result = c.call('org.example.Service.GetTime')
        print(f"Current time: {result['time']}")
except varlink.ConnectionError as e:
    print(f"连接失败: {e}")
except varlink.MethodNotFound as e:
    print(f"方法不存在: {e}")
except Exception as e:
    print(f"未知错误: {e}")

org.example.Service.v1.0.interface

interface org.example.Service v1.0 {
    method GetTime() -> (time string)
    method ListProcesses() -> (processes list<Process>)
}

运行：

# 启动服务
sudo mkdir -p /run/time-service.sock.d
sudo chown $USER:$USER /run/time-service.sock.d
python3 server.py

# 查看服务
varlink list unix:/run/time-service.sock
# 输出：
# org.example.Service /org.example.Service GetTime

# 查看接口
varlink introspect unix:/run/time-service.sock org.example.Service
# 输出内容和 interface 文件定义一致

python3 client.py
# 输出：
# Current time: 2023-10-05 14:30:00.123456

参考资料

• https://varlink.org/
• https://github.com/varlink/varlink
• https://pypi.org/project/varlink/
• https://pkg.go.dev/github.com/varlink/go/varlink