tlmqtt：基于 Java 的高性能 MQTT Broker 开源项目
项目简介
tlmqtt是一款基于Java开发的轻量级、高并发 MQTT Broker，底层采用Netty异步通信框架与Project Reactor 响应式编程模型，实现了完整的MQTT 3.1.1 协议解析，支持 QoS 0/1/2 消息处理、主题通配符匹配、消息持久化等核心功能。项目模块化设计清晰，提供认证、数据桥接、存储等可扩展组件，适合构建物联网（IoT）、实时通信等场景的消息中间件。
核心特性
1\. 协议与消息处理
- 完整实现MQTT 3.1.1 协议，支持 CONNECT、PUBLISH、SUBSCRIBE 等核心报文解析
- 全链路支持QoS 0/1/2 消息投递，包含消息重发机制（默认 5 秒超时）
- 基于通配符（
+、#）的主题过滤与消息转发 - 支持保留消息（Retain Message）和遗嘱消息（Will Message）
2\. 扩展性与集成能力
- 认证模块：支持文件配置、MySQL 数据库、HTTP 接口认证，可自定义认证逻辑
- 数据桥接：内置 Kafka、MySQL 桥接组件，支持通过接口扩展自定义目标系统
- 存储模块：提供内存存储实现，支持扩展 Redis、数据库等持久化方案
- 多协议支持：同时提供 MQTT（1883 端口）和 WebSocket（8083 端口）接入能力
3\. 生产级特性
- 基于 Netty 的高性能异步 IO，支持高并发连接
- 可配置的 SSL/TLS 加密（通过配置文件启用）
- 会话持久化（CleanSession=0 时自动恢复未完成消息）
- 模块化架构设计，支持按需引入组件（如仅使用核心模块或扩展认证功能）
模块架构
| 模块名称 | 功能描述 |
|---|---|
tlmqtt-core |
核心协议解析与消息处理逻辑 |
tlmqtt-auth |
认证模块（文件 / 数据库 / HTTP） |
tlmqtt-bridge |
数据桥接模块（Kafka/MySQL） |
tlmqtt-common |
公共工具类与配置管理 |
tlmqtt-store |
存储接口与默认实现 |
tlmqtt-boot |
启动引导模块（MQTT/WS 服务） |
快速开始
环境要求
- Java 8+
- Maven 3.6+
- 可选依赖：MySQL（认证 / 存储）、Kafka（数据桥接）
1\. 克隆项目
bash
git clone https://github.com/ZHSQJM/tlmqtt.git
cd tlmqtt
2\. 配置文件
修改tlmqtt-common/src/main/resources/config.yml：
yaml
# 会话配置
session:
timeout: 30 # 会话超时时间（分钟）
delay: 5 # 消息重发延迟（秒）
# 端口配置
port:
mqtt: 1883 # MQTT默认端口
websocket: 8083 # WebSocket端口
# SSL配置（可选）
ssl:
enabled: false
certPath: /path/to/cert.crt
privatePath: /path/to/private.key
# 认证配置（默认开启）
auth:
enabled: true
user:
- username: "admin"
password: "123456"
3\. 启动服务
方式一：直接运行（MQTT 服务）
java
import com.tlmqtt.bootstrap.TlBootstrap;
import com.tlmqtt.server.TlMqttServer;
public class Main {
public static void main(String[] args) {
TlBootstrap bootstrap = new TlBootstrap();
bootstrap.setServer(TlMqttServer.class).start();
}
}
方式二：启动 WebSocket 服务
java
import com.tlmqtt.bootstrap.TlBootstrap;
import com.tlmqtt.server.TlWebSocketServer;
public class Main {
public static void main(String[] args) {
TlBootstrap bootstrap = new TlBootstrap();
bootstrap.setServer(TlWebSocketServer.class).start();
}
}
功能示例
1\. 认证功能配置
（1）文件认证（声明式）
在config.yml中添加用户：
yaml
auth:
user:
- username: "user1"
password: "pass1"
- username: "user2"
password: "pass2"
（2）编程式认证（自定义逻辑）
java
bootstrap.setAuth(false); // 关闭默认认证
// 或添加自定义认证器
bootstrap.addAuthentication(new AbstractTlAuthentication() {
@Override
public boolean authenticate(String username, String password) {
return "admin".equals(username) && "secret".equals(password);
}
});
2\. 数据桥接至 MySQL
java
import com.tlmqtt.bridge.mysql.TlMySqlInfo;
TlMySqlInfo mysqlInfo = new TlMySqlInfo();
mysqlInfo.setHost("127.0.0.1")
.setPort(3306)
.setDatabase("mqtt_db")
.setTable("messages")
.setUsername("root")
.setPassword("123456");
bootstrap.addBridgeMysql(mysqlInfo); // 注册MySQL桥接
3\. 自定义存储（以 Redis 为例）
java
import com.tlmqtt.store.SessionService;
import com.tlmqtt.store.impl.RedisSessionService;
SessionService redisSession = new RedisSessionService("localhost", 6379);
bootstrap.setSessionService(redisSession); // 替换默认内存存储
贡献与反馈
欢迎通过以下方式参与项目共建：
- 提交 Issue：反馈 Bug 或提出功能建议
- 提交 PR：优化代码或新增功能（建议先创建 Issue 沟通方案）
- Star/Fork：支持项目持续发展
联系方式：
- 邮箱：2534798858@qq.com
- 项目地址：https://github.com/ZHSQJM/tlmqtt
许可证
本项目采用MIT 许可证，允许商业使用、修改和再发布。详情见LICENSE文件。
tlmqtt致力于为物联网开发者提供轻量、高效的 MQTT 消息服务，期待您的加入！ 🚀
