手把手教大家写书写一个Mqtt网关

摘要：物联网是现在比较热门的软件领域，众多物联网厂商都有自己的物联网平台，而物联网平台其中一个核心的模块就是Mqtt网关。

本文分享自华为云社区《一文带你掌握物联网mqtt网关搭建背后的技术原理》，作者：张俭。
前言

物联网是现在比较热门的软件领域，众多物联网厂商都有自己的物联网平台，而物联网平台其中一个核心的模块就是Mqtt网关。这篇文章的目的是手把手教大家写书写一个mqtt网关，后端存储支持Kafka/Pulsar，支持mqtt 连接、断链、发送消息、订阅消息。技术选型：

• Netty java最流行的网络框架
  
• netty-codec-mqtt netty的子项目，mqtt编解码插件
  
• Pulsar/Kafka 流行的消息中间件作为后端存储
  

核心pom依赖如下

1. <dependency>
2. <groupId>io.netty</groupId>
3. <artifactId>netty-codec-mqtt</artifactId>
4. </dependency>
5. <dependency>
6. <groupId>io.netty</groupId>
7. <artifactId>netty-common</artifactId>
8. </dependency>
9. <dependency>
10. <groupId>io.netty</groupId>
11. <artifactId>netty-transport</artifactId>
12. </dependency>
13. <dependency>
14. <groupId>org.apache.pulsar</groupId>
15. <artifactId>pulsar-client-original</artifactId>
16. <version>${pulsar.version}</version>
17. </dependency>
18. <dependency>
19. <groupId>org.apache.kafka</groupId>
20. <artifactId>kafka-clients</artifactId>
21. <version>${kafka.version}</version>
22. </dependency>
23. <dependency>
24. <groupId>org.eclipse.paho</groupId>
25. <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
26. <version>${mqtt-client.version}</version>
27. <scope>test</scope>
28. </dependency>

复制代码

软件参数设计

软件参数可谓是非常常见，复杂的开源项目，参数甚至可以达到上百个、配置文件长达数千行。我们需要的配置有
MqttServer监听的端口

监听端口的配置即使是写demo也非常必要，常常用在单元测试中，由于单元测试跑完之后，即使网络服务器关闭，操作系统也不会立即释放端口，所以单元测试的时候指定随机端口非常关键，在java中，我们可以通过这样的工具类来获取一个空闲的端口。未配置的话，我们就使用mqtt的默认端口1883。

1. package io.github.protocol.mqtt.broker.util;
3. import java.io.IOException;
4. import java.io.UncheckedIOException;
5. import java.net.ServerSocket;
7. public class SocketUtil {
9. public static int getFreePort() {
10. try (ServerSocket serverSocket = new ServerSocket(0)) {
11. return serverSocket.getLocalPort();
12.         } catch (IOException e) {
13. throw new UncheckedIOException(e);
14.         }
15.     }
17. }

复制代码

后端存储配置

我们的mqtt网关是没有可靠的存储能力的，依赖后端的消息中间件来做持久化处理。后端规划支持Pulsar、Kafka两种类型。定义枚举类如下

1. public enum ProcessorType {
2.     KAFKA,
3.     PULSAR,
4. }

复制代码

对应的KafkaProcessorConfig、PulsarProcessorConfig比较简单，包含基础的连接地址即可，如果后续要做性能调优、安全，这块还是会有更多的配置项

1. @Setter
2. @Getter
3. public class KafkaProcessorConfig {
5. private String bootstrapServers = "localhost:9092";
7. public KafkaProcessorConfig() {
8.     }
9. }
10. @Setter
11. @Getter
12. public class PulsarProcessorConfig {
14. private String httpUrl = "http://localhost:8080";
16. private String serviceUrl = "pulsar://localhost:6650";
18. public PulsarProcessorConfig() {
19.     }
20. }

复制代码

启动netty MqttServer

我们通过netty启动一个mqttServer，添加mqtt解码器

1. package io.github.protocol.mqtt.broker;
3. import io.github.protocol.mqtt.broker.processor.KafkaProcessor;
4. import io.github.protocol.mqtt.broker.processor.KafkaProcessorConfig;
5. import io.github.protocol.mqtt.broker.processor.MqttProcessor;
6. import io.github.protocol.mqtt.broker.processor.PulsarProcessor;
7. import io.github.protocol.mqtt.broker.processor.PulsarProcessorConfig;
8. import io.github.protocol.mqtt.broker.util.SocketUtil;
9. import io.netty.bootstrap.ServerBootstrap;
10. import io.netty.channel.ChannelFuture;
11. import io.netty.channel.ChannelInitializer;
12. import io.netty.channel.ChannelOption;
13. import io.netty.channel.ChannelPipeline;
14. import io.netty.channel.EventLoopGroup;
15. import io.netty.channel.nio.NioEventLoopGroup;
16. import io.netty.channel.socket.SocketChannel;
17. import io.netty.channel.socket.nio.NioServerSocketChannel;
18. import io.netty.handler.codec.mqtt.MqttDecoder;
19. import io.netty.handler.codec.mqtt.MqttEncoder;
20. import io.netty.handler.logging.LogLevel;
21. import io.netty.handler.logging.LoggingHandler;
22. import lombok.extern.slf4j.Slf4j;
24. @Slf4j
25. public class MqttServer {
27. private final MqttServerConfig mqttServerConfig;
29. public MqttServer() {
30. this(new MqttServerConfig());
31.     }
33. public MqttServer(MqttServerConfig mqttServerConfig) {
34. this.mqttServerConfig = mqttServerConfig;
35. if (mqttServerConfig.getPort() == 0) {
36.             mqttServerConfig.setPort(SocketUtil.getFreePort());
37.         }
38.     }
40. public void start() throws Exception {
41.         EventLoopGroup bossGroup = new NioEventLoopGroup(1);
42.         EventLoopGroup workerGroup = new NioEventLoopGroup();
43. try {
44.             ServerBootstrap b = new ServerBootstrap();
45.             b.group(bossGroup, workerGroup)
46.                     .channel(NioServerSocketChannel.class)
47.                     .option(ChannelOption.SO_BACKLOG, 100)
48.                     .handler(new LoggingHandler(LogLevel.INFO))
49.                     .childHandler(new ChannelInitializer<SocketChannel>() {
50.                         @Override
51. public void initChannel(SocketChannel ch) throws Exception {
52.                             ChannelPipeline p = ch.pipeline();
53. // decoder
54.                             p.addLast(new MqttDecoder());
55.                             p.addLast(MqttEncoder.INSTANCE);
56.                         }
57.                     });
59. // Start the server.
60.             ChannelFuture f = b.bind(mqttServerConfig.getPort()).sync();
62. // Wait until the server socket is closed.
63.             f.channel().closeFuture().sync();
64.         } finally {
65. // Shut down all event loops to terminate all threads.
66.             bossGroup.shutdownGracefully();
67.             workerGroup.shutdownGracefully();
68.         }
69.     }
71. private MqttProcessor processor(MqttServerConfig config) {
72. return switch (config.getProcessorType()) {
73. case KAFKA -> new KafkaProcessor(config.getMqttAuth(), config.getKafkaProcessorConfig());
74. case PULSAR -> new PulsarProcessor(config.getMqttAuth(), config.getPulsarProcessorConfig());
75.         };
76.     }
78. public int getPort() {
79. return mqttServerConfig.getPort();
80.     }
82. }

复制代码

MqttserverStarter.java

我们写一个简单的main函数用来启动mqttServer，方便调测

1. package io.github.protocol.mqtt.broker;
3. public class MqttServerStarter {
5. public static void main(String[] args) throws Exception {
6. new MqttServer().start();
7.     }
9. }

复制代码

客户端使用eclipse mqtt client进行测试

1. package io.github.protocol.mqtt;
3. import lombok.extern.log4j.Log4j2;
4. import org.eclipse.paho.client.mqttv3.MqttClient;
5. import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
6. import org.eclipse.paho.client.mqttv3.MqttException;
7. import org.eclipse.paho.client.mqttv3.MqttMessage;
8. import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
10. @Log4j2
11. public class MqttClientPublishExample {
13. public static void main(String[] args) throws Exception {
14.         String topic = "MQTT Examples";
15.         String content = "Message from MqttPublishExample";
16.         int qos = 2;
17.         String broker = "tcp://127.0.0.1:1883";
18.         String clientId = "JavaSample";
19.         MemoryPersistence persistence = new MemoryPersistence();
21. try {
22.             MqttClient sampleClient = new MqttClient(broker, clientId, persistence);
23.             MqttConnectOptions connOpts = new MqttConnectOptions();
24.             connOpts.setCleanSession(true);
25.             log.info("Connecting to broker: {}", broker);
26.             sampleClient.connect(connOpts);
27.             log.info("Connected");
28.             log.info("Publishing message: {}", content);
29.             MqttMessage message = new MqttMessage(content.getBytes());
30.             message.setQos(qos);
31.             sampleClient.publish(topic, message);
32.             log.info("Message published");
33.             sampleClient.disconnect();
34.             log.info("Disconnected");
35.             System.exit(0);
36.         } catch (MqttException me) {
37.             log.error("reason {} msg {}", me.getReasonCode(), me.getMessage(), me);
38.         }
39.     }
41. }

复制代码

然后我们先运行MqttServer，再运行MqttClient，发现MqttClient卡住了

1. Connecting to broker: tcp://127.0.0.1:1883

复制代码

这是为什么呢，我们通过抓包发现仅仅只有客户端发送了Mqtt connect信息，服务端并没有响应
但是根据mqtt标准协议，发送Connect消息，必须要有ConnAck响应
所以我们需要在接收到Connect后，返回connAck消息。我们创建一个MqttHandler，让他继承ChannelInboundHandlerAdapter, 用来接力MqttDecoder解码完成后的消息，这里要重点继承其中的channelRead方法，以及channelInactive方法，用来释放断链时需要释放的资源

1. package com.github.shoothzj.mqtt;
3. import io.netty.channel.ChannelHandlerContext;
4. import io.netty.channel.ChannelInboundHandlerAdapter;
5. import lombok.extern.slf4j.Slf4j;
7. @Slf4j
8. public class MqttHandler extends ChannelInboundHandlerAdapter {
10.     @Override
11. public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
12. super.channelRead(ctx, msg);
13.     }
15. }

复制代码

然后把这个handler加入到netty的职责链中，放到解码器的后面
在mqtt handler中插入我们的代码

1. @Override
2. public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
3. super.channelRead(ctx, msg);
4. if (msg instanceof MqttConnectMessage) {
5.             handleConnect(ctx, (MqttConnectMessage) msg);
6.         } else {
7.             log.error("Unsupported type msg [{}]", msg);
8.         }
9.     }
11. private void handleConnect(ChannelHandlerContext ctx, MqttConnectMessage connectMessage) {
12.         log.info("connect msg is [{}]", connectMessage);
13.     }

复制代码

打印出connectMessage如下

1. [MqttConnectMessage[fixedHeader=MqttFixedHeader[messageType=CONNECT, isDup=false, qosLevel=AT_MOST_ONCE, isRetain=false, remainingLength=22], variableHeader=MqttConnectVariableHeader[name=MQTT, version=4, hasUserName=false, hasPassword=false, isWillRetain=false, isWillFlag=false, isCleanSession=true, keepAliveTimeSeconds=60], payload=MqttConnectPayload[clientIdentifier=JavaSample, willTopic=null, willMessage=null, userName=null, password=null]]]

复制代码

1. 通常，mqtt connect message中会包含qos、用户名、密码等信息，由于我们启动客户端的时候也没有携带用户名和密码，这里获取到的都为null，我们先不校验这些消息，直接给客户端返回connack消息，代表连接成功

复制代码

1. final MqttConnAckMessage ackMessage = MqttMessageBuilders.connAck().returnCode(CONNECTION_ACCEPTED).build();
2.         ctx.channel().writeAndFlush(ackMessage);

复制代码

我们再运行起Server和Client，随后可以看到已经走过了Connect阶段，进入了publish message过程，接下来我们再实现更多的其他场景
附上此阶段的MqttHandler代码

1. package com.github.shoothzj.mqtt;import io.netty.channel.ChannelHandlerContext;import io.netty.channel.ChannelInboundHandlerAdapter;import io.netty.handler.codec.mqtt.MqttConnAckMessage;import io.netty.handler.codec.mqtt.MqttConnectMessage;import io.netty.handler.codec.mqtt.MqttConnectPayload;import io.netty.handler.codec.mqtt.MqttConnectVariableHeader;import io.netty.handler.codec.mqtt.MqttFixedHeader;import io.netty.handler.codec.mqtt.MqttMessageBuilders;import lombok.extern.slf4j.Slf4j;import static io.netty.handler.codec.mqtt.MqttConnectReturnCode.CONNECTION_ACCEPTED;@Slf4jpublic class MqttHandler extends ChannelInboundHandlerAdapter {    @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { super.channelRead(ctx, msg); if (msg instanceof MqttConnectMessage) {            handleConnect(ctx, (MqttConnectMessage) msg);        } else {            log.error("Unsupported type msg [{}]", msg);        }    } private void handleConnect(ChannelHandlerContext ctx, MqttConnectMessage connectMessage) {        log.info("connect msg is [{}]", connectMessage);        final MqttFixedHeader fixedHeader = connectMessage.fixedHeader();        final MqttConnectVariableHeader variableHeader = connectMessage.variableHeader();        final MqttConnectPayload connectPayload = connectMessage.payload();        final MqttConnAckMessage ackMessage = MqttMessageBuilders.connAck().returnCode(CONNECTION_ACCEPTED).build();
2.         ctx.channel().writeAndFlush(ackMessage);    }}

复制代码

我们当前把所有的逻辑都放在MqttHandler里面，不方便后续的扩展。抽象出一个MqttProcessor接口来处理具体的请求，MqttHandler负责解析MqttMessage的类型并分发。MqttProcess接口设计如下

1. package io.github.protocol.mqtt.broker.processor;
3. import io.netty.channel.ChannelHandlerContext;
4. import io.netty.handler.codec.mqtt.MqttConnAckMessage;
5. import io.netty.handler.codec.mqtt.MqttConnectMessage;
6. import io.netty.handler.codec.mqtt.MqttMessage;
7. import io.netty.handler.codec.mqtt.MqttPubAckMessage;
8. import io.netty.handler.codec.mqtt.MqttPublishMessage;
9. import io.netty.handler.codec.mqtt.MqttSubAckMessage;
10. import io.netty.handler.codec.mqtt.MqttSubscribeMessage;
11. import io.netty.handler.codec.mqtt.MqttUnsubAckMessage;
12. import io.netty.handler.codec.mqtt.MqttUnsubscribeMessage;
14. public interface MqttProcessor {
16.     void processConnect(ChannelHandlerContext ctx, MqttConnectMessage msg) throws Exception;
18.     void processConnAck(ChannelHandlerContext ctx, MqttConnAckMessage msg) throws Exception;
20.     void processPublish(ChannelHandlerContext ctx, MqttPublishMessage msg) throws Exception;
22.     void processPubAck(ChannelHandlerContext ctx, MqttPubAckMessage msg) throws Exception;
24.     void processPubRec(ChannelHandlerContext ctx, MqttMessage msg) throws Exception;
26.     void processPubRel(ChannelHandlerContext ctx, MqttMessage msg) throws Exception;
28.     void processPubComp(ChannelHandlerContext ctx, MqttMessage msg) throws Exception;
30.     void processSubscribe(ChannelHandlerContext ctx, MqttSubscribeMessage msg) throws Exception;
32.     void processSubAck(ChannelHandlerContext ctx, MqttSubAckMessage msg) throws Exception;
34.     void processUnsubscribe(ChannelHandlerContext ctx, MqttUnsubscribeMessage msg) throws Exception;
36.     void processUnsubAck(ChannelHandlerContext ctx, MqttUnsubAckMessage msg) throws Exception;
38.     void processPingReq(ChannelHandlerContext ctx, MqttMessage msg) throws Exception;
40.     void processPingResp(ChannelHandlerContext ctx, MqttMessage msg) throws Exception;
42.     void processDisconnect(ChannelHandlerContext ctx) throws Exception;
44.     void processAuth(ChannelHandlerContext ctx, MqttMessage msg) throws Exception;
46. }

复制代码

我们允许这些方法抛出异常，当遇到极难处理的故障时，把mqtt连接断掉（如后端存储故障），等待客户端的重连。
MqttHandler中来调用MqttProcessor，相关MqttHandler代码如下

1.    Preconditions.checkArgument(message instanceof MqttMessage);
2.         MqttMessage msg = (MqttMessage) message;
3. try {
4. if (msg.decoderResult().isFailure()) {
5.                 Throwable cause = msg.decoderResult().cause();
6. if (cause instanceof MqttUnacceptableProtocolVersionException) {
7. // Unsupported protocol version
8.                     MqttConnAckMessage connAckMessage = (MqttConnAckMessage) MqttMessageFactory.newMessage(
9. new MqttFixedHeader(MqttMessageType.CONNACK,
10. false, MqttQoS.AT_MOST_ONCE, false, 0),
11. new MqttConnAckVariableHeader(
12.                                     MqttConnectReturnCode.CONNECTION_REFUSED_UNACCEPTABLE_PROTOCOL_VERSION,
13. false), null);
14.                     ctx.writeAndFlush(connAckMessage);
15.                     log.error("connection refused due to invalid protocol, client address [{}]",
16.                             ctx.channel().remoteAddress());
17.                     ctx.close();
18. return;
19.                 } else if (cause instanceof MqttIdentifierRejectedException) {
20. // ineligible clientId
21.                     MqttConnAckMessage connAckMessage = (MqttConnAckMessage) MqttMessageFactory.newMessage(
22. new MqttFixedHeader(MqttMessageType.CONNACK,
23. false, MqttQoS.AT_MOST_ONCE, false, 0),
24. new MqttConnAckVariableHeader(MqttConnectReturnCode.CONNECTION_REFUSED_IDENTIFIER_REJECTED,
25. false), null);
26.                     ctx.writeAndFlush(connAckMessage);
27.                     log.error("ineligible clientId, client address [{}]", ctx.channel().remoteAddress());
28.                     ctx.close();
29. return;
30.                 }
31. throw new IllegalStateException(msg.decoderResult().cause().getMessage());
32.             }
33.             MqttMessageType messageType = msg.fixedHeader().messageType();
34. if (log.isDebugEnabled()) {
35.                 log.debug("Processing MQTT Inbound handler message, type={}", messageType);
36.             }
37. switch (messageType) {
38. case CONNECT:
39.                     Preconditions.checkArgument(msg instanceof MqttConnectMessage);
40.                     processor.processConnect(ctx, (MqttConnectMessage) msg);
41. break;
42. case CONNACK:
43.                     Preconditions.checkArgument(msg instanceof MqttConnAckMessage);
44.                     processor.processConnAck(ctx, (MqttConnAckMessage) msg);
45. break;
46. case PUBLISH:
47.                     Preconditions.checkArgument(msg instanceof MqttPublishMessage);
48.                     processor.processPublish(ctx, (MqttPublishMessage) msg);
49. break;
50. case PUBACK:
51.                     Preconditions.checkArgument(msg instanceof MqttPubAckMessage);
52.                     processor.processPubAck(ctx, (MqttPubAckMessage) msg);
53. break;
54. case PUBREC:
55.                     processor.processPubRec(ctx, msg);
56. break;
57. case PUBREL:
58.                     processor.processPubRel(ctx, msg);
59. break;
60. case PUBCOMP:
61.                     processor.processPubComp(ctx, msg);
62. break;
63. case SUBSCRIBE:
64.                     Preconditions.checkArgument(msg instanceof MqttSubscribeMessage);
65.                     processor.processSubscribe(ctx, (MqttSubscribeMessage) msg);
66. break;
67. case SUBACK:
68.                     Preconditions.checkArgument(msg instanceof MqttSubAckMessage);
69.                     processor.processSubAck(ctx, (MqttSubAckMessage) msg);
70. break;
71. case UNSUBSCRIBE:
72.                     Preconditions.checkArgument(msg instanceof MqttUnsubscribeMessage);
73.                     processor.processUnsubscribe(ctx, (MqttUnsubscribeMessage) msg);
74. break;
75. case UNSUBACK:
76.                     Preconditions.checkArgument(msg instanceof MqttUnsubAckMessage);
77.                     processor.processUnsubAck(ctx, (MqttUnsubAckMessage) msg);
78. break;
79. case PINGREQ:
80.                     processor.processPingReq(ctx, msg);
81. break;
82. case PINGRESP:
83.                     processor.processPingResp(ctx, msg);
84. break;
85. case DISCONNECT:
86.                     processor.processDisconnect(ctx);
87. break;
88. case AUTH:
89.                     processor.processAuth(ctx, msg);
90. break;
91. default:
92. throw new UnsupportedOperationException("Unknown MessageType: " + messageType);
93.             }
94.         } catch (Throwable ex) {
95.             ReferenceCountUtil.safeRelease(msg);
96.             log.error("Exception was caught while processing MQTT message, ", ex);
97.             ctx.close();
98.         }

复制代码

这里的代码，主要是针对MqttMessage的不同类型，调用MqttProcessor的不同方法，值得一提的有两点

• 提前判断了一些解码异常，fast fail
  
• 全局捕获异常，并进行断链处理
  

维护MqttSession

维护Mqtt会话的session，主要用来持续跟踪客户端会话信息，跟踪在系统中占用的资源等，考虑到无论是何种后端实现，都需要维护Mqtt的Session，我们构筑一个AbstractMqttProcessor来维护MqttSession

1. package io.github.protocol.mqtt.broker.processor;
3. import io.github.protocol.mqtt.broker.MqttSessionKey;
4. import io.github.protocol.mqtt.broker.auth.MqttAuth;
5. import io.github.protocol.mqtt.broker.util.ChannelUtils;
6. import io.github.protocol.mqtt.broker.util.MqttMessageUtil;
7. import io.netty.channel.ChannelHandlerContext;
8. import io.netty.handler.codec.mqtt.MqttConnAckMessage;
9. import io.netty.handler.codec.mqtt.MqttConnAckVariableHeader;
10. import io.netty.handler.codec.mqtt.MqttConnectMessage;
11. import io.netty.handler.codec.mqtt.MqttConnectReturnCode;
12. import io.netty.handler.codec.mqtt.MqttFixedHeader;
13. import io.netty.handler.codec.mqtt.MqttMessage;
14. import io.netty.handler.codec.mqtt.MqttMessageFactory;
15. import io.netty.handler.codec.mqtt.MqttMessageIdVariableHeader;
16. import io.netty.handler.codec.mqtt.MqttMessageType;
17. import io.netty.handler.codec.mqtt.MqttPubAckMessage;
18. import io.netty.handler.codec.mqtt.MqttPublishMessage;
19. import io.netty.handler.codec.mqtt.MqttQoS;
20. import io.netty.handler.codec.mqtt.MqttSubAckMessage;
21. import io.netty.handler.codec.mqtt.MqttSubAckPayload;
22. import io.netty.handler.codec.mqtt.MqttSubscribeMessage;
23. import io.netty.handler.codec.mqtt.MqttSubscribePayload;
24. import io.netty.handler.codec.mqtt.MqttUnsubAckMessage;
25. import io.netty.handler.codec.mqtt.MqttUnsubscribeMessage;
26. import lombok.extern.slf4j.Slf4j;
27. import org.apache.commons.lang3.StringUtils;
29. import java.util.stream.IntStream;
31. @Slf4j
32. public abstract class AbstractProcessor implements MqttProcessor {
34. protected final MqttAuth mqttAuth;
36. public AbstractProcessor(MqttAuth mqttAuth) {
37. this.mqttAuth = mqttAuth;
38.     }
40.     @Override
41. public void processConnect(ChannelHandlerContext ctx, MqttConnectMessage msg) throws Exception {
42.         String clientId = msg.payload().clientIdentifier();
43.         String username = msg.payload().userName();
44.         byte[] pwd = msg.payload().passwordInBytes();
45. if (StringUtils.isBlank(clientId) || StringUtils.isBlank(username)) {
46.             MqttConnAckMessage connAckMessage = (MqttConnAckMessage) MqttMessageFactory.newMessage(
47. new MqttFixedHeader(MqttMessageType.CONNACK,
48. false, MqttQoS.AT_MOST_ONCE, false, 0),
49. new MqttConnAckVariableHeader(MqttConnectReturnCode.CONNECTION_REFUSED_IDENTIFIER_REJECTED,
50. false), null);
51.             ctx.writeAndFlush(connAckMessage);
52.             log.error("the clientId username pwd cannot be empty, client address[{}]", ctx.channel().remoteAddress());
53.             ctx.close();
54. return;
55.         }
56. if (!mqttAuth.connAuth(clientId, username, pwd)) {
57.             MqttConnAckMessage connAckMessage = (MqttConnAckMessage) MqttMessageFactory.newMessage(
58. new MqttFixedHeader(MqttMessageType.CONNACK,
59. false, MqttQoS.AT_MOST_ONCE, false, 0),
60. new MqttConnAckVariableHeader(MqttConnectReturnCode.CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD,
61. false), null);
62.             ctx.writeAndFlush(connAckMessage);
63.             log.error("the clientId username pwd cannot be empty, client address[{}]", ctx.channel().remoteAddress());
64.             ctx.close();
65. return;
66.         }
68.         MqttSessionKey mqttSessionKey = new MqttSessionKey();
69.         mqttSessionKey.setUsername(username);
70.         mqttSessionKey.setClientId(clientId);
71.         ChannelUtils.setMqttSession(ctx.channel(), mqttSessionKey);
72.         log.info("username {} clientId {} remote address {} connected",
73.                 username, clientId, ctx.channel().remoteAddress());
74.         onConnect(mqttSessionKey);
75.         MqttConnAckMessage mqttConnectMessage = (MqttConnAckMessage) MqttMessageFactory.newMessage(
76. new MqttFixedHeader(MqttMessageType.CONNACK,
77. false, MqttQoS.AT_MOST_ONCE, false, 0),
78. new MqttConnAckVariableHeader(MqttConnectReturnCode.CONNECTION_ACCEPTED, false),
79. null);
80.         ctx.writeAndFlush(mqttConnectMessage);
81.     }
83. protected void onConnect(MqttSessionKey mqttSessionKey) {
84.     }
86.     @Override
87. public void processConnAck(ChannelHandlerContext ctx, MqttConnAckMessage msg) throws Exception {
88.         MqttSessionKey mqttSession = ChannelUtils.getMqttSession(ctx.channel());
89. if (mqttSession == null) {
90.             log.error("conn ack, client address {} not authed", ctx.channel().remoteAddress());
91.             ctx.close();
92.         }
93.     }
95.     @Override
96. public void processPublish(ChannelHandlerContext ctx, MqttPublishMessage msg) throws Exception {
97.         MqttSessionKey mqttSession = ChannelUtils.getMqttSession(ctx.channel());
98. if (mqttSession == null) {
99.             log.error("publish, client address {} not authed", ctx.channel().remoteAddress());
100.             ctx.close();
101. return;
102.         }
103. if (msg.fixedHeader().qosLevel() == MqttQoS.FAILURE) {
104.             log.error("failure. clientId {}, username {} ", mqttSession.getClientId(), mqttSession.getUsername());
105. return;
106.         }
107. if (msg.fixedHeader().qosLevel() == MqttQoS.EXACTLY_ONCE) {
108.             log.error("does not support QoS2 protocol. clientId {}, username {} ",
109.                     mqttSession.getClientId(), mqttSession.getUsername());
110. return;
111.         }
112.         onPublish(ctx, mqttSession, msg);
113.     }
115. protected void onPublish(ChannelHandlerContext ctx, MqttSessionKey mqttSessionKey,
116.                              MqttPublishMessage msg) throws Exception {
117.     }
119.     @Override
120. public void processPubAck(ChannelHandlerContext ctx, MqttPubAckMessage msg) throws Exception {
121.         MqttSessionKey mqttSession = ChannelUtils.getMqttSession(ctx.channel());
122. if (mqttSession == null) {
123.             log.error("pub ack, client address {} not authed", ctx.channel().remoteAddress());
124.             ctx.close();
125.         }
126.     }
128.     @Override
129. public void processPubRec(ChannelHandlerContext ctx, MqttMessage msg) throws Exception {
130.         MqttSessionKey mqttSession = ChannelUtils.getMqttSession(ctx.channel());
131. if (mqttSession == null) {
132.             log.error("pub rec, client address {} not authed", ctx.channel().remoteAddress());
133.             ctx.close();
134.         }
135.     }
137.     @Override
138. public void processPubRel(ChannelHandlerContext ctx, MqttMessage msg) throws Exception {
139.         MqttSessionKey mqttSession = ChannelUtils.getMqttSession(ctx.channel());
140. if (mqttSession == null) {
141.             log.error("pub rel, client address {} not authed", ctx.channel().remoteAddress());
142.             ctx.close();
143.         }
144.     }
146.     @Override
147. public void processPubComp(ChannelHandlerContext ctx, MqttMessage msg) throws Exception {
148.         MqttSessionKey mqttSession = ChannelUtils.getMqttSession(ctx.channel());
149. if (mqttSession == null) {
150.             log.error("pub comp, client address {} not authed", ctx.channel().remoteAddress());
151.             ctx.close();
152.         }
153.     }
155.     @Override
156. public void processSubscribe(ChannelHandlerContext ctx, MqttSubscribeMessage msg) throws Exception {
157.         MqttSessionKey mqttSession = ChannelUtils.getMqttSession(ctx.channel());
158. if (mqttSession == null) {
159.             log.error("sub, client address {} not authed", ctx.channel().remoteAddress());
160.             ctx.close();
161.         }
162.         onSubscribe(ctx, mqttSession, msg.payload());
163.         MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.SUBACK,
164. false, MqttQoS.AT_MOST_ONCE, false, 0);
165.         IntStream intStream = msg.payload().topicSubscriptions().stream().mapToInt(s -> s.qualityOfService().value());
166.         MqttSubAckPayload payload = new MqttSubAckPayload(intStream.toArray());
167.         ctx.writeAndFlush(MqttMessageFactory.newMessage(
168.                 fixedHeader,
169.                 MqttMessageIdVariableHeader.from(msg.variableHeader().messageId()),
170.                 payload));
171.     }
173. protected void onSubscribe(ChannelHandlerContext ctx, MqttSessionKey mqttSessionKey,
174.                                MqttSubscribePayload subscribePayload) throws Exception {
175.     }
177.     @Override
178. public void processSubAck(ChannelHandlerContext ctx, MqttSubAckMessage msg) throws Exception {
179.         MqttSessionKey mqttSession = ChannelUtils.getMqttSession(ctx.channel());
180. if (mqttSession == null) {
181.             log.error("sub ack, client address {} not authed", ctx.channel().remoteAddress());
182.             ctx.close();
183.         }
184.     }
186.     @Override
187. public void processUnsubscribe(ChannelHandlerContext ctx, MqttUnsubscribeMessage msg) throws Exception {
188.         MqttSessionKey mqttSession = ChannelUtils.getMqttSession(ctx.channel());
189. if (mqttSession == null) {
190.             log.error("unsub, client address {} not authed", ctx.channel().remoteAddress());
191.             ctx.close();
192.         }
193.     }
195.     @Override
196. public void processUnsubAck(ChannelHandlerContext ctx, MqttUnsubAckMessage msg) throws Exception {
197.         MqttSessionKey mqttSession = ChannelUtils.getMqttSession(ctx.channel());
198. if (mqttSession == null) {
199.             log.error("unsub ack, client address {} not authed", ctx.channel().remoteAddress());
200.             ctx.close();
201.         }
202.     }
204.     @Override
205. public void processPingReq(ChannelHandlerContext ctx, MqttMessage msg) throws Exception {
206.         ctx.writeAndFlush(MqttMessageUtil.pingResp());
207.     }
209.     @Override
210. public void processPingResp(ChannelHandlerContext ctx, MqttMessage msg) throws Exception {
211.         MqttSessionKey mqttSession = ChannelUtils.getMqttSession(ctx.channel());
212. if (mqttSession == null) {
213.             log.error("ping resp, client address {} not authed", ctx.channel().remoteAddress());
214.             ctx.close();
215.         }
216.     }
218.     @Override
219. public void processDisconnect(ChannelHandlerContext ctx) throws Exception {
220.         MqttSessionKey mqttSession = ChannelUtils.getMqttSession(ctx.channel());
221. if (mqttSession == null) {
222.             log.error("disconnect, client address {} not authed", ctx.channel().remoteAddress());
223.         }
224.         onDisconnect(mqttSession);
225.     }
227. protected void onDisconnect(MqttSessionKey mqttSessionKey) {
228.     }
230.     @Override
231. public void processAuth(ChannelHandlerContext ctx, MqttMessage msg) throws Exception {
232.         MqttSessionKey mqttSession = ChannelUtils.getMqttSession(ctx.channel());
233. if (mqttSession == null) {
234.             log.error("auth, client address {} not authed", ctx.channel().remoteAddress());
235.             ctx.close();
236.         }
237.     }
238. }

复制代码

可以看到，这里的AbstractProcessor主要是维护了MqttSessionKey，校验MqttSessionKey，并拦截publish中不支持的Qos2、Failure。同时，也影响了mqtt心跳请求。同样的，我们允许在onPublish、onSubscribe中抛出异常。
基于消息队列实现的mqtt网关的基础思想也比较简单，简而言之就是，有publish消息的时候向消息队列中生产消息。有订阅的时候就从消息队列中拉取消息。由此延伸出来，我们可能需要维护每个mqtt topic和producer、consumer的对应关系，因为像kafka、pulsar这些消息中间件的消费者都是区分topic的，片段通用代码如下:

1. protected final ReentrantReadWriteLock.ReadLock rLock;
3. protected final ReentrantReadWriteLock.WriteLock wLock;
5. protected final Map<MqttSessionKey, List<MqttTopicKey>> sessionProducerMap;
7. protected final Map<MqttSessionKey, List<MqttTopicKey>> sessionConsumerMap;
9. protected final Map<MqttTopicKey, P> producerMap;
11. protected final Map<MqttTopicKey, C> consumerMap;
13. public AbstractMqProcessor(MqttAuth mqttAuth) {
14. super(mqttAuth);
15.         ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
16.         rLock = lock.readLock();
17.         wLock = lock.writeLock();
18. this.sessionProducerMap = new HashMap<>();
19. this.sessionConsumerMap = new HashMap<>();
20. this.producerMap = new HashMap<>();
21. this.consumerMap = new HashMap<>();
22.     }
24.     @Override
25. protected void onConnect(MqttSessionKey mqttSessionKey) {
26.         wLock.lock();
27. try {
28.             sessionProducerMap.put(mqttSessionKey, new ArrayList<>());
29.             sessionConsumerMap.put(mqttSessionKey, new ArrayList<>());
30.         } finally {
31.             wLock.unlock();
32.         }
33.     }
35.     @Override
36. protected void onDisconnect(MqttSessionKey mqttSessionKey) {
37.         wLock.lock();
38. try {
39. // find producers
40.             List<MqttTopicKey> produceTopicKeys = sessionProducerMap.get(mqttSessionKey);
41. if (produceTopicKeys != null) {
42. for (MqttTopicKey mqttTopicKey : produceTopicKeys) {
43.                     P producer = producerMap.get(mqttTopicKey);
44. if (producer != null) {
45.                         ClosableUtils.close(producer);
46.                         producerMap.remove(mqttTopicKey);
47.                     }
48.                 }
49.             }
50.             sessionProducerMap.remove(mqttSessionKey);
51.             List<MqttTopicKey> consumeTopicKeys = sessionConsumerMap.get(mqttSessionKey);
52. if (consumeTopicKeys != null) {
53. for (MqttTopicKey mqttTopicKey : consumeTopicKeys) {
54.                     C consumer = consumerMap.get(mqttTopicKey);
55. if (consumer != null) {
56.                         ClosableUtils.close(consumer);
57.                         consumerMap.remove(mqttTopicKey);
58.                     }
59.                 }
60.             }
61.             sessionConsumerMap.remove(mqttSessionKey);
62.         } finally {
63.             wLock.unlock();
64.         }
65.     }
66. }

复制代码

kafka processor实现

1. 由于kafka producer不区分topic，我们可以在kafka processor中复用producer，在将来单个kafka producer的性能到达上限时，我们可以将kafka producer扩展为kafka producer列表进行轮询处理，消费者由于mqtt协议可能针对每个订阅topic有不同的行为，不合适复用同一个消费者实例。我们在构造函数中启动KafkaProducer

复制代码

1. private final KafkaProcessorConfig kafkaProcessorConfig;
3. private final KafkaProducer<String, ByteBuffer> producer;
5. public KafkaProcessor(MqttAuth mqttAuth, KafkaProcessorConfig kafkaProcessorConfig) {
6. super(mqttAuth);
7. this.kafkaProcessorConfig = kafkaProcessorConfig;
8. this.producer = createProducer();
9.     }
11. protected KafkaProducer<String, ByteBuffer> createProducer() {
12.         Properties properties = new Properties();
13.         properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProcessorConfig.getBootstrapServers());
14.         properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
15.         properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteBufferSerializer.class);
16. return new KafkaProducer<>(properties);
17.     }

复制代码

1. 处理MqttPublish消息，MqttPublish消息包含如下几个关键参数

复制代码

1. MqttQoS mqttQoS = publishMessage.fixedHeader().qosLevel();
2. String topic = publishMessage.variableHeader().topicName();
3. ByteBuffer byteBuffer = publishMessage.payload().nioBuffer();

复制代码

其中

• qos代表这条消息的质量级别，0没有任何保障，1代表至少一次，2代表恰好一次。当前仅支持qos0、qos1
  
• topicName就是topic的名称
  
• ByteBuffer就是消息的内容
  

根据topic、qos发送消息，代码如下

1.   String topic = msg.variableHeader().topicName();
2.         ProducerRecord<String, ByteBuffer> record = new ProducerRecord<>(topic, msg.payload().nioBuffer());
3. switch (msg.fixedHeader().qosLevel()) {
4. case AT_MOST_ONCE -> producer.send(record, (metadata, exception) -> {
5. if (exception != null) {
6.                     log.error("mqttSessionKey {} send message to kafka error", mqttSessionKey, exception);
7. return;
8.                 }
9.                 log.debug("mqttSessionKey {} send message to kafka success, topic {}, partition {}, offset {}",
10.                         mqttSessionKey, metadata.topic(), metadata.partition(), metadata.offset());
11.             });
12. case AT_LEAST_ONCE -> {
13. try {
14.                     RecordMetadata recordMetadata = producer.send(record).get();
15.                     log.info("mqttSessionKey {} send message to kafka success, topic {}, partition {}, offset {}",
16.                             mqttSessionKey, recordMetadata.topic(),
17.                             recordMetadata.partition(), recordMetadata.offset());
18.                     ctx.writeAndFlush(MqttMessageUtil.pubAckMessage(msg.variableHeader().packetId()));
19.                 } catch (Exception e) {
20.                     log.error("mqttSessionKey {} send message to kafka error", mqttSessionKey, e);
21.                 }
22.             }
23. case EXACTLY_ONCE, FAILURE -> throw new IllegalStateException(
24.                     String.format("mqttSessionKey %s can not reach here", mqttSessionKey));
25.         }

复制代码

处理订阅消息，我们暂时仅根据订阅的topic，创建topic进行消费即可，由于kafka原生客户端建议的消费代码模式如下

1. while (true) {
2.   ConsumerRecords<String, byte[]> records = consumer.poll(Duration.ofSeconds(1));
3. for (ConsumerRecord<String, byte[]> record : records) {
4. // do logic
5.   }
6. }

复制代码

我们需要切换到其他线程对consumer进行消息，书写一个KafkaConsumerListenerWrapper的wrapper，转换为listener异步消费模型

1. package io.github.protocol.mqtt.broker.processor;
3. import lombok.extern.slf4j.Slf4j;
4. import org.apache.kafka.clients.admin.AdminClient;
5. import org.apache.kafka.clients.admin.AdminClientConfig;
6. import org.apache.kafka.clients.admin.KafkaAdminClient;
7. import org.apache.kafka.clients.admin.NewTopic;
8. import org.apache.kafka.clients.admin.TopicDescription;
9. import org.apache.kafka.clients.consumer.ConsumerConfig;
10. import org.apache.kafka.clients.consumer.ConsumerRecord;
11. import org.apache.kafka.clients.consumer.ConsumerRecords;
12. import org.apache.kafka.clients.consumer.KafkaConsumer;
13. import org.apache.kafka.common.errors.UnknownTopicOrPartitionException;
14. import org.apache.kafka.common.errors.WakeupException;
15. import org.apache.kafka.common.serialization.ByteArrayDeserializer;
16. import org.apache.kafka.common.serialization.StringDeserializer;
18. import java.time.Duration;
19. import java.util.Collections;
20. import java.util.Properties;
21. import java.util.concurrent.ExecutionException;
23. @Slf4j
24. public class KafkaConsumerListenerWrapper implements AutoCloseable {
26. private final AdminClient adminClient;
28. private final KafkaConsumer<String, byte[]> consumer;
30. public KafkaConsumerListenerWrapper(KafkaProcessorConfig config, String username) {
31.         Properties adminProperties = new Properties();
32.         adminProperties.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, config.getBootstrapServers());
33. this.adminClient = KafkaAdminClient.create(adminProperties);
34.         Properties properties = new Properties();
35.         properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, config.getBootstrapServers());
36.         properties.put(ConsumerConfig.GROUP_ID_CONFIG, username);
37.         properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
38.         properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class);
39. this.consumer = new KafkaConsumer<>(properties);
40.     }
42. public void start(String topic, KafkaMessageListener listener) throws Exception {
43. try {
44.             TopicDescription topicDescription = adminClient.describeTopics(Collections.singletonList(topic))
45.                     .values().get(topic).get();
46.             log.info("topic info is {}", topicDescription);
47.         } catch (ExecutionException ee) {
48. if (ee.getCause() instanceof UnknownTopicOrPartitionException) {
49.                 log.info("topic {} not exist, create it", topic);
50.                 adminClient.createTopics(Collections.singletonList(new NewTopic(topic, 1, (short) 1)));
51.             } else {
52.                 log.error("find topic info {} error", topic, ee);
53.             }
54.         } catch (Exception e) {
55. throw new IllegalStateException("find topic info error", e);
56.         }
57.         consumer.subscribe(Collections.singletonList(topic));
58.         log.info("consumer topic {} start", topic);
59. new Thread(() -> {
60. try {
61. while (true) {
62.                     ConsumerRecords<String, byte[]> records = consumer.poll(Duration.ofSeconds(1));
63. for (ConsumerRecord<String, byte[]> record : records) {
64.                         listener.messageReceived(record);
65.                     }
66.                 }
67.             } catch (WakeupException we) {
68.                 consumer.close();
69.             } catch (Exception e) {
70.                 log.error("consumer topic {} consume error", topic, e);
71.                 consumer.close();
72.             }
73.         }).start();
74.         Thread.sleep(5_000);
75.     }
77.     @Override
78. public void close() throws Exception {
79.         log.info("wake up {} consumer", consumer);
80.         consumer.wakeup();
81.     }
82. }
83.     @Override
84. protected void onSubscribe(ChannelHandlerContext ctx, MqttSessionKey mqttSessionKey,
85.                                MqttSubscribePayload subscribePayload) throws Exception {
86. for (MqttTopicSubscription topicSubscription : subscribePayload.topicSubscriptions()) {
87.             KafkaConsumerListenerWrapper consumer = createConsumer(mqttSessionKey, topicSubscription.topicName());
88.             subscribe(ctx, consumer, topicSubscription.topicName());
89.         }
90.     }
92. private KafkaConsumerListenerWrapper createConsumer(MqttSessionKey mqttSessionKey, String topic) {
93.         MqttTopicKey mqttTopicKey = new MqttTopicKey();
94.         mqttTopicKey.setTopic(topic);
95.         mqttTopicKey.setMqttSessionKey(mqttSessionKey);
97.         wLock.lock();
98. try {
99.             KafkaConsumerListenerWrapper consumer = consumerMap.get(mqttTopicKey);
100. if (consumer == null) {
101.                 consumer = new KafkaConsumerListenerWrapper(kafkaProcessorConfig, mqttSessionKey.getUsername());
102.                 sessionConsumerMap.compute(mqttSessionKey, (mqttSessionKey1, mqttTopicKeys) -> {
103. if (mqttTopicKeys == null) {
104.                         mqttTopicKeys = new ArrayList<>();
105.                     }
106.                     mqttTopicKeys.add(mqttTopicKey);
107. return mqttTopicKeys;
108.                 });
109.                 consumerMap.put(mqttTopicKey, consumer);
110.             }
111. return consumer;
112.         } finally {
113.             wLock.unlock();
114.         }
115.     }
117. protected void subscribe(ChannelHandlerContext ctx,
118.                              KafkaConsumerListenerWrapper consumer, String topic) throws Exception {
119.         BoundInt boundInt = new BoundInt(65535);
120.         consumer.start(topic, record -> {
121.             log.info("receive message from kafka, topic {}, partition {}, offset {}",
122.                     record.topic(), record.partition(), record.offset());
123.             MqttPublishMessage mqttPublishMessage = MqttMessageUtil.publishMessage(
124.                     MqttQoS.AT_LEAST_ONCE, topic, boundInt.nextVal(), record.value());
125.             ctx.writeAndFlush(mqttPublishMessage);
126.         });
127.     }

复制代码

在上述的代码中，有一个需要通篇注意的点：日志打印的时候，注意要将关键的信息携带，比如：topic、mqtt username、mqtt clientId等，在写demo的时候没有感觉，但是海量请求下需要定位问题的时候，就知道这些信息的关键之处了。
使用BountInt这个简单的工具类来生成从0~65535的packageId，满足协议的要求
pulsar processor实现

pulsar相比kafka来说，更适合作为mqtt协议的代理。原因有如下几点：

• pulsar支持百万topic、topic实现更轻量
  
• pulsar原生支持listener的消费模式，不需要每个消费者启动一个线程
  
• pulsar支持share的消费模式，消费模式更灵活
  
• pulsar消费者的subscribe可确保成功创建订阅，相比kafka的消费者没有这样的语义保障
  

1. protected final ReentrantReadWriteLock.ReadLock rLock;
3. protected final ReentrantReadWriteLock.WriteLock wLock;
5. protected final Map<MqttSessionKey, List<MqttTopicKey>> sessionProducerMap;
7. protected final Map<MqttSessionKey, List<MqttTopicKey>> sessionConsumerMap;
9. protected final Map<MqttTopicKey, Producer<byte[]>> producerMap;
11. protected final Map<MqttTopicKey, Consumer<byte[]>> consumerMap;
13. private final PulsarProcessorConfig pulsarProcessorConfig;
15. private final PulsarAdmin pulsarAdmin;
17. private final PulsarClient pulsarClient;
19. public PulsarProcessor(MqttAuth mqttAuth, PulsarProcessorConfig pulsarProcessorConfig) {
20. super(mqttAuth);
21.         ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
22.         rLock = lock.readLock();
23.         wLock = lock.writeLock();
24. this.sessionProducerMap = new HashMap<>();
25. this.sessionConsumerMap = new HashMap<>();
26. this.producerMap = new HashMap<>();
27. this.consumerMap = new HashMap<>();
28. this.pulsarProcessorConfig = pulsarProcessorConfig;
29. try {
30. this.pulsarAdmin = PulsarAdmin.builder()
31.                     .serviceHttpUrl(pulsarProcessorConfig.getHttpUrl())
32.                     .build();
33. this.pulsarClient = PulsarClient.builder()
34.                     .serviceUrl(pulsarProcessorConfig.getServiceUrl())
35.                     .build();
36.         } catch (Exception e) {
37. throw new IllegalStateException("Failed to create pulsar client", e);
38.         }
39.     }

复制代码

处理publish消息

1.    @Override
2. protected void onPublish(ChannelHandlerContext ctx, MqttSessionKey mqttSessionKey,
3.                              MqttPublishMessage msg) throws Exception {
4.         String topic = msg.variableHeader().topicName();
5.         Producer<byte[]> producer = getOrCreateProducer(mqttSessionKey, topic);
6.         int len = msg.payload().readableBytes();
7.         byte[] messageBytes = new byte[len];
8.         msg.payload().getBytes(msg.payload().readerIndex(), messageBytes);
9. switch (msg.fixedHeader().qosLevel()) {
10. case AT_MOST_ONCE -> producer.sendAsync(messageBytes).
11.                     thenAccept(messageId -> log.info("clientId [{}],"
12.                                     + " username [{}]. send message to pulsar success messageId: {}",
13.                             mqttSessionKey.getClientId(), mqttSessionKey.getUsername(), messageId))
14.                     .exceptionally((e) -> {
15.                         log.error("clientId [{}], username [{}]. send message to pulsar fail: ",
16.                                 mqttSessionKey.getClientId(), mqttSessionKey.getUsername(), e);
17. return null;
18.                     });
19. case AT_LEAST_ONCE -> {
20. try {
21.                     MessageId messageId = producer.send(messageBytes);
22.                     MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBACK,
23. false, MqttQoS.AT_MOST_ONCE, false, 0);
24.                     MqttPubAckMessage pubAckMessage = (MqttPubAckMessage) MqttMessageFactory.newMessage(fixedHeader,
25.                             MqttMessageIdVariableHeader.from(msg.variableHeader().packetId()), null);
26.                     log.info("clientId [{}], username [{}]. send pulsar success. messageId: {}",
27.                             mqttSessionKey.getClientId(), mqttSessionKey.getUsername(), messageId);
28.                     ctx.writeAndFlush(pubAckMessage);
29.                 } catch (PulsarClientException e) {
30.                     log.error("clientId [{}], username [{}]. send pulsar error: {}",
31.                             mqttSessionKey.getClientId(), mqttSessionKey.getUsername(), e.getMessage());
32.                 }
33.             }
34. case EXACTLY_ONCE, FAILURE -> throw new IllegalStateException(
35.                     String.format("mqttSessionKey %s can not reach here", mqttSessionKey));
36.         }
37.     }
39. private Producer<byte[]> getOrCreateProducer(MqttSessionKey mqttSessionKey, String topic) throws Exception {
40.         MqttTopicKey mqttTopicKey = new MqttTopicKey();
41.         mqttTopicKey.setTopic(topic);
42.         mqttTopicKey.setMqttSessionKey(mqttSessionKey);
44.         rLock.lock();
45. try {
46.             Producer<byte[]> producer = producerMap.get(mqttTopicKey);
47. if (producer != null) {
48. return producer;
49.             }
50.         } finally {
51.             rLock.unlock();
52.         }
54.         wLock.lock();
55. try {
56.             Producer<byte[]> producer = producerMap.get(mqttTopicKey);
57. if (producer == null) {
58.                 producer = createProducer(topic);
59.                 sessionProducerMap.compute(mqttSessionKey, (mqttSessionKey1, mqttTopicKeys) -> {
60. if (mqttTopicKeys == null) {
61.                         mqttTopicKeys = new ArrayList<>();
62.                     }
63.                     mqttTopicKeys.add(mqttTopicKey);
64. return mqttTopicKeys;
65.                 });
66.                 producerMap.put(mqttTopicKey, producer);
67.             }
68. return producer;
69.         } finally {
70.             wLock.unlock();
71.         }
72.     }
74. protected Producer<byte[]> createProducer(String topic) throws Exception {
75. return pulsarClient.newProducer(Schema.BYTES).topic(topic).create();
76.     }

复制代码

处理subscribe消息

1.     @Override
2. protected void onSubscribe(ChannelHandlerContext ctx, MqttSessionKey mqttSessionKey,
3.                                MqttSubscribePayload subscribePayload) throws Exception {
4. for (MqttTopicSubscription topicSubscription : subscribePayload.topicSubscriptions()) {
5.             subscribe(ctx, mqttSessionKey, topicSubscription.topicName());
6.         }
7.     }
9. protected void subscribe(ChannelHandlerContext ctx, MqttSessionKey mqttSessionKey,
10.                              String topic) throws Exception {
11.         MqttTopicKey mqttTopicKey = new MqttTopicKey();
12.         mqttTopicKey.setTopic(topic);
13.         mqttTopicKey.setMqttSessionKey(mqttSessionKey);
15.         wLock.lock();
16. try {
17.             Consumer<byte[]> consumer = consumerMap.get(mqttTopicKey);
18. if (consumer == null) {
19.                 consumer = createConsumer(ctx, mqttSessionKey.getUsername(), topic);
20.                 sessionConsumerMap.compute(mqttSessionKey, (mqttSessionKey1, mqttTopicKeys) -> {
21. if (mqttTopicKeys == null) {
22.                         mqttTopicKeys = new ArrayList<>();
23.                     }
24.                     mqttTopicKeys.add(mqttTopicKey);
25. return mqttTopicKeys;
26.                 });
27.                 consumerMap.put(mqttTopicKey, consumer);
28.             }
29.         } finally {
30.             wLock.unlock();
31.         }
32.     }
34. protected Consumer<byte[]> createConsumer(ChannelHandlerContext ctx, String username,
35.                                               String topic) throws Exception {
36.         BoundInt boundInt = new BoundInt(65535);
37. try {
38.             PartitionedTopicStats partitionedStats = pulsarAdmin.topics().getPartitionedStats(topic, false);
39.             log.info("topic {} partitioned stats {}", topic, partitionedStats);
40.         } catch (PulsarAdminException.NotFoundException nfe) {
41.             log.info("topic {} not found", topic);
42.             pulsarAdmin.topics().createPartitionedTopic(topic, 1);
43.         }
44. return pulsarClient.newConsumer(Schema.BYTES).topic(topic)
45.                 .messageListener((consumer, msg) -> {
46.                     log.info("receive message from pulsar, topic {}, message {}", topic, msg.getMessageId());
47.                     MqttPublishMessage mqttPublishMessage = MqttMessageUtil.publishMessage(
48.                             MqttQoS.AT_LEAST_ONCE, topic, boundInt.nextVal(), msg.getData());
49.                     ctx.writeAndFlush(mqttPublishMessage);
50.                 })
51.                 .subscriptionName(username).subscribe();
52.     }

复制代码

集成测试

kafka

我们可以通过embedded-kafka-java这个项目来启动用做单元测试的kafka broker。通过如下的group引入依赖

1. <dependency>
2. <groupId>io.github.embedded-middleware</groupId>
3. <artifactId>embedded-kafka-core</artifactId>
4. <version>0.0.2</version>
5. <scope>test</scope>
6. </dependency>

复制代码

我们就可以通过如下的代码启动基于kafka的mqtt broker

1. @Slf4j
2. public class MqttKafkaTestUtil {
4. public static MqttServer setupMqttKafka() throws Exception {
5.         EmbeddedKafkaServer embeddedKafkaServer = new EmbeddedKafkaServer();
6. new Thread(() -> {
7. try {
8.                 embeddedKafkaServer.start();
9.             } catch (Exception e) {
10.                 log.error("kafka broker started exception ", e);
11.             }
12.         }).start();
13.         Thread.sleep(5_000);
14.         MqttServerConfig mqttServerConfig = new MqttServerConfig();
15.         mqttServerConfig.setPort(0);
16.         mqttServerConfig.setProcessorType(ProcessorType.KAFKA);
17.         KafkaProcessorConfig kafkaProcessorConfig = new KafkaProcessorConfig();
18.         kafkaProcessorConfig.setBootstrapServers(String.format("localhost:%d", embeddedKafkaServer.getKafkaPort()));
19.         mqttServerConfig.setKafkaProcessorConfig(kafkaProcessorConfig);
20.         MqttServer mqttServer = new MqttServer(mqttServerConfig);
21. new Thread(() -> {
22. try {
23.                 mqttServer.start();
24.             } catch (Exception e) {
25.                 log.error("mqsar broker started exception ", e);
26.             }
27.         }).start();
28.         Thread.sleep(5000L);
29. return mqttServer;
30.     }
32. }

复制代码

kafka端到端测试用例，比较简单，通过mqtt client publish一条消息，然后消费出来

1. @Log4j2
2. public class MqttKafkaPubSubTest {
4.     @Test
5. public void pubSubTest() throws Exception {
6.         MqttServer mqttServer = MqttKafkaTestUtil.setupMqttKafka();
7.         String topic = UUID.randomUUID().toString();
8.         String content = "test-msg";
9.         String broker = String.format("tcp://localhost:%d", mqttServer.getPort());
10.         String clientId = UUID.randomUUID().toString();
11.         MemoryPersistence persistence = new MemoryPersistence();
12.         MqttClient sampleClient = new MqttClient(broker, clientId, persistence);
13.         MqttConnectOptions connOpts = new MqttConnectOptions();
14.         connOpts.setUserName(UUID.randomUUID().toString());
15.         connOpts.setPassword(UUID.randomUUID().toString().toCharArray());
16.         connOpts.setCleanSession(true);
17.         log.info("Mqtt connecting to broker");
18.         sampleClient.connect(connOpts);
19.         CompletableFuture<String> future = new CompletableFuture<>();
20.         log.info("Mqtt subscribing");
21.         sampleClient.subscribe(topic, (s, mqttMessage) -> {
22.             log.info("messageArrived");
23.             future.complete(mqttMessage.toString());
24.         });
25.         log.info("Mqtt subscribed");
26.         MqttMessage message = new MqttMessage(content.getBytes());
27.         message.setQos(1);
28.         log.info("Mqtt message publishing");
29.         sampleClient.publish(topic, message);
30.         log.info("Mqtt message published");
31.         TimeUnit.SECONDS.sleep(3);
32.         sampleClient.disconnect();
33.         String msg = future.get(5, TimeUnit.SECONDS);
34.         Assertions.assertEquals(content, msg);
35.     }
37. }

复制代码

pulsar

我们可以通过embedded-pulsar-java这个项目来启动用做单元测试的pulsar broker。通过如下的group引入依赖

1. <dependency>
2. <groupId>io.github.embedded-middleware</groupId>
3. <artifactId>embedded-pulsar-core</artifactId>
4. <version>0.0.2</version>
5. <scope>test</scope>
6. </dependency>

复制代码

我们就可以通过如下的代码启动基于pulsar的mqtt broker

1. @Slf4j
2. public class MqttPulsarTestUtil {
4. public static MqttServer setupMqttPulsar() throws Exception {
5.         EmbeddedPulsarServer embeddedPulsarServer = new EmbeddedPulsarServer();
6.         embeddedPulsarServer.start();
7.         MqttServerConfig mqttServerConfig = new MqttServerConfig();
8.         mqttServerConfig.setPort(0);
9.         mqttServerConfig.setProcessorType(ProcessorType.PULSAR);
10.         PulsarProcessorConfig pulsarProcessorConfig = new PulsarProcessorConfig();
11.         pulsarProcessorConfig.setHttpUrl(String.format("http://localhost:%d", embeddedPulsarServer.getWebPort()));
12.         pulsarProcessorConfig.setServiceUrl(String.format("pulsar://localhost:%d", embeddedPulsarServer.getTcpPort()));
13.         mqttServerConfig.setPulsarProcessorConfig(pulsarProcessorConfig);
14.         MqttServer mqttServer = new MqttServer(mqttServerConfig);
15. new Thread(() -> {
16. try {
17.                 mqttServer.start();
18.             } catch (Exception e) {
19.                 log.error("mqsar broker started exception ", e);
20.             }
21.         }).start();
22.         Thread.sleep(5000L);
23. return mqttServer;
24.     }
25. }

复制代码

pulsar端到端测试用例，比较简单，通过mqtt client publish一条消息，然后消费出来

1. @Log4j2
2. public class MqttPulsarPubSubTest {
4.     @Test
5. public void pubSubTest() throws Exception {
6.         MqttServer mqttServer = MqttPulsarTestUtil.setupMqttPulsar();
7.         String topic = UUID.randomUUID().toString();
8.         String content = "test-msg";
9.         String broker = String.format("tcp://localhost:%d", mqttServer.getPort());
10.         String clientId = UUID.randomUUID().toString();
11.         MemoryPersistence persistence = new MemoryPersistence();
12.         MqttClient sampleClient = new MqttClient(broker, clientId, persistence);
13.         MqttConnectOptions connOpts = new MqttConnectOptions();
14.         connOpts.setUserName(UUID.randomUUID().toString());
15.         connOpts.setPassword(UUID.randomUUID().toString().toCharArray());
16.         connOpts.setCleanSession(true);
17.         log.info("Mqtt connecting to broker");
18.         sampleClient.connect(connOpts);
19.         CompletableFuture<String> future = new CompletableFuture<>();
20.         log.info("Mqtt subscribing");
21.         sampleClient.subscribe(topic, (s, mqttMessage) -> {
22.             log.info("messageArrived");
23.             future.complete(mqttMessage.toString());
24.         });
25.         log.info("Mqtt subscribed");
26.         MqttMessage message = new MqttMessage(content.getBytes());
27.         message.setQos(1);
28.         log.info("Mqtt message publishing");
29.         sampleClient.publish(topic, message);
30.         log.info("Mqtt message published");
31.         TimeUnit.SECONDS.sleep(3);
32.         sampleClient.disconnect();
33.         String msg = future.get(5, TimeUnit.SECONDS);
34.         Assertions.assertEquals(content, msg);
35.     }
36. }

复制代码

性能优化

这里我们简单描述几个性能优化点，像一些调整线程数、buffer大小这类的参数调整就不在这里赘述了，这些需要具体的性能压测来决定参数的设置。
在linux上使用Epoll网络模型

[code]public class EventLoopUtil { /**     * @return an EventLoopGroup suitable for the current platform */ public static EventLoopGroup newEventLoopGroup(int nThreads, ThreadFactory threadFactory) { if (Epoll.isAvailable()) { return new EpollEventLoopGroup(nThreads, threadFactory);        } else { return new NioEventLoopGroup(nThreads, threadFactory);        }    } public static Class