四川有那些网站建设公司黄页推广app软件

四川有那些网站建设公司,黄页推广app软件,网站建设怎么做呢,通过网站做外贸设备接入设备接入流程图device-flow.fd9a8a41网络 协议 网关网络组件 (org.jetlinks.community.network.Network)真正与设备连接交互的网络层, 用于管理各种网络服务(MQTT,TCP等),动态配置, 启停. 只负责接收/发送报文,不负责任何处理逻辑。社区版, 网络组件的实现有…设备接入设备接入流程图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.Networkpublic 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.NetworkProviderpublic interface NetworkProviderP {/*** return 类型* see DefaultNetworkType*/NonnullNetworkType getType();/*** 使用配置创建一个网络组件* param properties 配置信息* return 网络组件*/NonnullNetwork createNetwork(Nonnull P properties);/*** 重新加载网络组件* param network 网络组件* param properties 配置信息*/void reload(Nonnull Network network, Nonnull P properties);/*** return 配置定义元数据*/NullableConfigMetadata getConfigMetadata();/*** 根据可序列化的配置信息创建网络组件配置* param properties 原始配置信息* return 网络配置信息*/NonnullMonoP 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 // Authenticatororg.jetlinks.core.codec.defaults.DeviceMessageCodec //DeviceMessageCodecorg.jetlinks.core.message.interceptor.DeviceMessageSenderInterceptor //DeviceMessageSenderInterceptororg.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(用Spring BeanPostProcessor hook自动加载 2.0是org.jetlinks.community.protocol.configuration.ProtocolAutoConfiguration配置类加载的)调用其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 MapString,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服务器; 例如 emqxorg.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 MonoDeviceOperator getDevice(String deviceId) {if (StringUtils.isEmpty(deviceId)) {return Mono.empty();} else {//先从缓存获取MonoDeviceOperator deviceOperator (Mono)this.operatorCache.getIfPresent(deviceId);if (null ! deviceOperator) {return deviceOperator;} else {//创建 DeviceOperatorDeviceOperator 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为例从注册中心拿的时候, 如果不存在, 则创建DeviceOperatorOverridepublic MonoDeviceOperator getDevice(String deviceId) {if (StringUtils.isEmpty(deviceId)) {return Mono.empty();}{MonoDeviceOperator deviceOperator operatorCache.getIfPresent(deviceId);if (null ! deviceOperator) {return deviceOperator;}}//创建 DeviceOperatorDeviceOperator 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 : 设备接入网关接口,利用网络组件来接入设备消息.DeviceMessageBusinessHandlerDeviceMessageBusinessHandler: 处理设备状态数据库同步,设备自动注册等逻辑等类.LocalDeviceInstanceServiceLocalDeviceInstanceService: 设备实例管理服务类.DeviceSessionManagerDeviceSessionManager: 设备会话管理器,可获取当前服务的会话信息.管理会话, 设备的上线下线;参考: org.jetlinks.community.standalone.configuration.DefaultDeviceSessionManager在 initFlux.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: 设备存储策略接口,实现此接口来进行自定义设备数据存储策略.DeviceGatewayHelperDeviceGatewayHelper: 统一处理设备消息,创建Session等操作的逻辑.DecodedClientMessageHandlerDecodedClientMessageHandler: 解码后的平台消息处理器如果是自定义实现网关或者在协议包里手动回复消息等处理, 则可以使用此接口直接将设备消息交给平台.(如果调用了DeviceGatewayHelper则不需要此操作).EventBus 事件总线EventBus: 事件总线,通过事件总线去订阅设备数据来实现解耦.(也可以用过Subscribe()注解订阅).DeviceMessageConnectorDeviceMessageConnector: 负责将设备消息转发到事件总线.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.Headersasync 是否异步,boolean类型.timeout 指定超时时间. 毫秒.frag_msg_id 分片主消息ID,为下发消息的messageIdfrag_num 分片总数frag_part 当前分片索引frag_last 是否为最后一个分片,当无法确定分片数量的时候,可以将分片设置到足够大,最后一个分片设置:frag_lasttrue来完成返回.keepOnline 与DeviceOnlineMessage配合使用,在TCP短链接,保持设备一直在线状态,连接断开不会设置设备离线.keepOnlineTimeoutSeconds 指定在线超时时间,在短链接时,如果超过此间隔没有收到消息则认为设备离线.ignoreStorage 不存储此消息数据,如: 读写属性回复默认也会记录到属性时序数据库中,设置为true后,将不记录.(1.9版本后支持)ignoreLog 不记录此消息到日志,如: 设置为true,将不记录此消息的日志.mergeLatest 是否合并最新属性数据,设置此消息头后,将会把最新的消息合并到消息体里( 需要开启最新数据存储)//jetlinks.device.storage.enable-last-data-in-dbtrue 是否将设备最新到数据存储到数据库网络组件(Network) 真正与设备连接交互的网络层, 但是它只负责接收/发送报文,不负责任何处理逻辑, 所以最佳的调试地方是网关组件(DeviceGateway)入口处.消息类型所有消息类型参考 org.jetlinks.community.device.enums.DeviceLogType属性相关消息获取设备属性(ReadPropertyMessage)对应设备回复的消息 ReadPropertyMessageReply.修改设备属性(WritePropertyMessage)对应设备回复的消息 WritePropertyMessageReply.设备上报属性(ReportPropertyMessage) 由设备上报.功能相关消息调用设备功能到消息(FunctionInvokeMessage)由平台发往设备,对应到返回消息 FunctionInvokeMessageReply.事件消息org.jetlinks.core.message.event.EventMessageEventMessage 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)可实现设备自动注册.如果配置了状态自管理,在检查子设备状态时,会发送指令ChildDeviceMessageDeviceStateCheckMessage, 网关需要返回ChildDeviceMessageReplyDeviceStateCheckMessageReply.自定义协议包将消息解析为 DeviceRegisterMessage,并设置header:productId(必选),deviceName(必选),configuration(可选)。平台将自动添加设备信息到设备实例中。如果是注册子设备,则解析为 ChildDeviceMessageDeviceRegisterMessage即可消息上报 (MQTT Broker)设备不是直接接入平台, 而是通过第三方MQTT服务, 如:emqx. 消息编解码与MQTT服务一样,从消息协议中使用 DefaultTransport.MQTT 来获取消息编解码器.网关处理逻辑org.jetlinks.community.network.mqtt.gateway.device.MqttClientDeviceGatewaypublic class MqttClientDeviceGateway extends AbstractDeviceGateway {public MqttClientDeviceGateway(String id,MqttClient mqttClient,DeviceRegistry registry,ProtocolSupports protocolSupport,String protocol,DeviceSessionManager sessionManager,DecodedClientMessageHandler clientMessageHandler,ListString 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 - {AtomicReferenceDuration 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 总是这个..Overridepublic MonoBoolean send(EncodedMessage encodedMessage) {return super.send(encodedMessage).doOnSuccess(r - monitor.sentMessage());}Overridepublic 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#handleDeviceMessagedevice - 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