您现在的位置是：首页 > 文章详情

tlmqtt：基于 Java 的高性能 MQTT Broker 开源项目

日期：2025-05-29点击：21收藏

项目简介

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); // 替换默认内存存储