- 该系列文章基于Thingsboard release-3.0分支的源码进行分析，可能与最新的特性有所区别。
初识项目
拉代码
把我们把代码pull下来，打开IDEA，相信大多数人的反应会是，卧槽，这什么项目，怎么这么多模块！
我刚把项目拉下来的时候也是一脸懵逼，完全不知道这么多module是怎么划分的，脸上大写的两个字，卧槽！
整理思路
既来之则安之，虽然是满脸的卧槽，但也只能静下心来慢慢分析了。其实tb的源码模块化做的还是不错的，起码让我写这代码我写不到这程度，不理解还是由于模块之间的依赖关系不清晰导致的。
找官方文档
根据tb的官网我们知道tb主要由一下这几种微服务
tb-http-transport
tb-mqtt-transport
tb-coap-transport
tb-core
tb-rule-engine
tb-js-executor
tb-web-ui
官网链接：[https://thingsboard.io/docs/reference/](https://thingsboard.io/docs/reference/)
从源码里面找到对应的服务
既然都已经知道有哪些服务了，那我只要在这一堆模块中找到对应服务的启动类就完事儿了，然后按照这个逻辑不就一下子就能找到对应服务的代码了嘛。照着这个思路，一番查找之后找到了一下几个服务
tb-http-transport---./transport/http
tb-mqtt-transport---./transport/mqtt
tb-coap-transport---./transport/coap
tb-js-executor---./msa/js-executor
tb-web-ui---./ui-ngx
其中js-executor和web-ui因为是ng和前端写的，所以可以分开来看，但是这几个transport就一个启动类是个什么鬼，而且也没有找到tb-core以及tb-rule-engine，只有ThingsboardInstallApplication和ThingsboardServerApplication这俩莫名其妙的启动类，这时候心里已经开始骂娘了。
虽然没法一下子就搞清楚项目，但起码还是通过几个启动类把服务大致划分了一下，接下来那就只能看一下这几个启动类对应服务的依赖关系了，看看能不能找到一些线索。
分析服务引用
分析transport服务的引用
由于transport是一个比较独立的模块，且不同的transport功能基本一致，所以先从这个开始分析，我先选取了mqtt-transport这个模块进行分析，它的pom文件展示了它的依赖关系
org.thingsboard.common.transport
mqtt
org.thingsboard.common
queue
可以看到这两个主要的依赖，从名字上基本上可以猜到queue肯定是transport的管道服务，因为之前搭建过tb的集群，所以基本猜到里面可能会有与kafka的管道相关的代码，而mqtt这个依赖基本上可以猜到是一个mqttserver服务，接下来就是分析这两个模块的源码。
完整的依赖
从这张图可以大致可以看出mqtt-transport这个服务通过transport-mqtt模块实现一个mqttserver的功能，然后通过queue模块与建立管道服务从而与其他服务进行交互，接下来就是具体看这两个模块的代码。
分析mqtt模块源码
源码路径位于 ./common/transport/mqtt
果然这个模块就是一个用netty封装的mqttserver，而且还通过配置文件的条件来判断是否要初始化这个mqttserver服务。
package org.thingsboard.server.transport.mqtt;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.util.ResourceLeakDetector;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Service;
import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
/**
* @author Andrew Shvayka
*/
@Service("MqttTransportService")
@ConditionalOnExpression("'${service.type:null}'=='tb-transport' || ('${service.type:null}'=='monolith' && '${transport.mqtt.enabled}'=='true')")
@Slf4j
public class MqttTransportService {
@Value("${transport.mqtt.bind_address}")
private String host;
@Value("${transport.mqtt.bind_port}")
private Integer port;
@Value("${transport.mqtt.netty.leak_detector_level}")
private String leakDetectorLevel;
@Value("${transport.mqtt.netty.boss_group_thread_count}")
private Integer bossGroupThreadCount;
@Value("${transport.mqtt.netty.worker_group_thread_count}")
private Integer workerGroupThreadCount;
@Value("${transport.mqtt.netty.so_keep_alive}")
private boolean keepAlive;
@Autowired
private MqttTransportContext context;
private Channel serverChannel;
private EventLoopGroup bossGroup;
private EventLoopGroup workerGroup;
@PostConstruct
public void init() throws Exception {
log.info("Setting resource leak detector level to {}", leakDetectorLevel);
ResourceLeakDetector.setLevel(ResourceLeakDetector.Level.valueOf(leakDetectorLevel.toUpperCase()));
log.info("Starting MQTT transport...");
bossGroup = new NioEventLoopGroup(bossGroupThreadCount);
workerGroup = new NioEventLoopGroup(workerGroupThreadCount);
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.childHandler(new MqttTransportServerInitializer(context))
.childOption(ChannelOption.SO_KEEPALIVE, keepAlive);
serverChannel = b.bind(host, port).sync().channel();
log.info("Mqtt transport started!");
}
@PreDestroy
public void shutdown() throws InterruptedException {
log.info("Stopping MQTT transport!");
try {
serverChannel.close().sync();
} finally {
workerGroup.shutdownGracefully();
bossGroup.shutdownGracefully();
}
log.info("MQTT transport stopped!");
}
}
对于熟悉netty的朋友来说这个就比较简单了，就是根据handler来处理接收到的消息，这个其实就没什么好多说了。不熟悉netty的朋友可以自己去百度一下就好了，就是简单的接收到消息之后丢给org.thingsboard.server.transport.mqtt.MqttTransportHandler这个类来处理而已，具体的消息处理则都是通过handler来进行。
分析MqttTransportHandler
消息都是从channelRead方法进入到handler，然后丢到processMqttMsg这个方法进行具体的分类处理。
private void processMqttMsg(ChannelHandlerContext ctx, MqttMessage msg) {
address = (InetSocketAddress) ctx.channel().remoteAddress();
if (msg.fixedHeader() == null) {
log.info("[{}:{}] Invalid message received", address.getHostName(), address.getPort());
processDisconnect(ctx);
return;
}
deviceSessionCtx.setChannel(ctx);
switch (msg.fixedHeader().messageType()) {
case CONNECT:
processConnect(ctx, (MqttConnectMessage) msg);
break;
case PUBLISH:
processPublish(ctx, (MqttPublishMessage) msg);
break;
case SUBSCRIBE:
processSubscribe(ctx, (MqttSubscribeMessage) msg);
break;
case UNSUBSCRIBE:
processUnsubscribe(ctx, (MqttUnsubscribeMessage) msg);
break;
case PINGREQ:
if (checkConnected(ctx, msg)) {
ctx.writeAndFlush(new MqttMessage(new MqttFixedHeader(PINGRESP, false, AT_MOST_ONCE, false, 0)));
transportService.reportActivity(sessionInfo);
}
break;
case DISCONNECT:
if (checkConnected(ctx, msg)) {
processDisconnect(ctx);
}
break;
default:
break;
}
}
通过这个handler，消息会按照类型来做不通的处理。越往下面看就会有更多的模块之间的依赖，将模块独立出来分析反而不是很方便，所以接下来换一种思路，从一个完整的鉴权流程来分析源码以及模块之间的依赖关系。
分析连接鉴权流程
tb默认的设备鉴权方式是token的方式，所以先挑这个来分析整个的数据上下行流程，因此我们详细分析一下processAuthTokenConnect这个方法，源码如下。
private void processAuthTokenConnect(ChannelHandlerContext ctx, MqttConnectMessage msg) {
String userName = msg.payload().userName();
log.info("[{}] Processing connect msg for client with user name: {}!", sessionId, userName);
if (StringUtils.isEmpty(userName)) {
ctx.writeAndFlush(createMqttConnAckMsg(CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD));
ctx.close();
} else {
transportService.process(ValidateDeviceTokenRequestMsg.newBuilder().setToken(userName).build(),
new TransportServiceCallback() {
@Override
public void onSuccess(ValidateDeviceCredentialsResponseMsg msg) {
onValidateDeviceResponse(msg, ctx);
}
@Override
public void onError(Throwable e) {
log.trace("[{}] Failed to process credentials: {}", address, userName, e);
ctx.writeAndFlush(createMqttConnAckMsg(MqttConnectReturnCode.CONNECTION_REFUSED_SERVER_UNAVAILABLE));
ctx.close();
}
});
}
}
通过将token提取出来，然后拼装成一个ValidateDeviceTokenRequestMsg对象之后调用transportService服务。其中transportService服务的实现类为org.thingsboard.server.common.transport.service.DefaultTransportService，该类属于transport-api模块。
@Slf4j
@Service
@ConditionalOnExpression("'${service.type:null}'=='monolith' || '${service.type:null}'=='tb-transport'")
public class DefaultTransportService implements TransportService {
@Value("${transport.rate_limits.enabled}")
private boolean rateLimitEnabled;
@Value("${transport.rate_limits.tenant}")
通过process将消息发送出去，process的时候又调用了transportApiRequestTemplate这个对象
@Override
public void process(TransportProtos.ValidateDeviceTokenRequestMsg msg, TransportServiceCallback callback) {
log.trace("Processing msg: {}", msg);
TbProtoQueueMsg protoMsg = new TbProtoQueueMsg(UUID.randomUUID(), TransportApiRequestMsg.newBuilder().setValidateTokenRequestMsg(msg).build());
AsyncCallbackTemplate.withCallback(transportApiRequestTemplate.send(protoMsg),
response -> callback.onSuccess(response.getValue().getValidateTokenResponseMsg()), callback::onError, transportCallbackExecutor);
}
这个对象对应的实现类位于queue模块，org.thingsboard.server.queue.common.DefaultTbQueueRequestTemplate，他的send方法会返回一个ListenableFuture
@Override
public ListenableFuture send(Request request) {
if (tickSize > maxPendingRequests) {
return Futures.immediateFailedFuture(new RuntimeException("Pending request map is full!"));
}
UUID requestId = UUID.randomUUID();
request.getHeaders().put(REQUEST_ID_HEADER, uuidToBytes(requestId));
request.getHeaders().put(RESPONSE_TOPIC_HEADER, stringToBytes(responseTemplate.getTopic()));
request.getHeaders().put(REQUEST_TIME, longToBytes(System.currentTimeMillis()));
SettableFuture future = SettableFuture.create();
ResponseMetaData responseMetaData = new ResponseMetaData(tickTs + maxRequestTimeout, future);
pendingRequests.putIfAbsent(requestId, responseMetaData);
log.trace("[{}] Sending request, key [{}], expTime [{}]", requestId, request.getKey(), responseMetaData.expTime);
requestTemplate.send(TopicPartitionInfo.builder().topic(requestTemplate.getDefaultTopic()).build(), request, new TbQueueCallback() {
@Override
public void onSuccess(TbQueueMsgMetadata metadata) {
log.trace("[{}] Request sent: {}", requestId, metadata);
}
@Override
public void onFailure(Throwable t) {
pendingRequests.remove(requestId);
future.setException(t);
}
});
return future;
}
send方法通过requestTemplate进行消息的发送，因为使用的是Kafka作为消息管道，所以其实现类为queue模块的org.thingsboard.server.queue.kafka.TbKafkaProducerTemplate类，调用其send方法将消息真实发送出去，其中producer就是一个kafka的producer实现，最后将信息发送到tb_transport.api.requests这个topical中。
@Override
public void send(TopicPartitionInfo tpi, T msg, TbQueueCallback callback) {
createTopicIfNotExist(tpi);
String key = msg.getKey().toString();
byte[] data = msg.getData();
ProducerRecord record;
Iterable headers = msg.getHeaders().getData().entrySet().stream().map(e -> new RecordHeader(e.getKey(), e.getValue())).collect(Collectors.toList());
record = new ProducerRecord(tpi.getFullTopicName(), null, key, data, headers);
producer.send(record, (metadata, exception) -> {
if (exception == null) {
if (callback != null) {
callback.onSuccess(new KafkaTbQueueMsgMetadata(metadata));
}
} else {
if (callback != null) {
callback.onFailure(exception);
} else {
log.warn("Producer template failure: {}", exception.getMessage(), exception);
}
}
});
}
乍看至下认证的流程已经完成了，其实仔细一看里面问题还很多，首先需要问的是这个认证信息是如何返回给mqtt-transport的？
像kafka这类消息管道都是异步进行数据的传输的，所以发送消息之后最多会收到一个是否发送成功的信息，实际上是没有认证的返回信息的，但是在MqttTransportHandler的processAuthTokenConnect方法里面明确是有对返回信息做进一步核实的代码的，那这个返回信息到底是怎么来的呢？
如何拿到认证的返回信息
在tb的代码中有很多类似的异步操作，并且都是要对返回信息进行确认的。比如这个认证信息，其实是通过queue模块的org.thingsboard.server.queue.common.DefaultTbQueueRequestTemplate方法来实现的，下面是处理的源码
@Override
public void init() {
queueAdmin.createTopicIfNotExists(responseTemplate.getTopic());
this.requestTemplate.init();
tickTs = System.currentTimeMillis();
responseTemplate.subscribe();
executor.submit(() -> {
long nextCleanupMs = 0L;
while (!stopped) {
try {
List responses = responseTemplate.poll(pollInterval);
if (responses.size() > 0) {
log.trace("Polling responses completed, consumer records count [{}]", responses.size());
} else {
continue;
}
responses.forEach(response -> {
byte[] requestIdHeader = response.getHeaders().get(REQUEST_ID_HEADER);
UUID requestId;
if (requestIdHeader == null) {
log.error("[{}] Missing requestId in header and body", response);
} else {
requestId = bytesToUuid(requestIdHeader);
log.trace("[{}] Response received: {}", requestId, response);
ResponseMetaData expectedResponse = pendingRequests.remove(requestId);
if (expectedResponse == null) {
log.trace("[{}] Invalid or stale request", requestId);
} else {
expectedResponse.future.set(response);
}
}
});
responseTemplate.commit();
tickTs = System.currentTimeMillis();
tickSize = pendingRequests.size();
if (nextCleanupMs {
if (value.expTime staleRequest = pendingRequests.remove(key);
if (staleRequest != null) {
log.trace("[{}] Request timeout detected, expTime [{}], tickTs [{}]", key, staleRequest.expTime, tickTs);
staleRequest.future.setException(new TimeoutException());
}
}
});
nextCleanupMs = tickTs + maxRequestTimeout;
}
} catch (Throwable e) {
log.warn("Failed to obtain responses from queue.", e);
try {
Thread.sleep(pollInterval);
} catch (InterruptedException e2) {
log.trace("Failed to wait until the server has capacity to handle new responses", e2);
}
}
}
});
}
服务启动的时候会创建一个response对应topical的消费者，对应的topical格式为tb_transport.api.responses.hostname，然后从这个topical拉取消息。
DefaultTbQueueRequestTemplate.send()方法在发送的时候会将当前需要发送的消息存入pendingRequests对象，以requestId为唯一ID，将返回的Future装箱之后作为value。当从topical中获取到消息，且ID可以对应上的时候会将pendingRequests对应key的value的状态设置为已完成，然后出发回调。
回调之后就可以拿到core返回的认证信息了，这样就完成了一个完整的认证流程。
本文章来源于互联网，如有侵权，请联系删除！原文地址：Thingsboard源码探索（1）
相关推荐: 区块链、大数据、云计算、物联网和人工智能的最初发明人或发明团队是谁？
区块链是与比特币一同诞生的。一般民众都将比特币简单视为一种电子货币。但从根本上来说，它搭建的是一种计算机协议主体结构的技术。从应用上来说，区块链是一种分布式账本；从技术上来说，区块链是P2P组网、非对称加密技术以及数据库技术等多种技术的综合。 区块链的技术架构…
