資訊專欄INFORMATION COLUMN
摘要：是一個分布式服務框架，以及治理方案。手寫注意要點手寫注意要點基于上文中對于協(xié)議的理解，如果我們自己去實現(xiàn)，需要考慮哪些技術呢其實基于圖的整個流程應該有一個大概的理解?；谑謱憣崿F(xiàn)基于手寫實現(xiàn)理解了協(xié)議后，我們基于來實現(xiàn)一個通信框架。
閱讀這篇文章之前，建議先閱讀和這篇文章關聯(lián)的內(nèi)容。
[1]詳細剖析分布式微服務架構下網(wǎng)絡通信的底層實現(xiàn)原理（圖解）
[2][年薪60W的技巧]工作了5年，你真的理解Netty以及為什么要用嗎？（深度干貨）
[4]BAT面試必問細節(jié)：關于Netty中的ByteBuf詳解
[5]通過大量實戰(zhàn)案例分解Netty中是如何解決拆包黏包問題的?
[6]基于Netty實現(xiàn)自定義消息通信協(xié)議（協(xié)議設計及解析應用實戰(zhàn)）
[7]全網(wǎng)最詳細最齊全的序列化技術及深度解析與應用實戰(zhàn)
在前面的內(nèi)容中，我們已經(jīng)由淺入深的理解了Netty的基礎知識和實現(xiàn)原理，相信大家已經(jīng)對Netty有了一個較為全面的理解。那么接下來，我們通過一個手寫RPC通信的實戰(zhàn)案例來帶大家了解Netty的實際應用。
為什么要選擇RPC來作為實戰(zhàn)呢？因為Netty本身就是解決通信問題，而在實際應用中，RPC協(xié)議框架是我們接觸得最多的一種，所以這個實戰(zhàn)能讓大家了解到Netty實際應用之外，還能理解RPC的底層原理。
RPC全稱為（Remote Procedure Call），是一種通過網(wǎng)絡從遠程計算機程序上請求服務，而不需要了解底層網(wǎng)絡技術的協(xié)議，簡單理解就是讓開發(fā)者能夠像調(diào)用本地服務一樣調(diào)用遠程服務。
既然是協(xié)議，那么它必然有協(xié)議的規(guī)范，如圖6-1所示。
為了達到“讓開發(fā)者能夠像調(diào)用本地服務那樣調(diào)用遠程服務”的目的，RPC協(xié)議需像圖6-1那樣實現(xiàn)遠程交互。
凡是滿足RPC協(xié)議的框架，我們成為RPC框架，在實際開發(fā)中，我們可以使用開源且相對成熟的RPC框架解決微服務架構下的遠程通信問題，常見的rpc框架：
基于上文中對于RPC協(xié)議的理解，如果我們自己去實現(xiàn)，需要考慮哪些技術呢？ 其實基于圖6-1的整個流程應該有一個大概的理解。
理解了RPC協(xié)議后，我們基于Netty來實現(xiàn)一個RPC通信框架。
代碼詳見附件 netty-rpc-example
需要引入的jar包:
org.springframework.boot spring-boot-starter org.projectlombok lombok com.alibaba fastjson 1.2.72 io.netty netty-all 模塊依賴關系:
provider依賴 netty-rpc-protocol和netty-rpc-api
cosumer依賴 netty-rpc-protocol和netty-rpc-api
public interface IUserService { String saveUser(String name);}
@Service@Slf4jpublic class UserServiceImpl implements IUserService { @Override public String saveUser(String name) { log.info("begin saveUser:"+name); return "Save User Success!"; }}注意，在當前步驟中，描述了case的部分，暫時先不用加，后續(xù)再加上
@ComponentScan(basePackages = {"com.example.spring","com.example.service"}) //case1(后續(xù)再加上)@SpringBootApplicationpublic class NettyRpcProviderMain { public static void main(String[] args) throws Exception { SpringApplication.run(NettyRpcProviderMain.class, args); new NettyServer("127.0.0.1",8080).startNettyServer(); //case2(后續(xù)再加上) }}開始寫通信協(xié)議模塊，這個模塊主要做幾個事情
之前我們講過自定義消息協(xié)議，我們在這里可以按照下面這個協(xié)議格式來定義好。
/* +----------------------------------------------+ | 魔數(shù) 2byte | 序列化算法 1byte | 請求類型 1byte | +----------------------------------------------+ | 消息 ID 8byte | 數(shù)據(jù)長度 4byte | +----------------------------------------------+ */@AllArgsConstructor@Datapublic class Header implements Serializable { /* +----------------------------------------------+ | 魔數(shù) 2byte | 序列化算法 1byte | 請求類型 1byte | +----------------------------------------------+ | 消息 ID 8byte | 數(shù)據(jù)長度 4byte | +----------------------------------------------+ */ private short magic; //魔數(shù)-用來驗證報文的身份（2個字節(jié)） private byte serialType; //序列化類型（1個字節(jié)） private byte reqType; //操作類型（1個字節(jié)） private long requestId; //請求id（8個字節(jié)） private int length; //數(shù)據(jù)長度（4個字節(jié)）}@Datapublic class RpcRequest implements Serializable { private String className; private String methodName; private Object[] params; private Class[] parameterTypes;}@Datapublic class RpcResponse implements Serializable { private Object data; private String msg;}@Datapublic class RpcProtocol implements Serializable { private Header header; private T content;} 上述消息協(xié)議定義中，涉及到幾個枚舉相關的類，定義如下
消息類型
public enum ReqType { REQUEST((byte)1), RESPONSE((byte)2), HEARTBEAT((byte)3); private byte code; private ReqType(byte code) { this.code=code; } public byte code(){ return this.code; } public static ReqType findByCode(int code) { for (ReqType msgType : ReqType.values()) { if (msgType.code() == code) { return msgType; } } return null; }}序列化類型
public enum SerialType { JSON_SERIAL((byte)0), JAVA_SERIAL((byte)1); private byte code; SerialType(byte code) { this.code=code; } public byte code(){ return this.code; }}public class RpcConstant { //header部分的總字節(jié)數(shù) public final static int HEAD_TOTAL_LEN=16; //魔數(shù) public final static short MAGIC=0xca;}這里演示兩種，一種是JSON方式，另一種是Java原生的方式
public interface ISerializer { byte[] serialize(T obj); T deserialize(byte[] data,Class clazz); byte getType();} public class JavaSerializer implements ISerializer{ @Override public byte[] serialize(T obj) { ByteArrayOutputStream byteArrayOutputStream= new ByteArrayOutputStream(); try { ObjectOutputStream outputStream= new ObjectOutputStream(byteArrayOutputStream); outputStream.writeObject(obj); return byteArrayOutputStream.toByteArray(); } catch (IOException e) { e.printStackTrace(); } return new byte[0]; } @Override public T deserialize(byte[] data, Class clazz) { ByteArrayInputStream byteArrayInputStream=new ByteArrayInputStream(data); try { ObjectInputStream objectInputStream= new ObjectInputStream(byteArrayInputStream); return (T) objectInputStream.readObject(); } catch (IOException e) { e.printStackTrace(); } catch (ClassNotFoundException e) { e.printStackTrace(); } return null; } @Override public byte getType() { return SerialType.JAVA_SERIAL.code(); }} public class JsonSerializer implements ISerializer{ @Override public byte[] serialize(T obj) { return JSON.toJSONString(obj).getBytes(); } @Override public T deserialize(byte[] data, Class clazz) { return JSON.parseObject(new String(data),clazz); } @Override public byte getType() { return SerialType.JSON_SERIAL.code(); }} 實現(xiàn)對序列化機制的管理
public class SerializerManager { private final static ConcurrentHashMap serializers=new ConcurrentHashMap(); static { ISerializer jsonSerializer=new JsonSerializer(); ISerializer javaSerializer=new JavaSerializer(); serializers.put(jsonSerializer.getType(),jsonSerializer); serializers.put(javaSerializer.getType(),javaSerializer); } public static ISerializer getSerializer(byte key){ ISerializer serializer=serializers.get(key); if(serializer==null){ return new JavaSerializer(); } return serializer; }} 由于自定義了消息協(xié)議，所以 需要自己實現(xiàn)編碼和解碼，代碼如下
@Slf4jpublic class RpcDecoder extends ByteToMessageDecoder { /* +----------------------------------------------+ | 魔數(shù) 2byte | 序列化算法 1byte | 請求類型 1byte | +----------------------------------------------+ | 消息 ID 8byte | 數(shù)據(jù)長度 4byte | +----------------------------------------------+ */ @Override protected void decode(ChannelHandlerContext ctx, ByteBuf in, List@Slf4jpublic class RpcEncoder extends MessageToByteEncoder> { /* +----------------------------------------------+ | 魔數(shù) 2byte | 序列化算法 1byte | 請求類型 1byte | +----------------------------------------------+ | 消息 ID 8byte | 數(shù)據(jù)長度 4byte | +----------------------------------------------+ */ @Override protected void encode(ChannelHandlerContext ctx, RpcProtocol 實現(xiàn)NettyServer構建。
@Slf4jpublic class NettyServer{ private String serverAddress; //地址 private int serverPort; //端口 public NettyServer(String serverAddress, int serverPort) { this.serverAddress = serverAddress; this.serverPort = serverPort; } public void startNettyServer() throws Exception { log.info("begin start Netty Server"); EventLoopGroup bossGroup=new NioEventLoopGroup(); EventLoopGroup workGroup=new NioEventLoopGroup(); try { ServerBootstrap bootstrap = new ServerBootstrap(); bootstrap.group(bossGroup, workGroup) .channel(NioServerSocketChannel.class) .childHandler(new RpcServerInitializer()); ChannelFuture channelFuture = bootstrap.bind(this.serverAddress, this.serverPort).sync(); log.info("Server started Success on Port:{}", this.serverPort); channelFuture.channel().closeFuture().sync(); }catch (Exception e){ log.error("Rpc Server Exception",e); }finally { workGroup.shutdownGracefully(); bossGroup.shutdownGracefully(); } }}public class RpcServerInitializer extends ChannelInitializer { @Override protected void initChannel(SocketChannel ch) throws Exception { ch.pipeline() .addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE,12,4,0,0)) .addLast(new RpcDecoder()) .addLast(new RpcEncoder()) .addLast(new RpcServerHandler()); }} public class RpcServerHandler extends SimpleChannelInboundHandler> { @Override protected void channelRead0(ChannelHandlerContext ctx, RpcProtocol msg) throws Exception { RpcProtocol resProtocol=new RpcProtocol<>(); Header header=msg.getHeader(); header.setReqType(ReqType.RESPONSE.code()); Object result=invoke(msg.getContent()); resProtocol.setHeader(header); RpcResponse response=new RpcResponse(); response.setData(result); response.setMsg("success"); resProtocol.setContent(response); ctx.writeAndFlush(resProtocol); } private Object invoke(RpcRequest request){ try { Class clazz=Class.forName(request.getClassName()); Object bean= SpringBeansManager.getBean(clazz); //獲取實例對象(CASE) Method declaredMethod=clazz.getDeclaredMethod(request.getMethodName(),request.getParameterTypes()); return declaredMethod.invoke(bean,request.getParams()); } catch (ClassNotFoundException | NoSuchMethodException e) { e.printStackTrace(); } catch (IllegalAccessException e) { e.printStackTrace(); } catch (InvocationTargetException e) { e.printStackTrace(); } return null; } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { super.exceptionCaught(ctx, cause); }} @Componentpublic class SpringBeansManager implements ApplicationContextAware { private static ApplicationContext applicationContext; @Override public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { SpringBeansManager.applicationContext=applicationContext; } public static T getBean(Class clazz){ return applicationContext.getBean(clazz); }} 需要注意，這個類的構建好之后，需要在netty-rpc-provider模塊的main方法中增加compone-scan進行掃描
@ComponentScan(basePackages = {"com.example.spring","com.example.service"}) //修改這里@SpringBootApplicationpublic class NettyRpcProviderMain { public static void main(String[] args) throws Exception { SpringApplication.run(NettyRpcProviderMain.class, args); new NettyServer("127.0.0.1",8080).startNettyServer(); // 修改這里 }}接下來開始實現(xiàn)消費端
public class RpcClientProxy { public T clientProxy(final Class interfaceCls,final String host,final int port){ return (T) Proxy.newProxyInstance (interfaceCls.getClassLoader(), new Class[]{interfaceCls}, new RpcInvokerProxy(host,port)); }} @Slf4jpublic class RpcInvokerProxy implements InvocationHandler { private String serviceAddress; private int servicePort; public RpcInvokerProxy(String serviceAddress, int servicePort) { this.serviceAddress = serviceAddress; this.servicePort = servicePort; } @Override public Object invoke(Object proxy, Method method, Object[] args) throws Throwable { log.info("begin invoke target server"); //組裝參數(shù) RpcProtocol protocol=new RpcProtocol<>(); long requestId= RequestHolder.REQUEST_ID.incrementAndGet(); Header header=new Header(RpcConstant.MAGIC, SerialType.JSON_SERIAL.code(), ReqType.REQUEST.code(),requestId,0); protocol.setHeader(header); RpcRequest request=new RpcRequest(); request.setClassName(method.getDeclaringClass().getName()); request.setMethodName(method.getName()); request.setParameterTypes(method.getParameterTypes()); request.setParams(args); protocol.setContent(request); //發(fā)送請求 NettyClient nettyClient=new NettyClient(serviceAddress,servicePort); //構建異步數(shù)據(jù)處理 RpcFuture future=new RpcFuture<>(new DefaultPromise<>(new DefaultEventLoop())); RequestHolder.REQUEST_MAP.put(requestId,future); nettyClient.sendRequest(protocol); return future.getPromise().get().getData(); }} 在netty-rpc-protocol這個模塊的protocol包路徑下，創(chuàng)建NettyClient
@Slf4jpublic class NettyClient { private final Bootstrap bootstrap; private final EventLoopGroup eventLoopGroup=new NioEventLoopGroup(); private String serviceAddress; private int servicePort; public NettyClient(String serviceAddress,int servicePort){ log.info("begin init NettyClient"); bootstrap=new Bootstrap(); bootstrap.group(eventLoopGroup) .channel(NioSocketChannel.class) .handler(new RpcClientInitializer()); this.serviceAddress=serviceAddress; this.servicePort=servicePort; } public void sendRequest(RpcProtocol protocol) throws InterruptedException { ChannelFuture future=bootstrap.connect(this.serviceAddress,this.servicePort).sync(); future.addListener(listener->{ if(future.isSuccess()){ log.info("connect rpc server {} success.",this.serviceAddress); }else{ log.error("connect rpc server {} failed .",this.serviceAddress); future.cause().printStackTrace(); eventLoopGroup.shutdownGracefully(); } }); log.info("begin transfer data"); future.channel().writeAndFlush(protocol); }} @Slf4jpublic class RpcClientInitializer extends ChannelInitializer { @Override protected void initChannel(SocketChannel ch) throws Exception { log.info("begin initChannel"); ch.pipeline() .addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE,12,4,0,0)) .addLast(new LoggingHandler()) .addLast(new RpcEncoder()) .addLast(new RpcDecoder()) .addLast(new RpcClientHandler()); }} 需要注意，Netty的通信過程是基于入站出站分離的，所以在獲取結果時，我們需要借助一個Future對象來完成。
@Slf4jpublic class RpcClientHandler extends SimpleChannelInboundHandler> { @Override protected void channelRead0(ChannelHandlerContext ctx, RpcProtocol msg) throws Exception { log.info("receive rpc server result"); long requestId=msg.getHeader().getRequestId(); RpcFuture future=RequestHolder.REQUEST_MAP.remove(requestId); future.getPromise().setSuccess(msg.getContent()); //返回結果 }} 在netty-rpc-protocol模塊中添加rpcFuture實現(xiàn)
@Datapublic class RpcFuture { //Promise是可寫的 Future, Future自身并沒有寫操作相關的接口, // Netty通過 Promise對 Future進行擴展,用于設置IO操作的結果 private Promise promise; public RpcFuture(Promise promise) { this.promise = promise; }} 保存requestid和future的對應結果
public class RequestHolder { public static final AtomicLong REQUEST_ID=new AtomicLong(); public static final Map REQUEST_MAP=new ConcurrentHashMap<>();} 需要源碼的同學，請關注公眾號[跟著Mic學架構]，回復關鍵字[rpc]，即可獲得
版權聲明：本博客所有文章除特別聲明外，均采用 CC BY-NC-SA 4.0 許可協(xié)議。轉載請注明來自
Mic帶你學架構！
如果本篇文章對您有幫助，還請幫忙點個關注和贊，您的堅持是我不斷創(chuàng)作的動力。歡迎關注「跟著Mic學架構」公眾號公眾號獲取更多技術干貨！
文章版權歸作者所有，未經(jīng)允許請勿轉載,若此文章存在違規(guī)行為，您可以聯(lián)系管理員刪除。
轉載請注明本文地址：http://systransis.cn/yun/123637.html
摘要：簡單來說就是把注冊的動作異步化，當異步執(zhí)行結束后會把執(zhí)行結果回填到中抽象類一般就是公共邏輯的處理，而這里的處理主要就是針對一些參數(shù)的判斷，判斷完了之后再調(diào)用方法。 閱讀這篇文章之前，建議先閱讀和這篇文章關聯(lián)的內(nèi)容。 1. 詳細剖析分布式微服務架構下網(wǎng)絡通信的底層實現(xiàn)原理（圖解） 2. (年薪60W的技巧)工作了5年，你真的理解Netty以及為什么要用嗎？（深度干貨）...
摘要：英文全名為，也叫遠程過程調(diào)用，其實就是一個計算機通信協(xié)議，它是一種通過網(wǎng)絡從遠程計算機程序上請求服務而不需要了解底層網(wǎng)絡技術的協(xié)議。 Hello,Dubbo 你好，dubbo，初次見面，我想和你交個朋友。 Dubbo你到底是什么？ 先給出一套官方的說法：Apache Dubbo是一款高性能、輕量級基于Java的RPC開源框架。 那么什么是RPC？ 文檔地址：http://dubbo.a...
摘要：前言此博客所述項目代碼已在開源歡迎大家一起貢獻點此進入最近一次寫博客還是年底謝謝大家持久以來的關注本篇博文將會教大家如何從到搭建一個簡單高效且拓展性強的框架什么是相信大家都或多或少使用過框架比如阿里的谷歌的的等等那么究竟什么是翻譯成中文 Cool-Rpc 前言 此博客所述項目代碼已在github開源,歡迎大家一起貢獻! 點此進入:Cool-RPC 最近一次寫博客還是17年底,謝謝大家...
摘要：從使用到原理學習線程池關于線程池的使用，及原理分析分析角度新穎面向切面編程的基本用法基于注解的實現(xiàn)在軟件開發(fā)中，分散于應用中多出的功能被稱為橫切關注點如事務安全緩存等。 Java 程序媛手把手教你設計模式中的撩妹神技 -- 上篇 遇一人白首，擇一城終老，是多么美好的人生境界，她和他歷經(jīng)風雨慢慢變老，回首走過的點點滴滴，依然清楚的記得當初愛情萌芽的模樣…… Java 進階面試問題列表 -...
閱讀 735·2023-04-25 19:43
閱讀 3981·2021-11-30 14:52
閱讀 3807·2021-11-30 14:52
閱讀 3871·2021-11-29 11:00
閱讀 3802·2021-11-29 11:00
閱讀 3904·2021-11-29 11:00
閱讀 3580·2021-11-29 11:00
閱讀 6182·2021-11-29 11:00
