项目简介

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 时自动恢复未完成消息）
• 模块化架构设计，支持按需引入组件（如仅使用核心模块或扩展认证功能）

模块架构

快速开始

环境要求

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

贡献与反馈

欢迎通过以下方式参与项目共建：

1. 提交 Issue：反馈 Bug 或提出功能建议
2. 提交 PR：优化代码或新增功能（建议先创建 Issue 沟通方案）
3. Star/Fork：支持项目持续发展

联系方式：

• 邮箱：2534798858@qq.com
• 项目地址：https://github.com/ZHSQJM/tlmqtt

许可证

本项目采用MIT 许可证，允许商业使用、修改和再发布。详情见LICENSE文件。

tlmqtt致力于为物联网开发者提供轻量、高效的 MQTT 消息服务，期待您的加入！ 🚀