上一篇聊了代驾系统的整体架构和三级派单这篇往下钻一层司机端和乘客端挂在 Netty 长连接上手机进电梯、切后台、弱网闪断是家常便饭连接断了之后怎么发现、怎么重连、断线期间的消息怎么不丢这套机制才是派单链路真正能用起来的地基。全部代码来自我们开源的代驾系统仓库地址放在文末。一、协议设计两个枚举定生死长连接第一件事不是写代码是定协议。我们把客户端要干什么和服务端回了什么拆成两个枚举。客户端操作类型NettyHandleEnumsCONNECT(1)第一次连接或重连时的初始化KEEPALIVE(2)心跳保活COORDINATE_DRIVER(3)司机坐标上报服务端返回码NettyCodeEnums截取几个关键的KEEPALIVE(1,心跳正常),KEEPALIVE_ERROR(-1,心跳异常),CONNECT(2,第一次连接正常),CONNECT_ERROR(-2,第一次连接异常),DRIVER_UNFINISHED_ORDER 和 ORDER_MOVE 这两个码——它们存在的意义就是**重连恢复**后面会讲到。 ## 二、服务端 pipeline10秒空闲就踢Netty的 pipeline 配置在 DataAcceptInitializer 里 javaChannelPipelinepipelinesocketChannel.pipeline();pipeline.addLast(newHttpServerCodec());pipeline.addLast(newChunkedWriteHandler());pipeline.addLast(newHttpObjectAggregator(1024*64));//增加心跳支持/** * 针对客户端如果在1分钟时间内没有向服务端发送读写心跳ALL则主动断开连接 */pipeline.addLast(newIdleStateHandler(0,0,10,TimeUnit.SECONDS));pipeline.addLast(dataAcceptHandler);pipeline.addLast(newWebSocketServerProtocolHandler(/ws));IdleStateHandler(0, 0, 10)意思是读空闲不查、写空闲不查但读写加起来 10 秒没有任何动静就触发 ALL_IDLE 事件。客户端心跳间隔必须小于这个值实际 App 端是几秒一次。触发超时后的处理在userEventTriggeredpublic void userEventTriggered(ChannelHandlerContext ctx, Object evt) { if (evt instanceof IdleStateEvent) { IdleStateEvent event (IdleStateEvent) evt; if (event.state() IdleState.ALL_IDLE) { log.info(客户端(ALL_IDLE 总超时 重新连接)); } } ctx.channel().writeAndFlush(new MsgVo(NettyCodeEnums.SOCKET_TIME_OUT).toJsonStringbuf()); channelInactive(ctx);//清除netty通道 }先给客户端推一个SOCKET_TIME_OUT(101)万一它还活着知道自己被踢了、立刻重连然后进channelInactive清理通道。三、CONNECT重连不只是重连是状态恢复连接管理用的是个单机 HashMapkey 带角色前缀public class UserChanelRel { private static HashMapString, Channel manage new HashMap(); // key 前缀区分角色driver:{id} / user:{id} public static void put(String senderId, Channel channel){ manage.put(senderId, channel); } }CONNECT 懆支的核心逻辑DataAcceptHandlercase CONNECT: //客户端连接--第一次或者重连处理 if (chanelContext.equals(CONNECT)) { //每次连接或者重连时把连接存入连接池中 UserChanelRel.put(didStr, ctx.channel()); channel.writeAndFlush(new MsgVo(NettyCodeEnums.CONNECT).toJsonStringbuf()); if (cal.isEquals(channelIdentity, StaticUtils.STATUS_YES)){ //校验司机是否处于服务状态 OrderMain orderMain orderMainService.selectOrderByDriver(did); if (orderMain ! null) { //反馈司机有未完成订单 MsgVo msgVo new MsgVo(NettyCodeEnums.DRIVER_UNFINISHED_ORDER); msgVo.setOrderCode(orderMain.getOrderCode()); channel.writeAndFlush(msgVo.toJsonStringbuf()); } }这里的重点是重连成功不等于完事。司机端重连后服务端要查他名下有没有未完成订单有就推DRIVER_UNFINISHED_ORDERApp 收到后把订单界面恢复出来。用户端更细——如果重连时带着 orderCode且订单正在行驶中服务端会把里程、等待时长、费用等完整状态通过ORDER_MOVE推回去用户端直接接着断线前的界面渲染。这是踩过坑才明白的只恢复连接不恢复状态用户看到的就是一个回到首页的 App订单凭空消失投诉就是这么来的。四、KEEPALIVE心跳顺手干三件正事心跳包不只是保活我们的心跳分支顺手做了三件事caseKEEPALIVE://1. 心跳包超过350字节判定为非法直接断开if(length350){ctx.channel().writeAndFlush(newMsgVo(NettyCodeEnums.SOCKET_ERROR).toJsonStringbuf());channelInactive(ctx);UserChanelRel.delChannel(didStr);}//2. 没走过CONNECT的心跳请求移除Channelchannel1UserChanelRel.get(didStr);if(channel1null){ctx.writeAndFlush(newMsgVo(NettyCodeEnums.SOCKET_ERROR).toJsonStringbuf());channelInactive(ctx);}先做两道安全校验包太大350 字节的是恶意流量踢没经过 CONNECT 注册就发心跳的也是非法连接踢。然后分角色处理。司机端心跳里带着经纬度所以心跳 位置上报一举两得if(cal.isEquals(channelIdentity,StaticUtils.STATUS_YES)){//更新司机位置DDriverdriverdDriverService.getById(did);driver.setLonVal(requestVo.getLongitude());driver.setLatVal(requestVo.getLatitude());driver.setAddress(requestVo.getAddress());//异常状态的司机名下没单则恢复为可接单Integerstatedriver.getState();if(cal.isEquals(state,StaticUtils.STATUS_SUCCESS)||cal.isEquals(state,StaticUtils.STATUS_FAIL)){OrderMainorderMainorderMainService.selectOrderByDriver(did);if(orderMainnull){driver.setState(StaticUtils.STATUS_YES);}}dDriverService.updateById(driver);//回推最新司机状态让App刷新接单开关MsgVomsgVonewMsgVo(NettyCodeEnums.DRIVER_STATUS_RESET);msgVo.setDriverStatus(driver.getState());ctx.channel().writeAndFlush(msgVo.toJsonStringbuf());}用户端心跳则承担消息补偿——断线期间推不出去的消息暂存在 Redis心跳一上来就补发}else{StringkeyredisUtilsService.getKey(didStr);if(StringUtils.isNotEmpty(key)){MsgVomsgVoJSONObject.parseObjedt(key,MsgVo.class);ctx.channel().writeAndFlush(msgVo.toJsonStringbuf());redisUtilsService.delete(didStr);}}五、坦白几个不足这套东西能跑但放在更大规模下有几个明确的坑10 秒的 ALL_IDLE 太短。弱网环境下 TCP 重传都不止 10 秒容易误杀。注释里能看到最早的版本是 120 秒后来调小了这个值应该按客户端心跳间隔的 3 倍以上来配。UserChanelRel 是单机 HashMap。多实例部署时A 机器上的连接B 机器查不到派单推送会丢。正确做法是连接关系存 Redis推送走消息队列广播。channelInactive 里更新司机离线状态的代码被注释掉了。现在的逻辑是司机掉线后状态不立刻变等下次心跳或派单时才发现。这是个妥协立刻置离线会误伤闪断的司机不置又会给已掉线司机派单我们在派单侧做了二次校验来兜底。写在最后长连接这块的完整代码NettyServer、DataAcceptInitializer、DataAcceptHandler、UserChanelRel、协议枚举都在开源仓库里配上一篇的派单和计价主链路是闭环的。Gitee 地址https://gitee.com/zhoujian6666/biaoma-ride-car-service协议以仓库 LICENSE 为准。下一篇准备聊聊抢单池的并发处理有兴趣的可以先 Star。上一篇聊了代驾系统的整体架构和三级派单这篇往下钻一层司机端和乘客端挂在 Netty 长连接上手机进电梯、切后台、弱网闪断是家常便饭连接断了之后怎么发现、怎么重连、断线期间的消息怎么不丢这套机制才是派单链路真正能用起来的地基。全部代码来自我们开源的代驾系统仓库地址放在文末。## 一、协议设计两个枚举定生死长连接第一件事不是写代码是定协议。我们把客户端要干什么和服务端回了什么拆成两个枚举。客户端操作类型NettyHandleEnums- CONNECT(1)第一次连接或重连时的初始化- KEEPALIVE(2)心跳保活- COORDINATE_DRIVER(3)司机坐标上报服务端返回码NettyCodeEnums截取几个关键的javaKEEPALIVE(1,心跳正常),KEEPALIVE_ERROR(-1,心跳异常),CONNECT(2,第一次连接正常),CONNECT_ERROR(-2,第一次连接异常),DRIVER_UNFINISHED_ORDER(29,司机有未完成订单),DRIVER_STATUS_RESET(30,刷新司机状态),ORDER_MOVE(100,订单行驶中),SOCKET_TIME_OUT(101,读写超时),LOGIN_ERROR(997,账号登录失效)注意 DRIVER_UNFINISHED_ORDER 和 ORDER_MOVE 这两个码——它们存在的意义就是重连恢复后面会讲到。## 二、服务端 pipeline10 秒空闲就踢Netty 的 pipeline 配置在 DataAcceptInitializer 里javaChannelPipeline pipeline socketChannel.pipeline();pipeline.addLast(new HttpServerCodec());pipeline.addLast(new ChunkedWriteHandler());pipeline.addLast(new HttpObjectAggregator(1024*64));//增加心跳支持/** * 针对客户端如果在1分钟时间内没有向服务端发送读写心跳ALL则主动断开连接 */pipeline.addLast(new IdleStateHandler(0, 0, 10, TimeUnit.SECONDS));pipeline.addLast(dataAcceptHandler);pipeline.addLast(new WebSocketServerProtocolHandler(/ws));IdleStateHandler(0, 0, 10) 意思是读空闲不查、写空闲不查但读写加起来 10 秒没有任何动静就触发 ALL_IDLE 事件。客户端心跳间隔必须小于这个值实际 App 端是几秒一次。触发超时后的处理在 userEventTriggeredjavapublic void userEventTriggered(ChannelHandlerContext ctx, Object evt) { if (evt instanceof IdleStateEvent) { IdleStateEvent event (IdleStateEvent) evt; if (event.state() IdleState.ALL_IDLE) { log.info(客户端(ALL_IDLE 总超时 重新连接)); } } ctx.channel().writeAndFlush(new MsgVo(NettyCodeEnums.SOCKET_TIME_OUT).toJsonStringbuf()); channelInactive(ctx);//清除netty通道}先给客户端推一个 SOCKET_TIME_OUT(101)万一它还活着知道自己被踢了、立刻重连然后进 channelInactive 清理通道。
上一篇聊了代驾系统的整体架构和三级派单这篇往下钻一层司机端和乘客端挂在 Netty 长连接上手机进电梯、切后台、弱网闪断是家常便饭连接断了之后怎么发现、怎么重连、断线期间的消息怎么不丢这套机制才是派单链路真正能用起来的地基。全部代码来自我们开源的代驾系统仓库地址放在文末。一、协议设计两个枚举定生死长连接第一件事不是写代码是定协议。我们把客户端要干什么和服务端回了什么拆成两个枚举。客户端操作类型NettyHandleEnumsCONNECT(1)第一次连接或重连时的初始化KEEPALIVE(2)心跳保活COORDINATE_DRIVER(3)司机坐标上报服务端返回码NettyCodeEnums截取几个关键的KEEPALIVE(1,心跳正常),KEEPALIVE_ERROR(-1,心跳异常),CONNECT(2,第一次连接正常),CONNECT_ERROR(-2,第一次连接异常),DRIVER_UNFINISHED_ORDER 和 ORDER_MOVE 这两个码——它们存在的意义就是**重连恢复**后面会讲到。 ## 二、服务端 pipeline10秒空闲就踢Netty的 pipeline 配置在 DataAcceptInitializer 里 javaChannelPipelinepipelinesocketChannel.pipeline();pipeline.addLast(newHttpServerCodec());pipeline.addLast(newChunkedWriteHandler());pipeline.addLast(newHttpObjectAggregator(1024*64));//增加心跳支持/** * 针对客户端如果在1分钟时间内没有向服务端发送读写心跳ALL则主动断开连接 */pipeline.addLast(newIdleStateHandler(0,0,10,TimeUnit.SECONDS));pipeline.addLast(dataAcceptHandler);pipeline.addLast(newWebSocketServerProtocolHandler(/ws));IdleStateHandler(0, 0, 10)意思是读空闲不查、写空闲不查但读写加起来 10 秒没有任何动静就触发 ALL_IDLE 事件。客户端心跳间隔必须小于这个值实际 App 端是几秒一次。触发超时后的处理在userEventTriggeredpublic void userEventTriggered(ChannelHandlerContext ctx, Object evt) { if (evt instanceof IdleStateEvent) { IdleStateEvent event (IdleStateEvent) evt; if (event.state() IdleState.ALL_IDLE) { log.info(客户端(ALL_IDLE 总超时 重新连接)); } } ctx.channel().writeAndFlush(new MsgVo(NettyCodeEnums.SOCKET_TIME_OUT).toJsonStringbuf()); channelInactive(ctx);//清除netty通道 }先给客户端推一个SOCKET_TIME_OUT(101)万一它还活着知道自己被踢了、立刻重连然后进channelInactive清理通道。三、CONNECT重连不只是重连是状态恢复连接管理用的是个单机 HashMapkey 带角色前缀public class UserChanelRel { private static HashMapString, Channel manage new HashMap(); // key 前缀区分角色driver:{id} / user:{id} public static void put(String senderId, Channel channel){ manage.put(senderId, channel); } }CONNECT 懆支的核心逻辑DataAcceptHandlercase CONNECT: //客户端连接--第一次或者重连处理 if (chanelContext.equals(CONNECT)) { //每次连接或者重连时把连接存入连接池中 UserChanelRel.put(didStr, ctx.channel()); channel.writeAndFlush(new MsgVo(NettyCodeEnums.CONNECT).toJsonStringbuf()); if (cal.isEquals(channelIdentity, StaticUtils.STATUS_YES)){ //校验司机是否处于服务状态 OrderMain orderMain orderMainService.selectOrderByDriver(did); if (orderMain ! null) { //反馈司机有未完成订单 MsgVo msgVo new MsgVo(NettyCodeEnums.DRIVER_UNFINISHED_ORDER); msgVo.setOrderCode(orderMain.getOrderCode()); channel.writeAndFlush(msgVo.toJsonStringbuf()); } }这里的重点是重连成功不等于完事。司机端重连后服务端要查他名下有没有未完成订单有就推DRIVER_UNFINISHED_ORDERApp 收到后把订单界面恢复出来。用户端更细——如果重连时带着 orderCode且订单正在行驶中服务端会把里程、等待时长、费用等完整状态通过ORDER_MOVE推回去用户端直接接着断线前的界面渲染。这是踩过坑才明白的只恢复连接不恢复状态用户看到的就是一个回到首页的 App订单凭空消失投诉就是这么来的。四、KEEPALIVE心跳顺手干三件正事心跳包不只是保活我们的心跳分支顺手做了三件事caseKEEPALIVE://1. 心跳包超过350字节判定为非法直接断开if(length350){ctx.channel().writeAndFlush(newMsgVo(NettyCodeEnums.SOCKET_ERROR).toJsonStringbuf());channelInactive(ctx);UserChanelRel.delChannel(didStr);}//2. 没走过CONNECT的心跳请求移除Channelchannel1UserChanelRel.get(didStr);if(channel1null){ctx.writeAndFlush(newMsgVo(NettyCodeEnums.SOCKET_ERROR).toJsonStringbuf());channelInactive(ctx);}先做两道安全校验包太大350 字节的是恶意流量踢没经过 CONNECT 注册就发心跳的也是非法连接踢。然后分角色处理。司机端心跳里带着经纬度所以心跳 位置上报一举两得if(cal.isEquals(channelIdentity,StaticUtils.STATUS_YES)){//更新司机位置DDriverdriverdDriverService.getById(did);driver.setLonVal(requestVo.getLongitude());driver.setLatVal(requestVo.getLatitude());driver.setAddress(requestVo.getAddress());//异常状态的司机名下没单则恢复为可接单Integerstatedriver.getState();if(cal.isEquals(state,StaticUtils.STATUS_SUCCESS)||cal.isEquals(state,StaticUtils.STATUS_FAIL)){OrderMainorderMainorderMainService.selectOrderByDriver(did);if(orderMainnull){driver.setState(StaticUtils.STATUS_YES);}}dDriverService.updateById(driver);//回推最新司机状态让App刷新接单开关MsgVomsgVonewMsgVo(NettyCodeEnums.DRIVER_STATUS_RESET);msgVo.setDriverStatus(driver.getState());ctx.channel().writeAndFlush(msgVo.toJsonStringbuf());}用户端心跳则承担消息补偿——断线期间推不出去的消息暂存在 Redis心跳一上来就补发}else{StringkeyredisUtilsService.getKey(didStr);if(StringUtils.isNotEmpty(key)){MsgVomsgVoJSONObject.parseObjedt(key,MsgVo.class);ctx.channel().writeAndFlush(msgVo.toJsonStringbuf());redisUtilsService.delete(didStr);}}五、坦白几个不足这套东西能跑但放在更大规模下有几个明确的坑10 秒的 ALL_IDLE 太短。弱网环境下 TCP 重传都不止 10 秒容易误杀。注释里能看到最早的版本是 120 秒后来调小了这个值应该按客户端心跳间隔的 3 倍以上来配。UserChanelRel 是单机 HashMap。多实例部署时A 机器上的连接B 机器查不到派单推送会丢。正确做法是连接关系存 Redis推送走消息队列广播。channelInactive 里更新司机离线状态的代码被注释掉了。现在的逻辑是司机掉线后状态不立刻变等下次心跳或派单时才发现。这是个妥协立刻置离线会误伤闪断的司机不置又会给已掉线司机派单我们在派单侧做了二次校验来兜底。写在最后长连接这块的完整代码NettyServer、DataAcceptInitializer、DataAcceptHandler、UserChanelRel、协议枚举都在开源仓库里配上一篇的派单和计价主链路是闭环的。Gitee 地址https://gitee.com/zhoujian6666/biaoma-ride-car-service协议以仓库 LICENSE 为准。下一篇准备聊聊抢单池的并发处理有兴趣的可以先 Star。上一篇聊了代驾系统的整体架构和三级派单这篇往下钻一层司机端和乘客端挂在 Netty 长连接上手机进电梯、切后台、弱网闪断是家常便饭连接断了之后怎么发现、怎么重连、断线期间的消息怎么不丢这套机制才是派单链路真正能用起来的地基。全部代码来自我们开源的代驾系统仓库地址放在文末。## 一、协议设计两个枚举定生死长连接第一件事不是写代码是定协议。我们把客户端要干什么和服务端回了什么拆成两个枚举。客户端操作类型NettyHandleEnums- CONNECT(1)第一次连接或重连时的初始化- KEEPALIVE(2)心跳保活- COORDINATE_DRIVER(3)司机坐标上报服务端返回码NettyCodeEnums截取几个关键的javaKEEPALIVE(1,心跳正常),KEEPALIVE_ERROR(-1,心跳异常),CONNECT(2,第一次连接正常),CONNECT_ERROR(-2,第一次连接异常),DRIVER_UNFINISHED_ORDER(29,司机有未完成订单),DRIVER_STATUS_RESET(30,刷新司机状态),ORDER_MOVE(100,订单行驶中),SOCKET_TIME_OUT(101,读写超时),LOGIN_ERROR(997,账号登录失效)注意 DRIVER_UNFINISHED_ORDER 和 ORDER_MOVE 这两个码——它们存在的意义就是重连恢复后面会讲到。## 二、服务端 pipeline10 秒空闲就踢Netty 的 pipeline 配置在 DataAcceptInitializer 里javaChannelPipeline pipeline socketChannel.pipeline();pipeline.addLast(new HttpServerCodec());pipeline.addLast(new ChunkedWriteHandler());pipeline.addLast(new HttpObjectAggregator(1024*64));//增加心跳支持/** * 针对客户端如果在1分钟时间内没有向服务端发送读写心跳ALL则主动断开连接 */pipeline.addLast(new IdleStateHandler(0, 0, 10, TimeUnit.SECONDS));pipeline.addLast(dataAcceptHandler);pipeline.addLast(new WebSocketServerProtocolHandler(/ws));IdleStateHandler(0, 0, 10) 意思是读空闲不查、写空闲不查但读写加起来 10 秒没有任何动静就触发 ALL_IDLE 事件。客户端心跳间隔必须小于这个值实际 App 端是几秒一次。触发超时后的处理在 userEventTriggeredjavapublic void userEventTriggered(ChannelHandlerContext ctx, Object evt) { if (evt instanceof IdleStateEvent) { IdleStateEvent event (IdleStateEvent) evt; if (event.state() IdleState.ALL_IDLE) { log.info(客户端(ALL_IDLE 总超时 重新连接)); } } ctx.channel().writeAndFlush(new MsgVo(NettyCodeEnums.SOCKET_TIME_OUT).toJsonStringbuf()); channelInactive(ctx);//清除netty通道}先给客户端推一个 SOCKET_TIME_OUT(101)万一它还活着知道自己被踢了、立刻重连然后进 channelInactive 清理通道。
分析器开发实战：从文法推导到Python实现)
