Jetlinks 物联网平台社区版源码学习分析

默认实现是org.jetlinks.supports.protocol.management.ClusterProtocolSupportManager(用Spring BeanPostProcessor hook自动加载 2.0是org.jetlinks.community.protocol.configuration.ProtocolAutoConfiguration配置类加载的)注意: 注册

七海总督

1044人浏览 · 2025-12-17 18:44:33

七海总督 · 2025-12-17 18:44:33 发布

设备接入

设备接入流程图

device-flow.fd9a8a41

网络 > 协议 > 网关

网络组件 (org.jetlinks.community.network.Network)

真正与设备连接交互的网络层, 用于管理各种网络服务(MQTT,TCP等),动态配置, 启停. 只负责接收/发送报文,不负责任何处理逻辑。

社区版, 网络组件的实现有四类:

org.jetlinks.community.network.tcp.server.TcpServer // 作为服务器接受设备端连接

org.jetlinks.community.network.tcp.client.TcpClient // 主动tcp连接设备端

org.jetlinks.community.network.mqtt.client.MqttClient //使用客户端连接第三方的MQTT服务器

org.jetlinks.community.network.mqtt.server.MqttServer //使用的本机MQTT服务, 接受设备端连接

网络组件, 支持提供关键的两个接口

org.jetlinks.community.network.Network

public interface Network {

/**

* ID唯一标识

* @return ID

String getId();

/**

* @return 网络类型

* @see DefaultNetworkType

NetworkType getType();

/**

* 关闭网络组件

void shutdown();

/**

* @return 是否存活

boolean isAlive();

/**

* 当{@link Network#isAlive()}为false是,是否自动重新加载.

* @return 是否重新加载

* @see NetworkProvider#reload(Network, Object)

boolean isAutoReload();

}

org.jetlinks.community.network.NetworkProvider

public interface NetworkProvider<P> {

/**

* @return 类型

* @see DefaultNetworkType

@Nonnull

NetworkType getType();

/**

* 使用配置创建一个网络组件

* @param properties 配置信息

* @return 网络组件

@Nonnull

Network createNetwork(@Nonnull P properties);

/**

* 重新加载网络组件

* @param network 网络组件

* @param properties 配置信息

void reload(@Nonnull Network network, @Nonnull P properties);

/**

* @return 配置定义元数据

@Nullable

ConfigMetadata getConfigMetadata();

/**

* 根据可序列化的配置信息创建网络组件配置

* @param properties 原始配置信息

* @return 网络配置信息

@Nonnull

Mono<P> createConfig(@Nonnull NetworkProperties properties);

...

每一个网络组件(org.jetlinks.community.network.Network) 对应有一个组件提供器对应 (org.jetlinks.community.network.NetworkProvider)

最终网络组件统一由 org.jetlinks.community.network.NetworkManager 管理;

默认实现是org.jetlinks.community.network.DefaultNetworkManager(用Spring BeanPostProcessor hook 加载的)

调用其org.jetlinks.community.network.DefaultNetworkManager#register方法, 传递 NetworkProvider 可以注册一个网络组件

实例组件数据是存在数据库的 network_config 表

协议相关 (org.jetlinks.core.ProtocolSupport)

用于自定义消息解析规则,用于认证、将设备发送给平台报文解析为平台统一的报文，以及处理平台下发给设备的指令。

协议(org.jetlinks.core.ProtocolSupport)主要由: 认证器(Authenticator), 消息编解码器(DeviceMessageCodec),消息发送拦截器(DeviceMessageSenderInterceptor) 以及配置元数据(ConfigMetadata)组成.

org.jetlinks.core.defaults.Authenticator // Authenticator

org.jetlinks.core.codec.defaults.DeviceMessageCodec //DeviceMessageCodec

org.jetlinks.core.message.interceptor.DeviceMessageSenderInterceptor //DeviceMessageSenderInterceptor

org.jetlinks.core.ProtocolSupport + org.jetlinks.core.metadata.DeviceMetadataCodec //ConfigMetadata

其默认自带的JetLinks V1.0 协议,在org.jetlinks.supports.official.JetLinksProtocolSupportProvider 提供

每一个协议(org.jetlinks.core.ProtocolSupport) 对应有一个组件提供器对应 (org.jetlinks.core.spi.ProtocolSupportProvider)

自定义协议, 即实现 org.jetlinks.core.spi.ProtocolSupportProvider 这个接口;

统一由org.jetlinks.supports.protocol.management.ProtocolSupportManager 管理;

调用其org.jetlinks.supports.protocol.management.ClusterProtocolSupportManager#store方法传递 ProtocolSupportDefinition 进行注册;

注意: 注册后只存起来, 顾名思义, 还跟集群的其他节点有关, 而且跟协议的实现类型(jar,js)有关, 加载也不一样; 参考 org.jetlinks.supports.protocol.management.ProtocolSupportDefinition的属性

public class ProtocolSupportDefinition implements Serializable {

private static final long serialVersionUID = -1;

private String id;//ID ; 数据库存的就是本地对象的数据

private String name;

private String description;

private String provider;//jar script 协议实现类型, 目前只有 jar, js 脚本

private byte state;//协议状态

private Map<String,Object> configuration;//配置元数据, jar 的话会存ProtocolSupportProvider类全限定名, jar包路径等

}

实例组件数据是存在数据库 dev_protocol 表

参考下: [自定义协议开发]

网关组件 (org.jetlinks.community.gateway.DeviceGateway)

设备上报数据的处理逻辑入口, 网关代表接入方式需要选择网络组件, 它关联协议, 配置协议

负责平台侧统一的设备接入, 使用网络组件处理对应的请求以及报文, 使用配置的协议解析为平台统一的设备消息(DeviceMessage),然后推送到事件总线。

org.jetlinks.community.gateway.supports.DeviceGateway 网关接口抽象;

设备网关支持提供商,用于提供对各种设备网关的支持.在启动设备网关时,会根据对应的提供商以及配置来创建设备网关. 实现统一管理网关配置,动态创建设备网关.

社区版, 网关的实现有三个

在 mqtt-component 项目有两个

org.jetlinks.community.network.mqtt.gateway.device.MqttClientDeviceGateway

使用MQTT客户端, 连接第三方MQTT服务器; 例如 emqx

org.jetlinks.community.network.mqtt.gateway.device.MqttServerDeviceGateway

本机作为MQTT服务端, 内置启用MQTT服务器, 接受设备接入;

在 tcp-component 项目有一个

org.jetlinks.community.network.tcp.device.TcpServerDeviceGateway

本机作为TCP服务器, 监听端口, 接受设备接入;

类似桥接网络组件和协议组件(作为一个中介者)

统一由org.jetlinks.community.gateway.DeviceGatewayManager管理;

默认实现是 org.jetlinks.community.gateway.supports.DefaultDeviceGatewayManager(用Spring BeanPostProcessor hook自动加载; 2.0是org.jetlinks.community.gateway.GatewayConfiguration配置类加载的)

调用其org.jetlinks.community.gateway.DeviceGatewayManager#start传递一个网关实例ID 启动网关; 其实是改变了一下网关实例的状态, 在有新消息时根据自身状态决定是否分发消息, 在此之前已经调了协议解析, 所有暂停/停止网关不会影响协议跟设备的交互;

参考: org.jetlinks.community.network.mqtt.gateway.device.MqttClientDeviceGateway 这个网关实现

实例组件是存在数据库 device_gateway 表

关键的对象概览

DeviceRegistry 设备注册中心

org.jetlinks.core.device.DeviceRegistry

设备注册中心, 用于统一管理设备以及产品的基本信息,缓存,进行设备指令下发等操作.

例如: 获取设备以及设备的配置缓存信息

registry

.getDevice(deviceId)

.flatMap(device->device.getSelfConfig("my-config"))

.flatMap(conf-> doSomeThing(...))

相当于总的设备管理器

根据设备ID. 通过 org.jetlinks.core.device.DeviceRegistry#getDevice 可以获取设备操作对象(DeviceOperator)

以 org.jetlinks.supports.cluster.ClusterDeviceRegistry#getDevice 为例:

public Mono<DeviceOperator> getDevice(String deviceId) {

if (StringUtils.isEmpty(deviceId)) {

return Mono.empty();

} else {

//先从缓存获取

Mono<DeviceOperator> deviceOperator = (Mono)this.operatorCache.getIfPresent(deviceId);

if (null != deviceOperator) {

return deviceOperator;

} else {

//创建 DeviceOperator

DeviceOperator deviceOperator = this.createOperator(deviceId);

return deviceOperator.getSelfConfig(DeviceConfigKey.productId).doOnNext((r) -> {

//放到缓存

this.operatorCache.put(deviceId, Mono.just(deviceOperator).filterWhen((device) -> {

return device.getSelfConfig(DeviceConfigKey.productId).hasElement();

}));

}).map((ignore) -> {

return deviceOperator;

});

}

private DefaultDeviceOperator createOperator(String deviceId) {

DefaultDeviceOperator device = new DefaultDeviceOperator(deviceId, this.supports, this.manager, this.handler, this, this.interceptor, this.stateChecker);

if (this.rpcChain != null) {

device.setRpcChain(this.rpcChain);

}

return device;

}

DeviceOperator 设备操作接口

org.jetlinks.core.device.DeviceOperator

设备操作接口,通过DeviceRegister.getDevice(deviceId)获取,用于对设备进行相关操作,如获取配置,发送消息等.

//通过getConfig 可以获取配置的协议元数据

DeviceOperator#getConfig

它是如何创建的?

org.jetlinks.core.defaults.DefaultDeviceOperator为例

从注册中心拿的时候, 如果不存在, 则创建DeviceOperator

@Override

public Mono<DeviceOperator> getDevice(String deviceId) {

if (StringUtils.isEmpty(deviceId)) {

return Mono.empty();

}

{

Mono<DeviceOperator> deviceOperator = operatorCache.getIfPresent(deviceId);

if (null != deviceOperator) {

return deviceOperator;

}

//创建 DeviceOperator

DeviceOperator deviceOperator = createOperator(deviceId);

return deviceOperator

//有productId说明是存在的设备

.getSelfConfig(DeviceConfigKey.productId)

.doOnNext(r -> operatorCache.put(deviceId, Mono

.just(deviceOperator)

.filterWhen(device -> device.getSelfConfig(DeviceConfigKey.productId).hasElement())

//设备被注销了？则移除之

.switchIfEmpty(Mono.fromRunnable(() -> operatorCache.invalidate(deviceId)))

))

.map(ignore -> deviceOperator);

}

DeviceProductOperator 产品操作接口

DeviceProductOperator: 产品操作接口,通过DeviceProductOperator.getProduct(productId)获取.

DeviceGateway 设备接入网关接口

DeviceGateway : 设备接入网关接口,利用网络组件来接入设备消息.

DeviceMessageBusinessHandler

DeviceMessageBusinessHandler: 处理设备状态数据库同步,设备自动注册等逻辑等类.

LocalDeviceInstanceService

LocalDeviceInstanceService: 设备实例管理服务类.

DeviceSessionManager

DeviceSessionManager: 设备会话管理器,可获取当前服务的会话信息.

管理会话, 设备的上线下线;

参考: org.jetlinks.community.standalone.configuration.DefaultDeviceSessionManager

在 init

Flux.interval(Duration.ofSeconds(10), Duration.ofSeconds(30), Schedulers.newSingle("device-session-checker"))

.flatMap(i -> this

.checkSession()//周期性调用 checkSession() 检查状态

.onErrorContinue((err, val) -> log.error(err.getMessage(), err)))

.subscribe();

DeviceDataStoragePolicy 设备数据存储策略(行式/列式/不存储)

DeviceDataStoragePolicy: 设备存储策略接口,实现此接口来进行自定义设备数据存储策略.

DeviceGatewayHelper

DeviceGatewayHelper: 统一处理设备消息,创建Session等操作的逻辑.

DecodedClientMessageHandler

DecodedClientMessageHandler: 解码后的平台消息处理器，如果是自定义实现网关或者在协议包里手动回复消息等处理, 则可以使用此接口直接将设备消息交给平台.(如果调用了DeviceGatewayHelper则不需要此操作).

EventBus 事件总线

EventBus: 事件总线,通过事件总线去订阅设备数据来实现解耦.(也可以用过@Subscribe()注解订阅).

DeviceMessageConnector

DeviceMessageConnector: 负责将设备消息转发到事件总线.

DeviceMessageSender 消息发送器

org.jetlinks.core.device.DeviceMessageSender

消息发送器,用于发送消息给设备.

DeviceMessage 消息对象

所有设备消息(即设备上报转换后, 平台可识别的消息) 派生自这个 org.jetlinks.core.message.DeviceMessage 接口

EncodedMessage 消息对象

设备端原始的消息, (下发/上报给的原始消息)

源码大体流程研究

先了解消息的组成

消息主要由 deviceId, messageId, headers, timestamp 组成.

deviceId为设备的唯一标识, messageId为消息的唯一标识,headers为消息头,通常用于对自定义消息处理的行为,如是否异步消息, 是否分片消息等.

常用的Headers org.jetlinks.core.message.Headers

async 是否异步,boolean类型.

timeout 指定超时时间. 毫秒.

frag_msg_id 分片主消息ID,为下发消息的messageId

frag_num 分片总数

frag_part 当前分片索引

frag_last 是否为最后一个分片,当无法确定分片数量的时候,可以将分片设置到足够大,最后一个分片设置:frag_last=true来完成返回.

keepOnline 与DeviceOnlineMessage配合使用,在TCP短链接,保持设备一直在线状态,连接断开不会设置设备离线.

keepOnlineTimeoutSeconds 指定在线超时时间,在短链接时,如果超过此间隔没有收到消息则认为设备离线.

ignoreStorage 不存储此消息数据,如: 读写属性回复默认也会记录到属性时序数据库中,设置为true后,将不记录.(1.9版本后支持)

ignoreLog 不记录此消息到日志,如: 设置为true,将不记录此消息的日志.

mergeLatest 是否合并最新属性数据,设置此消息头后,将会把最新的消息合并到消息体里( 需要开启最新数据存储)//jetlinks.device.storage.enable-last-data-in-db=true 是否将设备最新到数据存储到数据库

网络组件(Network) 真正与设备连接交互的网络层, 但是它只负责接收/发送报文,不负责任何处理逻辑, 所以最佳的调试地方是网关组件(DeviceGateway)入口处.

消息类型

所有消息类型参考 org.jetlinks.community.device.enums.DeviceLogType

属性相关消息

获取设备属性(ReadPropertyMessage)对应设备回复的消息 ReadPropertyMessageReply.

修改设备属性(WritePropertyMessage)对应设备回复的消息 WritePropertyMessageReply.

设备上报属性(ReportPropertyMessage) 由设备上报.

功能相关消息

调用设备功能到消息(FunctionInvokeMessage)由平台发往设备,对应到返回消息 FunctionInvokeMessageReply.

事件消息

org.jetlinks.core.message.event.EventMessage

EventMessage eventMessage = new EventMessage();

eventMessage.setDeviceId(deviceId);

eventMessage.setMessageId(fromDevice.path(MessageConstant.MESSAGE_KEY_MESSAGE_SIGN).asText() );

eventMessage.event(eventId);

HashMap data = JacksonUtils.jsonToBean(output.toString(), HashMap.class);

eventMessage.setData(data);

其他消息

DeviceOnlineMessage 设备上线消息,通常用于网关代理的子设备的上线操作.

DeviceOfflineMessage 设备离线消息,通常用于网关代理的子设备的下线操作.

ChildDeviceMessage 子设备消息,通常用于网关代理的子设备的消息.

ChildDeviceMessageReply 子设备消息回复,用于平台向网关代理的子设备发送消息后设备回复给平台的结果.

UpdateTagMessage 更新设备标签.

DerivedMetadataMessage 更新设备独立物模型.

设备自注册消息

DeviceRegisterMessage 设备注册消息,通过设置消息头message.addHeader("deviceName","设备名称");和 message.addHeader("productId","产品ID")可实现设备自动注册.

如果配置了状态自管理,在检查子设备状态时,会发送指令ChildDeviceMessage<DeviceStateCheckMessage>, 网关需要返回ChildDeviceMessageReply<DeviceStateCheckMessageReply>.

自定义协议包将消息解析为 DeviceRegisterMessage,

并设置header:productId(必选),deviceName(必选),configuration(可选)。

平台将自动添加设备信息到设备实例中。如果是注册子设备,则解析为 ChildDeviceMessage<DeviceRegisterMessage>即可

消息上报 (MQTT Broker)

设备不是直接接入平台, 而是通过第三方MQTT服务, 如:emqx. 消息编解码与MQTT服务一样,从消息协议中使用 DefaultTransport.MQTT 来获取消息编解码器.

网关处理逻辑

org.jetlinks.community.network.mqtt.gateway.device.MqttClientDeviceGateway

public class MqttClientDeviceGateway extends AbstractDeviceGateway {

public MqttClientDeviceGateway(String id,

MqttClient mqttClient,

DeviceRegistry registry,

ProtocolSupports protocolSupport,

String protocol,

DeviceSessionManager sessionManager,

DecodedClientMessageHandler clientMessageHandler,

List<String> topics,

int qos) {

super(id);

// mqtt的客户端

this.mqttClient = Objects.requireNonNull(mqttClient, "mqttClient");

//DeviceRegistry : 设备注册中心, 用于统一管理设备以及产品的基本信息,缓存,进行设备指令下发等操作(约等于设备统一管理器)

this.registry = Objects.requireNonNull(registry, "registry");

this.protocolSupport = Objects.requireNonNull(protocolSupport, "protocolSupport");

this.protocol = Objects.requireNonNull(protocol, "protocol");

this.topics = Objects.requireNonNull(topics, "topics");

this.helper = new DeviceGatewayHelper(registry, sessionManager, clientMessageHandler);

this.qos = qos;

}

private void doStart() {

if (disposable != null) {

disposable.dispose();

}

disposable = mqttClient

.subscribe(topics, qos)//关注MQTT主题, 当有新的消息时

.filter((msg) -> isStarted())//需当前网关处于启动状态

.flatMap(mqttMessage -> {

AtomicReference<Duration> timeoutRef = new AtomicReference<>();

return this

//注意这里是根据自定义协议 ProtocolSupportProvider::create 返回的 ProtocolSupport id 去匹配的

.getProtocol()

//通过 ProtocolSupport 获取其 DeviceMessageCodec

.flatMap(codec -> codec.getMessageCodec(getTransport()))

//使用消息编码器解码消息

.flatMapMany(codec -> codec.decode(FromDeviceMessageContext.of(

new UnknownDeviceMqttClientSession(getId() + ":unknown", mqttClient) {//实际给到协议编解码器的Session 总是这个..

@Override

public Mono<Boolean> send(EncodedMessage encodedMessage) {

return super

.send(encodedMessage)

.doOnSuccess(r -> monitor.sentMessage());

}

@Override

public void setKeepAliveTimeout(Duration timeout) {

timeoutRef.set(timeout);

}

, mqttMessage, registry)

))

.doOnError((err) -> log.error("解码MQTT客户端消息失败 {}:{}", //发生了错误

mqttMessage.getTopic(),

mqttMessage

.getPayload()

.toString(StandardCharsets.UTF_8),

err))

//消息向上转型

.cast(DeviceMessage.class)

.flatMap(message -> {

//设备网关监控 (主要是监控消息数量等指标的)

monitor.receivedMessage();

return helper//设备网关消息处理,会话管理工具类,用于统一封装对设备消息和会话的处理逻辑

.handleDeviceMessage(message,//最终主要源码见下: org.jetlinks.community.network.utils.DeviceGatewayHelper#handleDeviceMessage

device -> createDeviceSession(device, mqttClient),//<!> 注意这个会话在不存在时回调; 见下: [创建会话的回调]

ignore->{},//会话自定义回调,处理会话时用来自定义会话,比如重置连接信

() -> log.warn("无法从MQTT[{}]消息中获取设备信息:{}", mqttMessage.print(), message)//当设备在平台不存在时

);

})

.then()

//错误处理, 返回 empty

.onErrorResume((err) -> {

log.error("处理MQTT消息失败:{}", mqttMessage, err);

return Mono.empty();

});

}, Integer.MAX_VALUE)

.onErrorContinue((err, ms) -> log.error("处理MQTT客户端消息失败", err))

.subscribe();//Flux的API 触发计算

}

创建会话的回调

private MqttClientSession createDeviceSession(DeviceOperator device, MqttClient client) {

return new MqttClientSession(device.getDeviceId(), device, client, monitor);

}

DeviceGatewayHelper 处理

主要处理消息分支子设备分支, 上线/离线处理....

//如果设备状态为'离线' 则会先构造一个设备上线的消息 publish, 然后在 publish 设备原消息

org.jetlinks.community.network.utils.DeviceGatewayHelper#handleDeviceMessage

魔乐社区

魔乐社区（Modelers.cn) 是一个中立、公益的人工智能社区，提供人工智能工具、模型、数据的托管、展示与应用协同服务，为人工智能开发及爱好者搭建开放的学习交流平台。社区通过理事会方式运作，由全产业链共同建设、共同运营、共同享有，推动国产AI生态繁荣发展。

更多推荐

替你试过了，消费级显卡可以跑的开源文生图SOTA模型，顶级渲染、高密度文本绘图

魔乐社区

量化挑战赛冠军专访：4小时啃下W4A8量化，我靠的是这些经验

魔乐社区

小参数・大码力・易部署 | Qwen3.6-27B上线魔乐社区，基于昇腾的部署教程来了

继一周前模型开源发布后，千问再度开源Qwen3.6-27B —— 一个拥有270亿参数的稠密多模态模型，也是社区呼声最高的模型规格。Qwen3.6-27B 依然支持多模态思考与非思考模式，在智能体编程方面达到了旗舰级表现，全面超越前代开源旗舰 Qwen3.5-397B-A17B（总参数397B / 激活参数17B的MoE模型）。作为稠密架构，它无需MoE路由即可部署，是开发者在实用、可广泛部署规模