本文基于thingsboard-gateway的早期java版本，现在thingsboard-gateway已改为python语言，下图为thingsboard-gateway配置示意图：
一切缘起remoteConfiguration这个配置，当remoteConfiguration设置为true时网关就会向thingsboard（下面简称tb）发送获取设备共享属性configuration的请求，如果tb端针对该网关有configuration这个共享属性，那么就会将这个配置下发到gateway,gateway就会使用下发的配置创建对应的服务（比如mqtt、opc、modbus等），如果remoteConfiguration配置为false，那么gateway就会使用extensions部分配置的扩展进行初始化，下面是DefaultTenantManagerService初始化相关代码：
代码语言：javascript复制 @PostConstruct
public void init() {
gateways = new HashMap<>();
httpServices = new ArrayList<>();
for (TbTenantConfiguration configuration : configuration.getTenants()) {
isRemoteConfiguration = configuration.getRemoteConfiguration();
if (isRemoteConfiguration) {
String label = configuration.getLabel();
log.info("[{}] Initializing gateway", configuration.getLabel());
TenantServiceRegistry tenantServiceRegistry = new TenantServiceRegistry();
GatewayService service = null;
try {
// isRemoteConfiguration为true时，会响应服务端返回的配置信息
service = getGatewayService(configuration, c -> tenantServiceRegistry.updateExtensionConfiguration(c));
tenantServiceRegistry.setService(service);
gateways.put(label, tenantServiceRegistry);
} catch (Exception e) {
log.info("[{}] Failed to initialize the service ", label, e);
try {
if (service != null) {
service.destroy();
}
} catch (Exception exc) {
log.info("[{}] Failed to stop the service ", label, exc);
}
}
} else {
String label = configuration.getLabel();
log.info("[{}] Initializing gateway", configuration.getLabel());
GatewayService service = null;
try {
TenantServiceRegistry tenantServiceRegistry = new TenantServiceRegistry();
// isRemoteConfiguration为false时，会忽略服务端返回的配置信息
service = getGatewayService(configuration, c -> {});
tenantServiceRegistry.setService(service);
for (TbExtensionConfiguration extensionConfiguration : configuration.getExtensions()) {
log.info("[{}] Initializing extension: [{}]", configuration.getLabel(), extensionConfiguration.getType());
ExtensionService extension = tenantServiceRegistry.createExtensionServiceByType(service, extensionConfiguration.getType());
extension.init(extensionConfiguration, isRemoteConfiguration);
if (extensionConfiguration.getType().equals("HTTP")) {
httpServices.add((HttpService) extension);
}
}
gateways.put(label, (TenantServiceRegistry) tenantServiceRegistry);
} catch (Exception e) {
log.info("[{}] Failed to initialize the service ", label, e);
try {
if (service != null) {
service.destroy();
}
} catch (Exception exc) {
log.info("[{}] Failed to stop the service ", label, exc);
}
}
}
}
}gateway通过mqtt连接上tb后，会向tb发送获取configuration配置的rpc请求，具体在MqttGatewayService的initMqttClient方法中：
代码语言：javascript复制 private MqttClient initMqttClient() {
try {
MqttClientConfig mqttClientConfig = getMqttClientConfig();
mqttClientConfig.setUsername(connection.getSecurity().getAccessToken());
tbClient = MqttClient.create(mqttClientConfig, this);
tbClient.setCallback(this);
tbClient.setEventLoop(nioEventLoopGroup);
Promise<MqttConnectResult> connectResult = (Promise<MqttConnectResult>) tbClient.connect(connection.getHost(), connection.getPort());
connectResult.addListener(future -> {
if (future.isSuccess()) {
MqttConnectResult result = (MqttConnectResult) future.getNow();
log.debug("Gateway connect result code: [{}]", result.getReturnCode());
} else {
log.error("Unable to connect to mqtt server!");
if (future.cause() != null) {
log.error(future.cause().getMessage(), future.cause());
}
}
});
connectResult.get(connection.getConnectionTimeout(), TimeUnit.MILLISECONDS);
tbClient.on(DEVICE_ATTRIBUTES_TOPIC, this).await(connection.getConnectionTimeout(), TimeUnit.MILLISECONDS);
tbClient.on(DEVICE_GET_ATTRIBUTES_RESPONSE_PLUS_TOPIC, this).await(connection.getConnectionTimeout(), TimeUnit.MILLISECONDS);
tbClient.on(GATEWAY_RESPONSES_ATTRIBUTES_TOPIC, this).await(connection.getConnectionTimeout(), TimeUnit.MILLISECONDS);
tbClient.on(GATEWAY_ATTRIBUTES_TOPIC, this).await(connection.getConnectionTimeout(), TimeUnit.MILLISECONDS);
tbClient.on(GATEWAY_RPC_TOPIC, this).await(connection.getConnectionTimeout(), TimeUnit.MILLISECONDS);
// 向tb索取该网关对应的configuration配置，注意一定是共享属性
byte[] msgData = toBytes(newNode().put("sharedKeys", "configuration"));
persistMessage(DEVICE_GET_ATTRIBUTES_REQUEST_TOPIC, msgIdSeq.incrementAndGet(), msgData, null,
null,
error -> log.warn("Error getiing attributes", error));
return tbClient;
} catch (InterruptedException e) {
log.error(e.getMessage(), e);
Thread.currentThread().interrupt();
return null;
} catch (ExecutionException e) {
log.error(e.getMessage(), e);
throw new RuntimeException(e);
} catch (TimeoutException e) {
String message = "Unable to connect to ThingsBoard. Connection timed out after [" connection.getConnectionTimeout() "] milliseconds";
log.error(message, e);
throw new RuntimeException(message);
}
}共享属性查看页面
PS：
java版本thingsboard-gateway可以通过https://gitee.com/johnHust/thingsboard-gateway下载
