学习 RocketMQ，需要搞懂两个东西：通信和存储。这里花了一点时间写了一个 RocketMQ 的 wireshark lua 插件，过程挺有意思，写出来记录一下。
通过阅读这篇文章，你会了解到下面这些知识。
- wireshark lua 插件的骨架代码如何编写
- 插件版 Hello World 如何实现
- RocketMQ 的基本通信协议格式
- RocketMQ 在 PULL 有消息时的 Body 格式是什么样的
初探 Hello World 插件
从 wireshark 的 about 页面可以看到现在它支持的 Lua 版本，下面是我 v3.0.6 版本的 wireshark 对应的页面。
可以看到，目前支持的 Lua 版本是 5.2.4。下面我们来看一段骨架代码。
代码语言：javascript复制-- 声明协议
local NAME = "RocketMQ"
local PORTS = { 9876, 10911 }
local proto = Proto.new(NAME, "RocketMQ Protocol")
-- 声明 dissector 函数，处理包
function proto.dissector(tvb, pinfo, tree)
print("load plugin...demo")
pinfo.cols.protocol = proto.name;
pinfo.cols.info = "Hello, World"
end
-- 注册 dissector 到 wireshark
for _, port in ipairs(PORTS) do
DissectorTable.get("tcp.port"):add(port, proto)
end
找到 wireshark 插件目录，在我的电脑上这个路径是 /Applications/Wireshark.app/Contents/Resources/share/wireshark/，修改其中的 init.lua 文件
vim /Applications/Wireshark.app/Contents/Resources/share/wireshark/init.lua
增加一行加载上面 lua 文件的 dofile 调用。
代码语言：javascript复制...
dofile("/path/to/demo.lua")
执行前后效果如下。
解析 RocketMQ 协议
RocketMQ 的通信协议是比较简单的，整体的协议格式如下所示。
RocketMQ 的通信协议由四部分组成：
- 第一部分：头 4 个字节表示剩下三部分的总长度（不包括自己这 4 个字节）
- 第二部分：接下来的 4 个字节表示 Header 部分的长度
- 第三部分：接下来的 Header Length 长度的内容为协议头，是用 json 序列化后存储，主要用来表示不同的请求响应类型
- 第四部分：body 内容
以一个实际的包为例：
头四个字节 00 00 01 9b 表示整个包的长度 411(0x019b)，接下来的四个字节 00 00 00 d4 表示 Header Length，这里为 212(0xD4)，接下来的 212 个字节表示Header 的内容，可以看到这是一段 json 的字符串，最后的 195（411-4-212) 个字节表示 Body 的真正内容，具体的消息格式下面会再讲到。
接下来我们来写解析的程序。
解析的逻辑在 proto.dissector 方法中进行，它的签名如下所示。
代码语言：javascript复制function proto.dissector(tvb, pinfo, tree)
end
这些参数的释义如下：
- tvb 是 "Testy Virtual Buffer" 的缩写，是包含数据包的 buffer 内容
- pinfo 是 Packet Information 的缩写，表示 Packet 包相关的信息，可以获取包的源端口、目标端口等信息。
- tree 表示 wireshark UI 界面的展示树，解析包得到的信息都会添加到这个有层级关系的树中。
接下来我们把 RocketMQ 通信的四个部分展示到 wireshark 中。修改 proto.dissector 函数的代码如下所示。
代码语言：javascript复制function proto.dissector(tvb, pinfo, tree)
print("load plugin...demo")
local subtree = tree:add(proto, tvb())
pinfo.cols.protocol = proto.name;
pinfo.cols.info = ""
local length = tvb(0, 4):uint()
subtree:add("Total Length", length)
local headerLength = tvb(4, 4):uint()
subtree:add("Header Length", headerLength)
local headerData = tvb(8, headerLength):string()
subtree:add("Header", headerData)
local bodyDataLen = length - 4 - headerLength
local bodyData = tvb(8 headerLength, bodyDataLen):string()
subtree:add("Body", bodyData)
end
重新加载 lua 脚本，可以看到 Wireshark 中 RocketMQ 协议的几个部分已经显示出来了。
为了能区分是通信 Request 还是 Response，我们可以通过目标端口号来区分，新增一个方法。
代码语言：javascript复制function isRequest(pinfo)
local dstPort = pinfo.dst_port;
for _, port in ipairs(PORTS) do
if (dstPort == port) then
return true
end
end
return false
end
在 proto.dissector 中新增对请求和响应的区分，增加更可读的描述。
代码语言：javascript复制if (isRequest(pinfo)) then
pinfo.cols.info:append("[REQUEST]" .. "↑↑↑")
else
pinfo.cols.info:append("[RESPONSE]" .. "↓↓↓")
end
效果如下所示。
接下来我们要做的就是把 json 做解析，展示的更好看一点，先来看 header 和 body 为 json 格式时请求和响应。增加一个递归的方法，统一处理 json 格式的数据。
代码语言：javascript复制-- k,v 分别表示 json 的 key 和 value，tree 表示 UI 树
function parseAndAddTree(k, v, tree)
if (type(v) == 'table') then
local sizeStr = ""
if (#v > 0) then
sizeStr = "size: " .. #v
end;
local childTree = tree:add(k, sizeStr, tree)
for key, value in pairs(v) do
parseAndAddTree(key, value, childTree)
end
else
tree:add(k .. ":", json.stringify(v))
end
end
在 proto.dissector 方法中增加 Header 的解析，如下所示。
代码语言：javascript复制local subtree = tree:add(protoMQ, tvb())
local headerTree = subtree:add("Header", "")
-- 解析 json
local header = json.parse(headerData, 1, "}")
for k, v in pairs(header) do
parseAndAddTree(k, v, headerTree)
end
重新加载运行上面的代码，效果如下所示。
同时我们也可以在 RocketMQ 的源码中找到请求和响应 code 对应的更可读的字符串表示，
代码语言：javascript复制local requestCodeMap = {
[10] = "SEND_MESSAGE",
[11] = "PULL_MESSAGE",
[12] = "QUERY_MESSAGE",
...
}
local responseCode = {
[0] = "SUCCESS",
[1] = "SYSTEM_ERROR",
[2] = "SYSTEM_BUSY",
}
如果 Body 是 json 字符串的话也可以用这种方式来处理，如下所示。
但是在一些情况下，Body 并不是用 json 字符串来表示的，比如在 PULL 消息的时候，如果服务器有返回可消费的消息，这时 Body 中存储的并不是字符串，而是 RocketMQ 自定义的消息格式，如下所示。
写这段解析是个体力活，我参照 RocketMQ 的 Java 源码实现了一个 lua 版本，完整的代码如下所示，
代码语言：javascript复制function decodeMessageExt(bodyTree, pinfo, bodyData)
local bodyTree = bodyTree:add("Body", "")
pinfo.cols.info:append(">>>>#FOUND#")
local offset = 0;
bodyTree:add("totalSize", bodyData(offset, 4):int())
offset = offset 4;
local magicCode = string.format("0X%8.8X", bodyData(offset, 4):uint())
bodyTree:add("magicCode", magicCode)
offset = offset 4;
bodyTree:add("bodyCRC", bodyData(offset, 4):int())
offset = offset 4;
bodyTree:add("queueId", bodyData(offset, 4):int())
offset = offset 4;
bodyTree:add("flag", bodyData(offset, 4):int())
offset = offset 4;
bodyTree:add("queueOffset", bodyData(offset, 8):int64():tonumber())
offset = offset 8;
bodyTree:add("physicOffset", bodyData(offset, 8):int64():tonumber())
offset = offset 8;
bodyTree:add("sysFlag", bodyData(offset, 4):int())
offset = offset 4;
bodyTree:add("bornTimeStamp", bodyData(offset, 8):int64():tonumber())
offset = offset 8;
local bornHost = bodyData(offset, 1):uint()
.. "." .. bodyData(offset 1, 1):uint()
.. "." .. bodyData(offset 2, 1):uint()
.. "." .. bodyData(offset 3, 1):uint()
bodyTree:add("bornHost", bornHost)
offset = offset 4;
bodyTree:add("port", bodyData(offset, 4):int())
offset = offset 4;
bodyTree:add("storeTimestamp", bodyData(offset, 8):int64():tonumber())
offset = offset 8;
local storeHost = bodyData(offset, 1):uint()
.. "." .. bodyData(offset 1, 1):uint()
.. "." .. bodyData(offset 2, 1):uint()
.. "." .. bodyData(offset 3, 1):uint()
bodyTree:add("storeHost", storeHost)
offset = offset 4;
bodyTree:add("storePort", bodyData(offset, 4):int())
offset = offset 4;
--13 RECONSUMETIMES
bodyTree:add("reconsumeTimes", bodyData(offset, 4):int())
offset = offset 4;
--14 Prepared Transaction Offset
bodyTree:add("preparedTransactionOffset", bodyData(offset, 8):int64():tonumber())
offset = offset 8;
--15 BODY
local bodyLen = bodyData(offset, 4):int()
-- bodyTree:add("bodyLen", bodyLen)
offset = offset 4;
bodyTree:add("body:", bodyData(offset, bodyLen):string())
offset = offset bodyLen;
--16 TOPIC
local topicLen = bodyData(offset, 1):int()
offset = offset 1;
-- bodyTree:add("topicLen", topicLen)
local topic = bodyData(offset, topicLen):string()
bodyTree:add("topic:", topic)
pinfo.cols.info:append(" topic:" .. topic)
offset = offset topicLen;
--17 properties
local propertiesLength = bodyData(offset, 2):int()
offset = offset 2;
bodyTree:add("propertiesLength", propertiesLength)
if (propertiesLength > 0) then
local propertiesStr = bodyData(offset, propertiesLength):string()
offset = offset propertiesLength
local propertiesTree = bodyTree:add("propertiesStr", "size: " .. propertiesLength)
for k, v in string.gmatch(propertiesStr, "(%w )1(%w )") do
propertiesTree:add(k, v)
end
end
end
运行的效果如下所示。
完整的代码我放在了 github 上: https://github.com/arthur-zhang/rocketmq-wireshark-lua-plugin , 有兴趣的同学可以看看。除了前面文章中的那些功能，还有实现将 topic 等有用的信息提取到 Info 那一栏，方便查看通信的过程。
后记
没事折腾折腾还挺有意思的，在后台开发中 Lua 这门胶水语言除了在 OpenResty、Redis 中有不少用处之外，还有不少有趣的用途等待我们去发掘。
通过写这个插件，我自己对 RocketMQ 通信的细节更加清楚，后面可以再写写 RocketMQ 通信细节的文章。
