常见的开源mqttBroker很多，如：
Mosquitto、emqx；
这里简单的介绍了mqtt，然后利用springboot netty实现了简单的mqtt Broker。

mqtt Broker：springboot netty实现；
mqtt client：MQTT.fx工具软件；

1 开发

MQTT（Message Queuing Telemetry Transport，消息队列遥测传输协议），是一种基于发布/订阅（publish/subscribe）模式的"轻量级"通讯协议。

mqtt是运行在TCP/IP之上的应用层，所以mqtt Broker基本就是个TCP Server。
创建主要的类：
1） MqttBrokerChannelHandler：
server channel处理的类；
2 ）MqttBrokerChannelInitializer
server channel初始化的类
3）MqttBroker
server类。
4）MqttClientStartListener：监听到springboot启动后，启动MqttBroker。

mqtt数据协议：
数据结构包括：
1）固定报头（Fixed header）；
2）可变报头（Variable header）；
3）有效载荷（Payload）。

其中，MqttBrokerChannelHandler类中实现了mqtt数据的解析处理。

	    @Override
    public void channelRead(ChannelHandlerContext channelHandlerContext, Object msg) throws Exception {
        MqttMessage mqttMessage = (MqttMessage) msg;
        log.info("--------------------------begin---------------------------*");
        log.info("来自终端:" + channelHandlerContext.channel().remoteAddress());
        log.info("接收消息：" + mqttMessage.toString());
        try {
            MqttMessageType type = mqttMessage.fixedHeader().messageType();
            MessageStrategy messageStrategy =  messageStrategyManager.getMessageStrategy(type);
            if(messageStrategy!=null){
                messageStrategy.sendResponseMessage(channelHandlerContext.channel(),mqttMessage);
            }
        }catch (Exception e) {
            e.printStackTrace();
        }
        log.info("--------------------------end---------------------------*");
    }

在这里我们定义一个策略接口，实现对mqtt 消息的解析返回。

public interface MessageStrategy {
    void sendResponseMessage(Channel channel, MqttMessage mqttMessage);
}

然后我们就可以实现不同类型消息的策略处理，如对Connect消息的处理：

@Slf4j
public class ConnectAckMessageStrategy implements MessageStrategy{

    @Override
    public void sendResponseMessage(Channel channel, MqttMessage mqttMessage) {
        MqttConnectMessage mqttConnectMessage = (MqttConnectMessage)mqttMessage;

        /*---------------------------解析接收的消息----------------------------*/
        MqttFixedHeader mqttFixedHeader = mqttConnectMessage.fixedHeader();
        MqttConnectVariableHeader mqttConnectVariableHeader = mqttConnectMessage.variableHeader();

        /*---------------------------构建返回的消息---------------------------*/
        //	构建返回报文， 固定报头
        MqttConnAckVariableHeader mqttConnAckVariableHeader =new MqttConnAckVariableHeader(MqttConnectReturnCode.CONNECTION_ACCEPTED, mqttConnectVariableHeader.isCleanSession());
        MqttFixedHeader mqttFixedHeaderBack = new MqttFixedHeader(MqttMessageType.CONNACK,mqttFixedHeader.isDup(), MqttQoS.AT_MOST_ONCE, mqttFixedHeader.isRetain(), 0x02);

        //	构建CONNACK消息体
        MqttConnAckMessage connAck = new MqttConnAckMessage(mqttFixedHeaderBack, mqttConnAckVariableHeader);
        log.info("返回消息:"+connAck.toString());
        channel.writeAndFlush(connAck);
    }
}

最后我们定义了MessageStrategyManager类实现对不同类型消息的策略处理：

@Slf4j
@Component
public class MessageStrategyManager {
    public Map<MqttMessageType, MessageStrategy> messageStrategyMap = new HashMap<>();

    //根据消息类型获取对应的策略类
    public  MessageStrategy getMessageStrategy(MqttMessageType messageType){
        switch (messageType){
            case CONNECT:
                return new ConnectAckMessageStrategy();
            case PUBLISH:
                return new PublishAckMessageStrategy();
            case PUBREL:
                return new PublishCompleteMessageStrategy();
            case SUBSCRIBE:
                return new SubscribeAckMessageStrategy();
            case UNSUBSCRIBE:
                return new UnSubscribeAckMessageStrategy();
            case PINGREQ:
                return new PingMessageStrategy();
            default:
                return null;
        }
    }

    //根据消息类型获取返回消息的类型
    private  static MqttMessageType getResMqttMessageType(MqttMessageType messageType){
        switch (messageType){
            case CONNECT:
                return MqttMessageType.CONNACK;
            case PUBLISH:
                return MqttMessageType.PUBACK;
            case PUBREL:
                return MqttMessageType.PUBLISH;
            case SUBSCRIBE:
                return MqttMessageType.SUBACK;
            case UNSUBSCRIBE:
                return MqttMessageType.UNSUBACK;
            case PINGREQ:
                return MqttMessageType.PINGRESP;
            default:
                return null;
        }
    }
}

2 验证

2.1 MQTT.fx连接broker

MQTT.fx发送消息的类型：CONNECT
broker返回消息的类型：CONNACK

2.2 MQTT.fx向broker发送心跳

MQTT.fx发送消息的类型：PINGREQ
broker返回消息的类型：PINGRESP

2.3 MQTT.fx向broker发布topic

首先MQTT.fx发布topic为：

weather/beijing

发布内容为：

The low temperature is 20C and the high temperature is 25 C

从log看到broker已经收到topic：
MQTT.fx发送消息的类型：PUBLISH
这里qosLevel=AT_MOST_ONCE。

2.4 MQTT.fx断开连接broker

MQTT.fx发送消息的类型：DISCONNECT

代码详见：
https://gitee.com/linghufeixia/iot-simple
code3