Netty学习之旅------高仿Dubbo服务调用模型、私有协议实现、编码解码器使用实践
2017-03-09 11:16
731 查看
1、本文实现如下功能:
1、客户端与服务端基于单一长连接进行通信,客户端一条连接被多个线程使用。
2、实现私有协议
请求协议
协议头:4字节的请求序列号 2字节的请求类型 2字节数据包长度
数据部分:
服务调用:2字节请求服务名长度 若干字节请求服务名 2字节请求参数长度 若干字节参数
心跳包:数据部分无
响应信息
协议头:4字节的请求序列号 2字节的执行状态码 2字节数据长度
数据部分: 多字节的响应数据信息(data)
3、代码基于Netty5
几个核心类代码如下:
完整代码如下:http://download.csdn.net/detail/prestigeding/9775190
其中Server是服务器启动类,Client 是客户端启动类。
1、客户端与服务端基于单一长连接进行通信,客户端一条连接被多个线程使用。
2、实现私有协议
请求协议
协议头:4字节的请求序列号 2字节的请求类型 2字节数据包长度
数据部分:
服务调用:2字节请求服务名长度 若干字节请求服务名 2字节请求参数长度 若干字节参数
心跳包:数据部分无
响应信息
协议头:4字节的请求序列号 2字节的执行状态码 2字节数据长度
数据部分: 多字节的响应数据信息(data)
3、代码基于Netty5
几个核心类代码如下:
package persistent.prestige.demo.netty.dubbo.net; import java.util.List; import io.netty.buffer.ByteBuf; import io.netty.channel.ChannelHandlerContext; import io.netty.handler.codec.ByteToMessageDecoder; import persistent.prestige.demo.netty.dubbo.utils.LogUtils; /** * 请求消息 解码器,实现自定义协议,,,用于服务端 * * 自定义协议如下 4字节的请求序列号 2字节的请求类型 2字节数据长度 2字节服务调用长度 N字节的服务调用名 2字节请求参数长度 请求参数 * 其中请求类型为1:服务调用2:心跳包,如果是心跳包的话,只有协议头,也就是【4字节请求序列号,2字节请求类型,2字节数据长度】 * * @author prestigeding@126.com * */ public class RequestFrameDecoder extends ByteToMessageDecoder { /** 协议头 长度 */ private static final int HEAD_LENGTH = 8; /* * (non-Javadoc) * * @param ctx Handler执行环境 * * @param in //当前累积缓存区字节 * * @param out //解码后的消息 */ @Override protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception { // TODO Auto-generated method stub LogUtils.log("RequestFrameDecoder start calling...."); int r = in.readableBytes(); LogUtils.log("RequestFrameDecoder 累计缓存去大小:%d", r); if (r < HEAD_LENGTH) { LogUtils.log("RequestFrameDecoder 当前累积缓存区可读数据长度为%d,小于协议头的长度%d", r, HEAD_LENGTH); return; } int currentReaderIndex = in.readerIndex(); int requestId = in.getInt(currentReaderIndex);// 0-3 short requestType = in.getShort(currentReaderIndex + 4);// 4-5 short contentLength = in.getShort(currentReaderIndex + 6);// 6-7 头部信息共8个字节 LogUtils.log("RequestFrameDecoder requestId:%d,requestType:%d,contentLength:%d", requestId, requestType, contentLength); Request request = null; if (requestType == 1) { if (r < (HEAD_LENGTH + contentLength)) { // 如果累积缓存区可读容量小于一个请求包的长度,直接返回 LogUtils.log("RequestFrameDecoder 当前累积缓存区可读数据长度为%d,小于数据报的长度%d", r, HEAD_LENGTH + contentLength); return; } // 否则,直接读取数据 in.skipBytes(HEAD_LENGTH);// 先跳过协议头信息 int slen = in.readShort(); byte[] serviceData = new byte[slen]; in.readBytes(serviceData, 0, slen); short plen = in.readShort(); byte[] paramsData = new byte[plen]; in.readBytes(paramsData, 0, plen); out.add(request = new Request(requestId, requestType, new String(serviceData), new String(paramsData))); request.setContentLength(contentLength); //这里,其实上层应用不需要知道请求包的长度,这里属于多于的操作 LogUtils.log("RequestFrameDecoder 成功从服务端响应中解析出业务包,requestId:%d", requestId); } else if (requestType == 2) { // 此处不对请求类型进行校验 in.skipBytes(HEAD_LENGTH); out.add(request = new Request(requestId, requestType)); request.setContentLength(contentLength); //这里,其实上层应用不需要知道请求包的长度,这里属于多于的操作 LogUtils.log("RequestFrameDecoder 成功从服务端响应中解析出心跳包,requestId:%d", requestId); } } } package persistent.prestige.demo.netty.dubbo.net; import io.netty.buffer.ByteBuf; import io.netty.channel.ChannelHandlerContext; import io.netty.handler.codec.MessageToByteEncoder; import persistent.prestige.demo.netty.dubbo.utils.LogUtils; /** * 请求消息编码器,实现自定义协议,用于客户端 * * 自定义协议如下 * 4字节的请求序列号 2字节的请求类型 2字节数据长度 2字节服务调用长度 N字节的服务调用名 2字节请求参数长度 请求参数 * 其中请求类型为1:服务调用2:心跳包,如果是心跳包的话,只有协议头,也就是【4字节请求序列号,2字节请求类型,2字节数据长度】 * * @author prestigeding@126.com * */ public class RequestFrameEncoder extends MessageToByteEncoder<Request> { @Override protected void encode(ChannelHandlerContext ctx, Request msg, ByteBuf out) throws Exception { LogUtils.log("RequestFrameEncoder begin calling...requestId:%d,requestType:%d.",msg.getRequestId(), msg.getRequestType()); out.writeInt(msg.getRequestId()); // requestId out.writeShort(msg.getRequestType()); // requestType if(msg.getRequestType() == 1 ) { //服务调用 byte[] serviceData = msg.getService().getBytes(); byte[] paramsData = msg.getParams().getBytes(); int slen, plen; out.writeShort(( 2 + (slen = serviceData.length) + 2 + (plen = paramsData.length))); out.writeShort(slen); out.writeBytes(serviceData); out.writeShort(plen); out.writeBytes(paramsData); } else if(msg.getRequestType() == 2 ) { //心跳包 out.writeShort(0); } LogUtils.log("RequestFrameEncoder end call...."); } } package persistent.prestige.demo.netty.dubbo.net; import java.util.List; import io.netty.buffer.ByteBuf; import io.netty.channel.ChannelHandlerContext; import io.netty.handler.codec.ByteToMessageDecoder; import persistent.prestige.demo.netty.dubbo.utils.LogUtils; /** * 响应信息解码器 用于客户端 * @author dingwei2 * * 响应报文格式 * 4字节的请求序列号 2字节的执行状态码 2字节数据长度 多字节的响应数据信息(data) * */ public class ResponseFrameDecoder extends ByteToMessageDecoder { private static final int HEAD_LENGTH = 8; @Override protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception { LogUtils.log("ResponseFrameDecoder start calling...."); // TODO Auto-generated method stub int r = in.readableBytes(); LogUtils.log("ResponseFrameDecoder 累积缓存区中可读长度:%d", r); if(r < HEAD_LENGTH ) { LogUtils.log("ResponseFrameDecoder 当前累积缓存区可读数据长度为%d,小于协议头的长度%d", r, HEAD_LENGTH); return; } int currentReaderIndex = in.readerIndex(); int requestId = in.getInt(currentReaderIndex); // 0-3 4个字节 short code = in.getShort(currentReaderIndex + 4); // 4,5 short contentLength = in.getShort(currentReaderIndex + 6); //响应数据长度 LogUtils.log("ResponseFrameDecoder requestId:%d,code:%d,contentLength:%d", requestId, code, contentLength); int packetLen; if( r < (packetLen = HEAD_LENGTH + contentLength)) { LogUtils.log("ResponseFrameDecoder 可读长度:%d,需要包长度:%d", r, packetLen); return; } in.skipBytes( HEAD_LENGTH ); String data; if(contentLength > 0 ) { byte[] dst = new byte[contentLength]; in.readBytes(dst, 0, contentLength); data = new String(dst); } else { data = ""; } out.add(new Response(requestId, code, new String(data))); LogUtils.log("ResponseFrameDecoder 成功解析服务端响应信息并准备给下游处理器处理"); } } package persistent.prestige.demo.netty.dubbo.net; import io.netty.buffer.ByteBuf; import io.netty.channel.ChannelHandlerContext; import io.netty.handler.codec.MessageToByteEncoder; import persistent.prestige.demo.netty.dubbo.utils.LogUtils; /** * 响应信息 编码器,,用于服务器 * @author dingwei2 * * 响应报文格式 * 4字节的请求序列号 2字节的执行状态码 2字节数据长度 多字节的响应数据信息(data) * */ public class ResponseFrameEncoder extends MessageToByteEncoder<Response> { @Override protected void encode(ChannelHandlerContext ctx, Response msg, ByteBuf out) throws Exception { // TODO Auto-generated method stub LogUtils.log("ResponseFrameEncoder:服务端编码响应,requestId:%d,code:%d,data:%s", msg.getRequestId(), msg.getCode(), msg.getData()); out.writeInt(msg.getRequestId()); out.writeShort(msg.getCode()); byte[] data = msg.getData().getBytes(); out.writeShort(data.length); out.writeBytes(data); LogUtils.log("ResponseFrameEncoder:服务端编码响应成功"); } } package persistent.prestige.demo.netty.dubbo.net; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import io.netty.bootstrap.Bootstrap; import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelHandlerAdapter; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelInitializer; import io.netty.channel.nio.NioEventLoopGroup; import io.netty.channel.socket.SocketChannel; import io.netty.channel.socket.nio.NioSocketChannel; import persistent.prestige.demo.netty.dubbo.net.heart.HeartExecutorThreadPool; import persistent.prestige.demo.netty.dubbo.net.heart.HeartTask; import persistent.prestige.demo.netty.dubbo.utils.LogUtils; /** * netty 客户端 connection 一个NettyConnection 就是一条长连接 * @author prestigeding@126.com * */ public class NettyClientConnection { private String serverIp; private int serverPort; private boolean connect = false; private Bootstrap bootstrap; private NioEventLoopGroup workGroup; private ChannelFuture channelFuture; //客户端连接请求 private static final ConcurrentHashMap<Integer, ResponseFuture> requestMap = new ConcurrentHashMap<Integer, ResponseFuture>(); private static final AtomicInteger seq = new AtomicInteger(0); private HeartExecutorThreadPool heartExecutor = null; /** * * @param serverIp * @param serverPort */ public NettyClientConnection(String serverIp, int serverPort) { this.serverIp = serverIp; this.serverPort = serverPort; workGroup = new NioEventLoopGroup(); bootstrap = new Bootstrap(); bootstrap.group(workGroup) .channel(NioSocketChannel.class) .handler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel ch) throws Exception { // TODO Auto-generated method stub ch.pipeline().addLast(new ResponseFrameDecoder()) .addLast(new RequestFrameEncoder()) .addLast(new DispatchHandler()); } }); } private void connect() { if(!connect) { synchronized (this) { try { if(!connect) { channelFuture = bootstrap.connect(serverIp, serverPort).sync(); heartExecutor = new HeartExecutorThreadPool(0, this, 3); connect = true; heartExecutor.scheduleAtFixedRate(new HeartTask(this), 1, 3, TimeUnit.SECONDS);//单位为秒 } } catch (InterruptedException e) { // TODO Auto-generated catch block e.printStackTrace(); throw new NullPointerException("连接服务端失败"); } } } } /** * 发送服务调用 * @param service * @param params * @return */ public ResponseFuture sendMsg(String service, String params) { Integer requestId = seq.incrementAndGet(); System.out.println("客户端requestId:" + requestId); return send(new Request(requestId, (short)1, service, params)); } /** * 发送心跳包 * @return */ public ResponseFuture sendHeatMsg() { Integer requestId = seq.incrementAndGet(); return send(new Request(requestId, (short)2)); } private ResponseFuture send(Request request) { connect(); ResponseFuture responseFuture = new ResponseFuture(); requestMap.putIfAbsent(request.getRequestId(), responseFuture); channelFuture.channel().writeAndFlush(request); return responseFuture; } /** * 关闭连接 */ public void close() { try { channelFuture.channel().close().sync(); } catch (InterruptedException e) { // TODO Auto-generated catch block e.printStackTrace(); } finally { workGroup.shutdownGracefully(); if(heartExecutor != null) { heartExecutor.shutdownNow(); } } } /** * 客户端命令转发器 * @author prestigeding@126.com * */ private class DispatchHandler extends ChannelHandlerAdapter { /** * 得到服务端的响应后,放入结果中,供客户端使用结果 */ @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { LogUtils.log("客户端收到服务端响应,并开始处理数据"); Response response = (Response)msg; LogUtils.log("服务端响应requestId:", response.getRequestId()); ResponseFuture f = requestMap.remove(response.getRequestId()); if(f != null) { f.setResult(response); } else { LogUtils.log("警告:客户端丢失请求信息"); } } } }
完整代码如下:http://download.csdn.net/detail/prestigeding/9775190
其中Server是服务器启动类,Client 是客户端启动类。
相关文章推荐
- 使用dubbo+zookeper实现服务调用与使用webservice对比
- 使用Zipkin和Brave 实现dubbo服务调用跟踪
- 使用接口的方式调用远程服务 ------ 利用动态调用服务,实现.net下类似Dubbo的玩法。
- 学习笔记:使用Web Service Software Factory开发简易留言本服务以及Mobile调用实现-1.创建Service
- 使用Zipkin和Brave 实现dubbo服务调用跟踪
- 使用Django REST Framework来快速实现API调用服务——上篇(运行环境和模型层)
- 使用rmi协议,基于cglib实现组件管理和远程方法调用
- 在 PHP 中使用 SOAP 协议调用 Web服务(WebService)
- Dubbo学习过程、使用经验分享及实现原理简单介绍
- 计算机网络(自顶向下方法)学习笔记 1.5 协议层次和它们的服务模型
- 学习实践:使用模式,原则实现一个C++自动化测试程序
- 一步步学习SPD2010--第七章节--使用BCS业务连接服务(7)--导出和使用BDC模型
- 学习实践:使用模式,原则实现一个C++自动化测试程序
- [转载] Dubbo实现RPC调用使用入门
- 基于Dubbo的Hessian协议实现远程调用
- 学习实践:使用模式,原则实现一个C++数据库访问类
- 使用SQL Server 集成服务中结合SharePoint的客户端模型进行SharePoint列表的同步实践!
- dubbo学习过程、使用经验分享及实现原理简单介绍