MQTT应用已经非常广泛了，目前笔者团队需要用lua来实现mqtt通讯，而服务端需要使用.Net Core来实现，以下是相关技术研究的一个总结。
本文主要介绍.Net Core和Lua的MQTT通讯示例。所有源代码可以从这里下载。
1. MQTT 协议
开发之前，非常有必要仔细的了解一下MQTT协议的基本介绍和报文结构，可以参考简书用户“技术学习”的MQTT专栏,里面关于协议介绍的非常详细。
2. MQTT Lua
MQTT的lua实现并不是一个常见的需求，http://mqtt.org/上能下到的版本只支持到MQTT协议3.1版本，最新的3.1.1版本并不支持。
3.1.1版本改动不大，但是并不兼容3.1版本。
最后我这边使用的是Github上的个人开发者的一个版本,和3.1版本比较，改变了：
- CONNECT消息可变头部协议名称MQIsdp被改为MQTT
- CONNECT消息可变头部协议版本号，由0x03变成了0x04
- 增加了函数
wait_for_connack，发送CONNECT消息后必须在设置的时间（3秒）内获取到服务端返回的CONNACK消息才确认链接已经成功。
示例实现MQTT的客户端，只包括3个文件：
- mqtt.lua : 封装好的mqtt lua库
- utility.lua: mqtt.lua用到的工具相关封装
- testmqtt.lua: 我们编写的示例，调用mqtt.lua,以下是部分代码
local args={
host="192.168.1.69",
port=5555,
id="client1", -- MQTT client identifier,3.1.1版本支持长度超过23
topic_p="test1", -- 发布消息的topic
topic_s="test2", --订阅消息的topic
debug=true
}
......
local mqtt_client = MQTT.client.create(args.host, args.port, callback) --以host，port创建一个client对象，callback为接受消息的回调函数
--如果链接失败，重试直到链接成功
local connect_error = mqtt_client:connect(args.id)
while connect_error~=nil do
connect_error = mqtt_client:connect(args.id)
end
mqtt_client:publish(args.topic_p, "*** Lua test start ***")
mqtt_client:subscribe({ args.topic_s })
......
while (error_message == nil and running) do
error_message = mqtt_client:handler()
-- 循环publish消息
if (error_message == nil) then
mqtt_client:publish(args.topic_p, "*** Lua test message ***")
socket.sleep(1.0) -- seconds
end
end
......
示例的运行依赖luasocket，在ubuntu上用Luajit运行的结果是：
image.png
3. DotNetty
.Net Core可以使用MQTTnet结合Apollo服务器来实现MQTT通讯，后期再尝试。
目前使用DotNetty做MQTT服务端，DotNetty是微软的Azure团队用C#实现的Netty版本，DotNetty的简单介绍可以参考# 使用DotNetty编写跨平台网络通信程序. DotNetty的资料不多，可以参考Netty的资料来学习或者直接下载DotNetty源代码，里面有Test,Example可以参考。
DotNetty提供的MQTTDecoder和MqttDecoder直接把MQTT报文二进制转换成Packet对象，直接对对象处理就简单多了。以下是相关关键代码。
var bootstrap = new ServerBootstrap();
bootstrap
.Group(bossGroup, workerGroup) // 设置主和工作线程组
.Channel<TcpServerSocketChannel>() // 设置通道模式为TcpSocket
.ChildHandler(new ActionChannelInitializer<ISocketChannel>(channel =>
{ //工作线程连接器 是设置了一个管道，服务端主线程所有接收到的信息都会通过这个管道一层层往下传输
//同时所有出栈的消息 也要这个管道的所有处理器进行一步步处理
IChannelPipeline pipeline = channel.Pipeline;
pipeline.AddLast("decoder", new MqttDecoder(true, 256 * 1024));
pipeline.AddLast("encoder", new MqttEncoder());
pipeline.AddLast("echo", new EchoServerHandler());
}));
public override void ChannelRead(IChannelHandlerContext context, object message)
{
Console.WriteLine("Received from client...");
var buffer = message as Packet;
Console.WriteLine("PacketTyp="+buffer.PacketType);
if (buffer != null)
{
if (buffer is ConnectPacket)
{
Console.WriteLine("Connect from mqtt client,id=: " + ((ConnectPacket)buffer).ClientId);
context.WriteAsync(new ConnAckPacket
{
ReturnCode = ConnectReturnCode.Accepted,
SessionPresent = true
});//写入输出流
}
else
{
context.WriteAsync(new PublishPacket(QualityOfService.ExactlyOnce, false, false)
{
PacketId = GetRandomPacketId(),
TopicName = "test2",
Payload = Unpooled.WrappedBuffer(Encoding.UTF8.GetBytes("message from server"))
});//写入输出流
}
}
}
最后运行的效果是
image.png
接受到3种连接类型，进一步修改就可以根据不同的类型做不同的操作。
