一款高性、高吞吐量、高扩展性的物联网mqtt集群broker

package io.github.quickmsg.core.protocol; import io.github.quickmsg.common.acl.AclAction; import io.github.quickmsg.common.acl.AclManager; import io.github.quickmsg.common.auth.AuthManager; import io.github.quickmsg.common.channel.ChannelRegistry; import io.github.quickmsg.common.channel.MqttChannel; import io.github.quickmsg.common.config.AuthConfig; import io.github.quickmsg.common.config.ConnectModel; import io.github.quickmsg.common.context.ReceiveContext; import io.github.quickmsg.common.enums.ChannelStatus; import io.github.quickmsg.common.enums.Event; import io.github.quickmsg.common.message.*; import io.github.quickmsg.common.metric.CounterType; import io.github.quickmsg.common.metric.MetricManager; import io.github.quickmsg.common.metric.MetricManagerHolder; import io.github.quickmsg.common.protocol.Protocol; import io.github.quickmsg.common.topic.SubscribeTopic; import io.github.quickmsg.common.topic.TopicRegistry; import io.github.quickmsg.core.mqtt.MqttReceiveContext; import io.netty.buffer.Unpooled; import io.netty.handler.codec.mqtt.*; import lombok.extern.slf4j.Slf4j; import reactor.core.publisher.Mono; import reactor.core.scheduler.Schedulers; import reactor.util.context.ContextView; import java.util.ArrayList; import java.util.List; import java.util.Optional; import java.util.Set; import java.util.stream.Collectors; /** * @author luxurong */ @Slf4j public class ConnectProtocol implements Protocol<MqttConnectMessage> { private static List<MqttMessageType> MESSAGE_TYPE_LIST = new ArrayList<>(); private static final int MILLI_SECOND_PERIOD = 1_000; static { MESSAGE_TYPE_LIST.add(MqttMessageType.CONNECT); } private static void accept(MqttChannel mqttChannel1) { // metric.getMetricCounter(CounterEnum.CONNECT_COUNTER).decrement(); } @Override public Mono<Void> parseProtocol(SmqttMessage<MqttConnectMessage> smqttMessage, MqttChannel mqttChannel, ContextView contextView) { try { MqttConnectMessage message = smqttMessage.getMessage(); MetricManagerHolder.metricManager.getMetricRegistry().getMetricCounter(CounterType.CONNECT_EVENT).increment(); MqttReceiveContext mqttReceiveContext = (MqttReceiveContext) contextView.get(ReceiveContext.class); EventRegistry eventRegistry = mqttReceiveContext.getEventRegistry(); MqttConnectVariableHeader mqttConnectVariableHeader = message.variableHeader(); MqttConnectPayload mqttConnectPayload = message.payload(); String clientIdentifier = mqttConnectPayload.clientIdentifier(); ChannelRegistry channelRegistry = mqttReceiveContext.getChannelRegistry(); TopicRegistry topicRegistry = mqttReceiveContext.getTopicRegistry(); MetricManager metricManager = mqttReceiveContext.getMetricManager(); byte mqttVersion = (byte) mqttConnectVariableHeader.version(); AuthManager authManager = mqttReceiveContext.getAuthManager(); /*check clientIdentifier exist*/ MqttChannel existMqttChannel = channelRegistry.get(clientIdentifier); if (mqttReceiveContext.getConfiguration().getConnectModel() == ConnectModel.UNIQUE) { if (existMqttChannel != null && existMqttChannel.getStatus() == ChannelStatus.ONLINE) { return mqttChannel.write( MqttMessageBuilder.buildConnectAck(MqttConnectReturnCode.CONNECTION_REFUSED_IDENTIFIER_REJECTED, mqttVersion), false).then(mqttChannel.close()); } } else { if (existMqttChannel != null && existMqttChannel.getStatus() == ChannelStatus.ONLINE) { if (System.currentTimeMillis() - existMqttChannel.getConnectTime() > (mqttReceiveContext.getConfiguration().getNotKickSecond() * 1000)) { existMqttChannel.close().subscribe(); } else { return mqttChannel.write( MqttMessageBuilder.buildConnectAck(MqttConnectReturnCode.CONNECTION_REFUSED_IDENTIFIER_REJECTED, mqttVersion), false).then(mqttChannel.close()); } } } /*protocol version support*/ if (MqttVersion.MQTT_3_1.protocolLevel() != mqttVersion && MqttVersion.MQTT_3_1_1.protocolLevel() != mqttVersion && MqttVersion.MQTT_5.protocolLevel() != mqttVersion) { return mqttChannel.write( MqttMessageBuilder.buildConnectAck(MqttConnectReturnCode.CONNECTION_REFUSED_UNACCEPTABLE_PROTOCOL_VERSION, mqttVersion), false).then(mqttChannel.close()); } /*password check*/ if (authManager.auth(mqttConnectPayload.userName(),mqttConnectPayload.passwordInBytes(), clientIdentifier)) { /*cancel defer close not authenticate channel */ mqttChannel.disposableClose(); mqttChannel.setClientIdentifier(mqttConnectPayload.clientIdentifier()); if (mqttConnectVariableHeader.isWillFlag()) { mqttChannel.setWill(MqttChannel.Will.builder() .isRetain(mqttConnectVariableHeader.isWillRetain()) .willTopic(mqttConnectPayload.willTopic()) .willMessage(mqttConnectPayload.willMessageInBytes()) .mqttQoS(MqttQoS.valueOf(mqttConnectVariableHeader.willQos())) .build()); } mqttChannel.setAuthTime(System.currentTimeMillis()); mqttChannel.setConnectTime(System.currentTimeMillis()); mqttChannel.setKeepalive(mqttConnectVariableHeader.keepAliveTimeSeconds()); mqttChannel.setSessionPersistent(!mqttConnectVariableHeader.isCleanSession()); mqttChannel.setStatus(ChannelStatus.ONLINE); mqttChannel.setUsername(mqttConnectPayload.userName()); /*registry unread event close channel */ mqttChannel.getConnection().onReadIdle((long) mqttConnectVariableHeader.keepAliveTimeSeconds() * MILLI_SECOND_PERIOD << 1, () -> close(metricManager, mqttChannel, mqttReceiveContext, eventRegistry)); CloseMqttMessage closeMqttMessage = new CloseMqttMessage(); closeMqttMessage.setClientIdentifier(clientIdentifier); ClusterMessage clusterMessage = new ClusterMessage(closeMqttMessage); mqttReceiveContext.getClusterRegistry().spreadPublishMessage(clusterMessage) .subscribe(); /*registry will message send */ mqttChannel.registryClose(channel -> Optional.ofNullable(mqttChannel.getWill()) .ifPresent(will -> topicRegistry.getSubscribesByTopic(will.getWillTopic(), will.getMqttQoS()) .forEach(subscribeTopic -> { MqttChannel subscribeChannel = subscribeTopic.getMqttChannel(); subscribeChannel.write( MqttMessageBuilder .buildPub(false, subscribeTopic.getQoS(), subscribeTopic.getQoS() == MqttQoS.AT_MOST_ONCE ? 0 : subscribeChannel.generateMessageId(), will.getWillTopic(), Unpooled.wrappedBuffer(wil