京东到家基于Netty与WebSocket的实践

Posted 2021-04-25 51CTO技术栈

“

在京东到家商家中心系统中，商家提出了要在 Web 端实现自动打印的需求，不再需要人工盯守点击打印，直接打印小票，以节约人工成本。

解决思路

关于问题的两种思考逻辑：

• 可以用 ajax 来轮询服务端获取最新订单，也就是 pull。

• 可以用类似推送的设计来实现，也就是 push。

我们评估了两种思路的优缺点：

• ajax 方式实现简单，只需要定时从服务端 pull 数据即可，但也增加了很多次无效的轮询，即无形中增加服务端无效查询。

• push 方式实现稍复杂，需要服务端与 PC 端保持连接，这就需要建立长连接，最终通过长连接的方式来实现 push 效果。
  

经过讨论，我们选择了第二种，订单中心生产出的新订单，通过 MQ 的方式推送给 Web 端，最终获得一个比较好的用户体验。

方案介绍

关于长连接方案的选择，我们参考了不少帖子，最终选择了使用 websocket 协议来实现长连接，类似场景如 IM，服务端即时推送等都使用了这个协议。

接下来我们比较一下 websocket 的框架，比较主流的有 netty、tomcat、socketIO 三个框架：

• 基于支持 websocket 的容器，开发简单，例如 tomcat，但在高并发的支持不是很好，连接的时候容易连接断开，还有就是依赖容器。
  

• netty-socketIO 是在 netty4 基础之上做了一层封装，效率如同 netty 一样，是一个全平台方案，友好的 API。京东的 logbook 也是用了 socketIO 来传递日志，也是我们的一个备选方案。
  

• netty 是业内主流的 NIO 框架，netty 对 Java NIO 做了封装，让开发者更多关注业务，降低开发成本。

  很多著名的 RPC 框架都采用了 netty 作为传输层，友好的 API，功能强大，内置了很多编解码协议，实现 websocket 协议也是十分方便。
  

那我们横向比较一下这些框架：

所以在选型方面我们还是定位在 socketIO 与 netty 上面，在兼顾扩展性与灵活性的同时，我们也考虑到 netty 可以提供 http 的功能。

最终我们选择了使用 netty，当然 socketIO 封装了很多功能，也是十分强大，相比较来说 netty 更适合我们，比较轻量。

Netty 的特性

netty 具有异步非阻塞的特性，传统 IO 是面向流的，NIO 是面向缓冲区的，这也是它的非阻塞原因所在。

netty 的线程模型如图所示：

这种模型就是我们常说的 Reactor 模型，boss 线程其实是一个独立的 NIO 线程池，用于接收 client 请求，默认线程池大小为 1，worker 线程池用于处理具体的读写操作，默认线程池大小为 2*cpu 个数。

在上述模型中要特别注意 ExecutionHandler，ExecutionHandler 是运行在 worker 线程中的，所以耗时的操作最好在线程池中运行， 比如 IO 或者计算，不然会影响整个 netty 的吞吐。

了解了这些，我们根据自己的业务设计出流程，如下图所示：

• 步骤 1：Web 端请求服务端进行注册，注册成功保持长连接。

• 步骤 2：服务端发送 MQ。

• 步骤 3：netty 将收到的消息推送给 Web 端。
  

• 步骤 4：Web 端调用打印控件进行打印，打印控件需提前安装好（打印控件是 PC 上安装的一个驱动程序，用过 JS 方式来调用）。
  

如果调用 JS 成功，控件将把打印信息放入打印队列，如果不成功，重复步骤 4。

当然现在的结构只是单机版，不满足生产条件，那将来的结构可能会演变成如下图所示：

我们会在服务端与 netty 之间建立路由层，路由层的主要职责有：

• 收集集群存活信息

• 记录落点，就是落在哪一台机器上面

• 接收消息与分发消息

有了这三种能力，我们就可以轻松的指定信息分发策略。我们希望使用 http 协议来路由，这就需要 netty 有 http 短连接接收的能力 ，所以 netty 整体上需要长短连接两种能力。

下面是部分代码：

netty 启动类，我们通过 spring 来启动 netty，因为 netty 启动会阻塞主线程，所以需要在子线程中来启动 netty，下面是启动参数。

接着来写我们的 ChannelInitializer，HttpServerCodec 为编解码器，WSServerProtocolHandler 为 websocket 协议握手。

我们更关注业务层面自定义的两个 hander，httpRequestHandler，authorizeHandler。

httpRequestHandler 的作用是处理 URL 是否合法，接收参数。

httpRequestHandler 此方法中也可以根据 URL 来过滤，自定义自己的短连接请求。

authorizeHandler 的作用是校验数据是否正确，如果正确会将 channel 保存到 map 中，通过 map 建立起业务 ID 与通道之间的关系。

校验的过程我们在 authorizeHandler 中的 channelRead 展开，如果未通过，直接关闭当前 channel。

如果通过校验，则通过 ctx.fireChannelRead(msg)；方法将信息传入下一个 handler 去处理。

在项目里主要是以传递参数来进行数据校验的，也就是通过 URL 传参来实现。

在 httpRequestHandler 中我们将 URL 参数 set 到 channel 的 attr 中，并传递给了下一个 handler，也就是 authorizeHandler。

所以在 authorize 方法中我们可以利用 get() 方法得到参数值，u 是经过加密的数据，我们需要在这里进行解密，解密失败，可认为校验失败。

当然如果有跨应用的服务，也可以通过 Cookie 的方式来进行加密串的读写，通过 request.getHeader 是可以获取 Cookie 中的信息，这就看具体业务了。

示例代码如下：

这个 map 可以理解为 servlet 中的 session，当有信息需要传送给某个客户端时，我们调用 map.get(key) 方式到当前该客户端的 channel，调用 writeAndFlush 方法将信息发送出去，下面举例通过接收 MQ 消息后的处理逻辑。

接下来有人可能想到，如果通道关闭了怎么办？map 中的 channel 是不是就失效了呢？

其实我们还需要有一个类似心跳的机制去维护 channel，间接的去维护这个 map。

如果是通道正常关闭，可以通过 channelInactive 方法来监听。

如果是长时间空闲，在项目中我们使用了增加的 IdleStateHandler 来处理，通过覆盖 userEventTriggered 方法来监听空闲 channel，当某个 channel 到达我们设置的超时时间时，netty 会回调此方法。

至此，核心部分已经处理完成，剩下的就是通过保存的 channel 来发送信息给客户端了。

最后在 Web 端，我们采用了 reconnecting-websocket，它是一个小型的 javascript 库，封装了 WebSocket API， 提供了在连接断开时自动重连的机制，能够帮助我们完成断开重连的操作。

遇到的问题

经过测试，在 ws 的 uri 后面不能传递参数，不然在 netty 实现 websocket 协议握手的时候会出现断开连接的情况。

针对这种情况在 websocketHandler 之前做了一层 httpHander 过滤，将传递参数放入 channel 的 attr 中，然后重写 request 的 uri，并传入下一个管道中，基本上解决了这个问题。

在读写空闲的时候尽量以发心跳包的方式维护连接，但在客户端由于网络不稳定或者是服务端重启，连接会断开，瞬间有可能接收不到订单消息，为此在客户端需要实现断开重连机制。

此问题我们采用 reconnecting-websocket的 JS 框架，此框架扩展了原生 websocket 的实现，做了断开重连机制，有效的防止断开后不能及时连接。

在测试过程中由于控件与小票机的问题，可能会出现打印异常或者小票机没纸的情况。Lodop 控件可以将打印信息放入电脑的打印队列。

如果没纸了，小票机会报警，再次放入小票纸，打印机会自动打印队列中的数据。

出现调用控件异常偶尔发生，现在处理办法是在 JS 中进行了的 try catch。

如果失败，进行重试，重试次数自定义，超过重试次数暂不做处理，此处还不太严谨，需要再进行优化。

总结

通过上面的实践，我们基本已经实现了 Web 端的自动打印，经过长时间的内部测试，服务端与客户端通信稳定，我们将灰度商家做用户体验。

在特定的场景下，选择适当的技术会提高我们的效率，否则会适得其反。

选择长连接，大家可以把握这三个大原则：

• 服务端是否需要主动推送数据到客户端以实现控制的效果。

• 对于实时性的要求是否苛刻。

• 对于客户端是否需要关注它在线状态的实时变化。

介绍：软件开发工程师，任职于达达京东到家后台研发团队，负责订单流程的开发工作。

编辑：陶家龙、孙淑娟

精彩文章推荐：