TTIoT云端物联网组件;面向JAVA;以事件为驱动;为设备提供安全可靠的连接通信能力
2.1版本重要更新
增加SSL加密功能&＃xff0c;同时支持https\wss\tls协议
在配置文件添加
ssl:
enabled: true
certificateType: PKCS12
certificatePath: /Users/shijun/Desktop/TTIOT/ca/server/3938.pfx
certificatePassword: 5J7HDiJM
2.0版本重要更新
多协议转换引擎-一个端口支持三种协议HTTP/WEBSOCKET/TCP&＃xff0c;上层协议使用的依然是MQTT&＃xff0c;支持使用这三种不同协议的客户端相互通信(详见使用说明.10)
使用HTTP协议向设备投递消息,URI:/mqtt,请求体如下&＃xff1a;
Headers:userName //设备帐号
tenantId //租户编号
password //设备密码
Body:
{
"mqttQos": 1, //消息质量
"payload": { //消息体
"rule": "chat",
"content": "hello"
},
"retain": false, //是否为retain消息
"topic": "/m" //推送的话题
}
支持消息路由引擎&＃64;Ruler-给消息定义规则(详见使用说明.12)
功能
认证授权(auth)
TTIoT的容器对发布的事件都会自动检查登陆状态&＃xff0c;对未授权的设备进行自动拦截&＃xff0c;只要在相关的类添加&＃64;Eventor(auth &＃61; true)即可
&＃64;Eventor(value &＃61; AppProtocalMqttConstant.PINGREQ,auth &＃61; true)
public class MqttPingEvent extends MqttApplicationEvent {
public MqttPingEvent(MqttMessage msg, ChannelHandlerContext context) {
super(msg, context);
}
}
多租户(saas)
TTIot的设备归属为租户&＃xff0c;其devName与topic对每个租户唯一&＃xff0c;clientId构成为&＃xff1a;tenantId-devName-xxxx(保留字段)&＃xff1b;并且提供接口将设备进行录入&＃xff0c;例如
dbHelper.saveTenant(tenant);
dbHelper.saveDev(device);
控制反转(IOC)
TTOT放弃第三方框架IOC框架&＃xff0c;内部维护了一个轻量的容器&＃xff0c;在项目启动时对所有的Listener扫描并且接管
消息的订阅与发布(Qos)
(1)支持Qos(Quality of Service)
TTIot提供mqtt三种消息质量模型即&＃xff1a;
Qos0&＃xff1a;最多一次的传输
Qos1&＃xff1a;至少一次的传输
Qos2&＃xff1a;只有一次的传输
该方案适应任何网络场景&＃xff0c;特别是网络较差的场景
(2)通配符与消息降级
TTIot提供主题层级分隔符/,单层通配符&＃43;,多层通配符#&＃xff0c;设备实现灵活订阅
心跳检测(Idle)
TTIot 提供两种测活机制
(1)服务端定义
TTiot:
heartbeatTimeout: 10
(2)客户端通过设置可变头里面的keepAliveTimeSeconds定义
保留消息与消息遗嘱
(1)Topic只有唯一的retain消息&＃xff0c;Broker会保存每个Topic的最后一条retain消息&＃xff1b;每个Client订阅Topic后会立即读取到retain消息&＃xff0c;不必要等待发送。订阅Topic时可以使用通配符&＃xff0c;就会收到匹配的每个Topic的retain消息&＃xff1b;发布消息时把retain设置为true&＃xff0c;即为保留信息。
(2)MQTT本身就是为信号不稳定的网络设计的&＃xff0c;所以难免一些客户端会无故的和Broker断开连接&＃xff1b;当客户端连接到Broker时&＃xff0c;可以指定LWT&＃xff0c;Broker会定期检测客户端是否有异常&＃xff1b;当客户端异常掉线时&＃xff0c;Broker就往连接时指定的topic里推送当时指定的LWT消息。
使用说明
配置文件
配置文件延续了yaml风格&＃xff0c;默认命名为TTIotBootstrap.yml&＃xff0c;在自己模块resources目录下添加TTIotBootstrap.yml文件即可
TTiot:
port: 8726 #端口
heartbeatTimeout: 10 #心跳间隔时间
netty:
bossGroupCount: -1 #-1代表线程数量取决于cpu
workerGroupCount: -1 #-1代表线程数量取决于cpu,此参数决定了netty的worker线程与TTIot内置的核心线程
redis: #redis相关
项目启动
new ServerLauncher().launch();
自定义数据源
(1) 实现DbHelper接口
(2) 添加数据源插件&＃xff0c;Broker提供默认的redis实现
new ServerLauncher().dbHelper(new RedisDbHelper(new RedisSourceProvider())).launch();
自定义统一异常处理器
(1) 继承ExceptionHandlerAdapter&＃xff0c;或者实现 Thread.UncaughtExceptionHandler 接口
(2) 添加异常处理插件&＃xff0c;Broker提供默认的ExceptionHandlerAdapter实现
new ServerLauncher().exceptionHandler(new ExceptionHandlerDemo()).launch();
依赖注入
由于Listener是由TTIot接管的&＃xff0c;在Listener内部系统提供了两种级别的注入方式&＃xff0c;选择需要注入的对象添加&＃64;Inject注解即可
(1)service 注入
&＃64;Listener(asynchronous &＃61; true)
public class LoginLogListener extends MqttApplicationListener {
&＃64;Inject
private SessionService sessionService;
}
(20 dao 注入
public class DeviceService{
&＃64;Inject
private DbHelper dbHelper;
}
自定义消息处理器
TTIot 提供&＃64;Listener 注解来处理具体的事件信息&＃xff0c;在&＃64;Listener里面有两个参数
replace 为true时&＃xff0c;替换该事件的默认处理器&＃xff0c;为false时&＃xff0c;对该事件新增一个处理器
asynchronous 为true时&＃xff0c;申明该处理器为一个异步处理器&＃xff0c;由TTIot内置的线程池进行接管&＃xff0c;线程池大小由TTiot:netty:bossGroupCount 决定&＃xff0c;为false时&＃xff0c;由netty 的EventLoopGroup 接管&＃xff1b;
处理器可以用来实现额外的功能&＃xff0c;并且跟默认业务解耦&＃xff0c;例如&＃xff0c;异步的设备登陆日志
&＃64;Listener(asynchronous &＃61; true)
public class LoginLogListener extends MqttApplicationListener {
&＃64;Override
public void onApplicationEvent(MqttConnectEvent mqttConnectEvent) {
DbDemo.saveLogin(mqttConnectEvent.getTimestamp(), mqttConnectEvent.getDevName(), mqttConnectEvent.getTenantId());
}
}
无侵入设计
如果开发者想自定义事件的默认处理逻辑&＃xff0c;无需修改源码&＃xff0c;只需要加上如下注解上即可&＃xff0c;TTot则会更换默认的处理逻辑&＃xff0c;开发者也可以利用TTot的事件驱动&＃xff0c;自定义事件
&＃64;Listener(replace &＃61; true)
事件发布
TTIot 提供多种默认的事件供开发者使用&＃xff0c;给topic推送消息&＃xff0c;例如
Context.me().publishEvent(new MqttPublishTopicEvent(String topic,MqttQoS mqttQoS,byte[] bytes,String tenantId,boolean isRetain);
例如&＃xff0c;给device单独推送消息
Context.me().publishEvent(new MqttPublishDevEvent(String clientId, String topic, byte[] byteBuf, MqttQoS qoS,String tenantId);
docker-compose
首先&＃xff0c;需要安装git与docker
git clone https://gitee.com/cloudSj/TTIot.git
cd 到项目根目录
mvn clean install
mvn clean install package -Dmaven.test.skip
docker-compose up -d
webSocket接入
TTIOT内置了webSocket协议转换器&＃xff0c;开发者只需要关注业务逻辑&＃xff0c;无需关注协议处理、转换层面的相关问题。
HTTP接入及设备控制
TTIOT目前只接受POST类型的请求&＃xff0c;格式如下:
URI:/mqtt
Headers:userName //设备帐号
tenantId //租户编号
password //设备密码
Body:
{
"mqttQos": 1, //消息质量
"payload": { //消息体
"rule": "chat",
"content": "hello"
},
"retain": false, //是否为retain消息
"topic": "/m" //推送的话题
}
HTTP由于协议本身的特性只支持QOS类型为1或2
规则路由&＃64;Ruler
如果消息的内容符合
{
"rule": "chat",
"content": "hello"
}
消息格式&＃xff0c;TTIOT将自动触发规则路由器&＃xff0c;将消息自动投递到与rule值对应的处理器上如&＃xff1a;
&＃64;Ruler(name &＃61; "chat")
public class ChatRuleHandler implements RuleHandler {
&＃64;Override
public void handler(MqttPayloadVo mqttPayloadVo) {
System.out.println(mqttPayloadVo);
}
}
因此&＃xff0c;开发者可以自定义业务处理逻辑
TTIoT 开源计划
docker部署组件
HTTP组件以及控制设备HTTPAPI
物模型组件
网关\集群与分布式扩展组件
安卓端组件
规则引擎组件(后续将会持续更新)
SSL加密传输
另外&＃xff1a;作者诚邀开发人员提交下位机例如PLC/单片机/安卓相关代码&＃xff0c;如有计划者可以通过邮箱跟作者取得联系
联系作者
关注公众号:(不定期推送关于IoT的干货以及源码解析)
感谢&＃xff1a;lombok、netty、hutool 提供的工具以及源码
参与贡献
Fork项目到自己的repo
clone到本地
修改代码(dev分支)
commit后push到自己的库(dev分支)
pull request
等待作者合并
京公网安备 11010802041100号