zmq-server 0.1.0

A ZeroMQ based message system with topic subscription, authentication, and heartbeat

ZMQ Server

一个基于 ZeroMQ 的通用消息系统，支持 Topic 订阅、心跳机制、用户认证和消息转发。

功能特性

• Topic 订阅发布：客户端可以订阅感兴趣的 Topic，服务器只将消息发送给订阅了该 Topic 的客户端
• 用户认证：支持用户名密码认证，确保系统安全
• 心跳机制：自动检测客户端连接状态，清理超时客户端
• 消息转发：服务器只负责消息转发，不关心消息内容
• 自定义处理：支持自定义消息处理器，灵活处理各种消息类型
• 客户端ID管理：支持手动设置客户端ID，相同ID不能同时在线

安装

本地安装

pip install -r requirements.txt

Docker 运行

构建镜像

# 创建 Dockerfile
echo 'FROM python:3.10-slim
WORKDIR /app
COPY . /app
RUN pip install -r requirements.txt
CMD python -c "from zmq_server import ZMQServer; s = ZMQServer(users={\"admin\":\"admin123\",\"user1\":\"password1\"}); s.start()"' > Dockerfile

# 构建镜像
docker build -t zmq-server .

启动容器

# 启动服务器容器
docker run -d --name zmq-server -p 5555:5555 zmq-server

# 查看日志
docker logs zmq-server

# 停止容器
docker stop zmq-server

# 移除容器
docker rm zmq-server

快速开始

服务器端

from zmq_core import ZMQServer
import threading

# 创建服务器
server = ZMQServer(
    host="*", 
    port=5555, 
    users={
        "admin": "admin123",
        "user1": "password1"
    }
)

# 启动服务器
server_thread = threading.Thread(target=server.start, daemon=True)
server_thread.start()

客户端

from zmq_server import ZMQClient

# 创建客户端（带自定义ID）
client = ZMQClient(
    server_host="localhost",
    server_port=5555,
    client_id="my-client-001"  # 自定义客户端ID
)
client.connect()

# 认证
client.authenticate("user1", "password1")

# 订阅 Topic
client.subscribe("tech")
client.subscribe("news")

# 发布消息
client.publish("tech", "Hello World!")

# 开始接收消息
client.start_receiving()

运行示例

基础示例

python zmq_example.py

新功能测试

python test_new_features.py

服务器测试

python test_server.py

客户端测试

python test_client.py

API 参考

ZMQServer

方法	说明
__init__(host, port, users)	初始化服务器
add_user(username, password)	添加用户
start()	启动服务器
stop()	停止服务器
register_handler(message_type, handler)	注册自定义消息处理器
get_statistics()	获取服务器统计信息

ZMQClient

方法	说明
__init__(server_host, server_port, client_id)	初始化客户端
connect()	连接到服务器
authenticate(username, password)	用户认证
subscribe(topic)	订阅 Topic
unsubscribe(topic)	取消订阅 Topic
publish(topic, content)	发布消息
start_receiving()	开始异步接收消息
stop_receiving()	停止接收消息
register_handler(message_type, handler)	注册自定义消息处理器
set_default_handler(handler)	设置默认消息处理器
get_subscribed_topics()	获取已订阅的 Topic
is_authenticated()	检查是否已认证
close()	关闭客户端

消息格式

认证消息

{
  "type": "auth",
  "username": "user1",
  "password": "password1"
}

订阅消息

{
  "type": "subscribe",
  "topic": "tech"
}

发布消息

{
  "type": "publish",
  "topic": "tech",
  "content": "Hello World!"
}

心跳消息

{
  "type": "heartbeat",
  "timestamp": 1234567890
}

配置

服务器配置

• host: 服务器监听地址（默认："*"）
• port: 服务器监听端口（默认：5555）
• users: 用户字典（默认：空）
• heartbeat_interval: 心跳间隔（默认：30秒）
• client_timeout: 客户端超时时间（默认：60秒）

客户端配置

• server_host: 服务器地址（默认："localhost"）
• server_port: 服务器端口（默认：5555）
• client_id: 客户端ID（默认：自动生成）

许可证

MIT License