我的mqtt协议和emqttd开源项目个人理解(23) - websocket客户端连接过程分析(Wireshark抓包+源码分析)

简介: 我的mqtt协议和emqttd开源项目个人理解(23) - websocket客户端连接过程分析(Wireshark抓包+源码分析)

我们可以使用emq自带的Dashboard插件,进行websocket调试,打开谷歌浏览器输入网址,其中192.168.83.128是emq所在的IP地址:


http://192.168.83.128:18083/#/websocket


用户名:admin,密码:public


WebSocket URI:ws(s)://192.168.83.128:8083/mqtt


TCP URI:tcp://192.168.83.128:1883



一、测试实践


测试环境:使用MQTT 3.1.1版本协议,Wireshark 2.4.1版本,使用wireshark抓包分析(如果是虚拟机,要抓VMnet8虚拟网卡)。


clientid:861694030142473

username:libaineu2004

password:12345678


cleanSession:true


keepalive:60s


1、测试方案1,使用普通的tcp连接,TCP URI:tcp://192.168.83.128:1883,wireshark过滤条件tcp.port == 1883。

(1)、MQTT Connect Command

鼠标点击wireshark的Frame页面,右键菜单,复制,选中树的所有可见项目

MQ Telemetry Transport Protocol, Connect Command

   Header Flags: 0x10 (Connect Command)

   Msg Len: 51

   Protocol Name Length: 4

   Protocol Name: MQTT

   Version: 4

   Connect Flags: 0xc2

   Keep Alive: 60

   Client ID Length: 15

   Client ID: 861694030142473

   User Name Length: 12

   User Name: libaineu2004

   Password Length: 8

   Password: 12345678


右键菜单,复制,字节为HEX + ASCII转储

0000   10 33 00 04 4d 51 54 54 04 c2 00 3c 00 0f 38 36  .3..MQTT...<..86

0010   31 36 39 34 30 33 30 31 34 32 34 37 33 00 0c 6c  1694030142473..l

0020   69 62 61 69 6e 65 75 32 30 30 34 00 08 31 32 33  ibaineu2004..123

0030   34 35 36 37 38                                   45678


(2)、MQTT Connect Ack

MQ Telemetry Transport Protocol, Connect Ack

   Header Flags: 0x20 (Connect Ack)

   Msg Len: 2

   Acknowledge Flags: 0x00

   Return Code: Connection Accepted (0)


0000   20 02 00 00                                       ...



2、测试方案2,使用websocket连接,WebSocket URI:ws(s)://192.168.83.128:8083/mqtt,wireshark过滤条件tcp.port == 8083。我们可以观察到mqtt连接的过程,websocket协议,完整的数据被拆分了好几帧传输。每一帧报文,都有WebSocket和Data字段。我们仅仅需要关注Data字段即可。我们发现,把所有Data字段组合拼接起来,就是和测试方案1完全相同的数据。MQTT Connect Command和Ack都完全相同。说明了即使使用了Websocket方式来传输,仍然遵循mqtt协议。


从本质上来讲,Websocket也是基于 TCP 协议的,同时借用了HTTP的协议来完成一部分握手。

主要解决 HTTP 协议中一个 request 对应一个 response 的尴尬。(http server 不能主动发送消息给 http client) 。通过 HTTP 完成 websocket 的握手过程,接着按照 websocket 协议进行通讯。


websocket 也有他自己的数据帧格式: http://blog.csdn.net/u010487568/article/details/20569027


(1)MQTT Connect Command


0000   10                                               .

0000   33                                               3

0000   00 04                                            ..

0000   4d 51 54 54                                      MQTT

0000   04                                               .

0000   c2                                               .

0000   00 3c                                            .<

0000   00 0f                                            ..

0000   38 36 31 36 39 34 30 33 30 31 34 32 34 37 33     861694030142473

0000   00 0c                                            ..

0000   6c 69 62 61 69 6e 65 75 32 30 30 34              libaineu2004

0000   00 08                                            ..

0000   31 32 33 34 35 36 37 38                          12345678


(2)MQTT Connect Ack


Data (4 bytes)

   Data: 20020000

   [Length: 4]

0000   20 02 00 00                                       ...



二、以下是源码分析,先来看emq v1.1.3版本的源码:


1、-module(emqttd_app).


%%--------------------------------------------------------------------

%% Start Servers

%%--------------------------------------------------------------------

start_servers(Sup) ->

   Servers = [

              {"emqttd wsclient supervisor", {supervisor, emqttd_ws_client_sup}},

%% Start http listener

start_listener({http, ListenOn, Opts}) ->

   mochiweb:start_http(http, ListenOn, Opts, {emqttd_http, handle_request, []});

打开8083端口,开启http服务器。Websocket也是基于 TCP 协议的,同时借用了HTTP的协议来完成一部分握手。



2、-module(emqttd_http).


%%--------------------------------------------------------------------

%% MQTT Over WebSocket

%%--------------------------------------------------------------------

handle_request('GET', "/mqtt", Req) ->

   lager:info("WebSocket Connection from: ~s", [Req:get(peer)]),

   Upgrade = Req:get_header_value("Upgrade"),

   Proto   = Req:get_header_value("Sec-WebSocket-Protocol"),

   case {is_websocket(Upgrade), Proto} of

       {true, "mqtt" ++ _Vsn} ->

           emqttd_ws:handle_request(Req);

       {false, _} ->

           lager:error("Not WebSocket: Upgrade = ~s", [Upgrade]),

           Req:respond({400, [], <<"Bad Request">>});

       {_, Proto} ->

           lager:error("WebSocket with error Protocol: ~s", [Proto]),

           Req:respond({400, [], <<"Bad WebSocket Protocol">>})

   end;

websocket的入口函数。重点关注emqttd_ws:handle_request(Req);



3、-module(emqttd_ws).

ws_loop(Data, State = #wsocket_state{peer = Peer, client_pid = ClientPid,
                                     parser_fun = ParserFun}, ReplyChannel) ->
    ?WSLOG(debug, Peer, "RECV ~p", [Data]),
    case catch ParserFun(iolist_to_binary(Data)) of
        {more, NewParser} ->
            State#wsocket_state{parser_fun = NewParser};
        {ok, Packet, Rest} ->
            gen_server:cast(ClientPid, {received, Packet}),
            ws_loop(Rest, reset_parser(State), ReplyChannel);
        {error, Error} ->
            ?WSLOG(error, Peer, "Frame error: ~p", [Error]),
            exit({shutdown, Error});
        {'EXIT', Reason} ->
            ?WSLOG(error, Peer, "Frame error: ~p", [Reason]),
            ?WSLOG(error, Peer, "Error data: ~p", [Data]),
            exit({shutdown, parser_error})
    end.


gen_server:cast(ClientPid, {received, Packet}), 接收来自客户端的消息。



4、-module(emqttd_ws_client).


handle_cast({received, Packet}, State = #wsclient_state{peer = Peer, proto_state = ProtoState}) ->
    case emqttd_protocol:received(Packet, ProtoState) of
        {ok, ProtoState1} ->
            noreply(State#wsclient_state{proto_state = ProtoState1});
        {error, Error} ->
            ?WSLOG(error, Peer, "Protocol error - ~p", [Error]),
            shutdown(Error, State);
        {error, Error, ProtoState1} ->
            shutdown(Error, State#wsclient_state{proto_state = ProtoState1});
        {stop, Reason, ProtoState1} ->
            stop(Reason, State#wsclient_state{proto_state = ProtoState1})
    end;

emqttd_protocol:received(Packet, ProtoState)。接收处理消息。



5、-module(emqttd_protocol).

process(Packet = ?CONNECT_PACKET(Var), State0) ->
    #mqtt_packet_connect{proto_ver  = ProtoVer,
                         proto_name = ProtoName,
                         username   = Username,
                         password   = Password,
                         clean_sess = CleanSess,
                         keep_alive = KeepAlive,
                         client_id  = ClientId} = Var,
    State1 = State0#proto_state{proto_ver  = ProtoVer,
                                proto_name = ProtoName,
                                username   = Username,
                                client_id  = ClientId,
                                clean_sess = CleanSess,
                                keepalive  = KeepAlive,
                                will_msg   = willmsg(Var),
                                connected_at = os:timestamp()},
    trace(recv, Packet, State1),
    {ReturnCode1, SessPresent, State3} =
    case validate_connect(Var, State1) of


validate_connect(Var, State1),校验clientid,username和password的有效性。


相关实践学习
消息队列RocketMQ版:基础消息收发功能体验
本实验场景介绍消息队列RocketMQ版的基础消息收发功能,涵盖实例创建、Topic、Group资源创建以及消息收发体验等基础功能模块。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列:消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验:基于MNS,0基础轻松构建 Web Client 本节课带您一起基于MNS,0基础轻松构建 Web Client
相关文章
|
2月前
|
消息中间件 监控 数据挖掘
基于RabbitMQ与Apache Flink构建实时分析系统
【8月更文第28天】本文将介绍如何利用RabbitMQ作为数据源,结合Apache Flink进行实时数据分析。我们将构建一个简单的实时分析系统,该系统能够接收来自不同来源的数据,对数据进行实时处理,并将结果输出到另一个队列或存储系统中。
108 2
|
2月前
|
消息中间件 存储 数据中心
RocketMQ的长轮询(Long Polling)实现分析
文章深入分析了RocketMQ的长轮询实现机制,长轮询结合了推送(push)和拉取(pull)两种消息消费模式的优点,通过客户端和服务端的配合,确保了消息的实时性同时将主动权保留在客户端。文中首先解释了长轮询的基本概念和实现步骤,然后通过一个简单的实例模拟了长轮询的过程,最后详细介绍了RocketMQ中DefaultMQPushConsumer的长轮询实现方式,包括PullMessage服务、PullMessageProcessor服务和PullCallback回调的工作原理。
59 1
|
2月前
|
物联网 C# 智能硬件
智能家居新篇章:WPF与物联网的智慧碰撞——通过MQTT协议连接与控制智能设备,打造现代科技生活的完美体验
【8月更文挑战第31天】物联网(IoT)技术的发展使智能家居设备成为现代家庭的一部分。通过物联网,家用电器和传感器可以互联互通,实现远程控制和状态监测等功能。本文将探讨如何在Windows Presentation Foundation(WPF)应用中集成物联网技术,通过具体示例代码展示其实现过程。文章首先介绍了MQTT协议及其在智能家居中的应用,并详细描述了使用Wi-Fi连接方式的原因。随后,通过安装Paho MQTT客户端库并创建MQTT客户端实例,演示了如何编写一个简单的WPF应用程序来控制智能灯泡。
50 0
|
2月前
|
消息中间件 Arthas Java
RocketMQ—一次连接namesvr失败的案例分析
项目组在使用RocketMQ时遇到Consumer连接Name Server失败的问题,异常显示连接特定地址失败。通过Arthas工具逐步分析代码执行路径,定位到创建Channel返回空值导致异常。进一步跟踪发现,问题源于Netty组件在初始化`ByteBufAllocator`时出现错误。分析依赖后确认存在Netty版本冲突。解决方法为排除冲突的Netty包,仅保留兼容版本。
144 0
RocketMQ—一次连接namesvr失败的案例分析
|
3月前
|
消息中间件 Java 物联网
消息队列 MQ操作报错合集之建立连接时发生了超时错误,该如何解决
消息队列(MQ)是一种用于异步通信和解耦的应用程序间消息传递的服务,广泛应用于分布式系统中。针对不同的MQ产品,如阿里云的RocketMQ、RabbitMQ等,它们在实现上述场景时可能会有不同的特性和优势,比如RocketMQ强调高吞吐量、低延迟和高可用性,适合大规模分布式系统;而RabbitMQ则以其灵活的路由规则和丰富的协议支持受到青睐。下面是一些常见的消息队列MQ产品的使用场景合集,这些场景涵盖了多种行业和业务需求。
消息队列 MQ操作报错合集之建立连接时发生了超时错误,该如何解决
|
2月前
|
关系型数据库 Serverless API
Serverless 函数实例问题之websocket连接如何解决
除了控制台,您还可以通过API或CLI使用PolarDB实例的Exec功能。API接口允许执行数据库命令,而阿里云CLI工具如`aliyun polardb`命令可用于管理数据库实例。详情参见[文档](https://help.aliyun.com/document_detail/137999.html)。
42 0
|
3月前
|
消息中间件 JavaScript Linux
消息队列 MQ操作报错合集之客户端在启动时遇到了连接错误,是什么原因
消息队列(MQ)是一种用于异步通信和解耦的应用程序间消息传递的服务,广泛应用于分布式系统中。针对不同的MQ产品,如阿里云的RocketMQ、RabbitMQ等,它们在实现上述场景时可能会有不同的特性和优势,比如RocketMQ强调高吞吐量、低延迟和高可用性,适合大规模分布式系统;而RabbitMQ则以其灵活的路由规则和丰富的协议支持受到青睐。下面是一些常见的消息队列MQ产品的使用场景合集,这些场景涵盖了多种行业和业务需求。
|
3月前
|
消息中间件 C语言 RocketMQ
消息队列 MQ操作报错合集之出现"Connection reset by peer"的错误,该如何处理
消息队列(MQ)是一种用于异步通信和解耦的应用程序间消息传递的服务,广泛应用于分布式系统中。针对不同的MQ产品,如阿里云的RocketMQ、RabbitMQ等,它们在实现上述场景时可能会有不同的特性和优势,比如RocketMQ强调高吞吐量、低延迟和高可用性,适合大规模分布式系统;而RabbitMQ则以其灵活的路由规则和丰富的协议支持受到青睐。下面是一些常见的消息队列MQ产品的使用场景合集,这些场景涵盖了多种行业和业务需求。
|
3月前
|
消息中间件 Java C语言
消息队列 MQ使用问题之在使用C++客户端和GBase的ESQL进行编译时出现core dump,该怎么办
消息队列(MQ)是一种用于异步通信和解耦的应用程序间消息传递的服务,广泛应用于分布式系统中。针对不同的MQ产品,如阿里云的RocketMQ、RabbitMQ等,它们在实现上述场景时可能会有不同的特性和优势,比如RocketMQ强调高吞吐量、低延迟和高可用性,适合大规模分布式系统;而RabbitMQ则以其灵活的路由规则和丰富的协议支持受到青睐。下面是一些常见的消息队列MQ产品的使用场景合集,这些场景涵盖了多种行业和业务需求。
|
13天前
|
消息中间件 运维 监控
云消息队列RabbitMQ实践解决方案评测报告
本报告旨在对《云消息队列RabbitMQ实践》解决方案进行综合评测。通过对该方案的原理理解、部署体验、设计验证以及实际应用价值等方面进行全面分析,为用户提供详尽的反馈与建议。
45 15
下一篇
无影云桌面