成人免费xxxxx在线视频软件_久久精品久久久_亚洲国产精品久久久_天天色天天色_亚洲人成一区_欧美一级欧美三级在线观看

Java 從零開始手寫 RPC—如何實現客戶端調用服務端？

作者：老馬嘯西風
開發 后端
寫完了客戶端和服務端，那么如何實現客戶端和服務端的調用呢?下面就讓我們一起來看一下。希望本文對你有所幫助!

[[429717]]

寫完了客戶端和服務端，那么如何實現客戶端和服務端的調用呢?

下面就讓我們一起來看一下。

接口定義

計算方法

  1. package com.github.houbb.rpc.common.service; 
  2.  
  3.  
  4. import com.github.houbb.rpc.common.model.CalculateRequest; 
  5. import com.github.houbb.rpc.common.model.CalculateResponse; 
  6.  
  7.  
  8. /** 
  9.  * <p> 計算服務接口 </p> 
  10.  * 
  11.  * <pre> Created: 2018/8/24 下午4:47  </pre> 
  12.  * <pre> Project: fake  </pre> 
  13.  * 
  14.  * @author houbinbin 
  15.  * @since 0.0.1 
  16.  */ 
  17. public interface Calculator { 
  18.  
  19.  
  20.     /** 
  21.      * 計算加法 
  22.      * @param request 請求入參 
  23.      * @return 返回結果 
  24.      */ 
  25.     CalculateResponse sum(final CalculateRequest request); 
  26.  
  27.  
  28. } 

pojo

對應的參數對象：

  • CalculateRequest
  1. package com.github.houbb.rpc.common.model; 
  2.  
  3.  
  4. import java.io.Serializable; 
  5.  
  6.  
  7. /** 
  8.  * <p> 請求入參 </p> 
  9.  * 
  10.  * <pre> Created: 2018/8/24 下午5:05  </pre> 
  11.  * <pre> Project: fake  </pre> 
  12.  * 
  13.  * @author houbinbin 
  14.  * @since 0.0.3 
  15.  */ 
  16. public class CalculateRequest implements Serializable { 
  17.  
  18.  
  19.     private static final long serialVersionUID = 6420751004355300996L; 
  20.  
  21.  
  22.     /** 
  23.      * 參數一 
  24.      */ 
  25.     private int one; 
  26.  
  27.  
  28.     /** 
  29.      * 參數二 
  30.      */ 
  31.     private int two; 
  32.  
  33.  
  34.     public CalculateRequest() { 
  35.     } 
  36.  
  37.  
  38.     public CalculateRequest(int one, int two) { 
  39.         this.one = one; 
  40.         this.two = two; 
  41.     } 
  42.  
  43.  
  44.     //getter setter toString 
  45.  
  46.  
  47. } 
  • CalculateResponse
  1. package com.github.houbb.rpc.common.model; 
  2.  
  3.  
  4. import java.io.Serializable; 
  5.  
  6.  
  7. /** 
  8.  * <p> 請求入參 </p> 
  9.  * 
  10.  * <pre> Created: 2018/8/24 下午5:05  </pre> 
  11.  * <pre> Project: fake  </pre> 
  12.  * 
  13.  * @author houbinbin 
  14.  * @since 0.0.3 
  15.  */ 
  16. public class CalculateResponse implements Serializable { 
  17.  
  18.  
  19.     private static final long serialVersionUID = -1972014736222511341L; 
  20.  
  21.  
  22.     /** 
  23.      * 是否成功 
  24.      */ 
  25.    private boolean success; 
  26.  
  27.  
  28.     /** 
  29.      * 二者的和 
  30.      */ 
  31.    private int sum; 
  32.  
  33.  
  34.     public CalculateResponse() { 
  35.     } 
  36.  
  37.  
  38.     public CalculateResponse(boolean success, int sum) { 
  39.         this.success = success; 
  40.         this.sum = sum; 
  41.     } 
  42.  
  43.  
  44.     //getter setter toString 
  45. } 

客戶端

核心部分

RpcClient 需要添加對應的 Handler，調整如下：

  1. Bootstrap bootstrap = new Bootstrap(); 
  2. ChannelFuture channelFuture = bootstrap.group(workerGroup) 
  3.         .channel(NioSocketChannel.class) 
  4.         .option(ChannelOption.SO_KEEPALIVE, true) 
  5.         .handler(new ChannelInitializer<Channel>(){ 
  6.             @Override 
  7.             protected void initChannel(Channel ch) throws Exception { 
  8.                 ch.pipeline() 
  9.                         .addLast(new LoggingHandler(LogLevel.INFO)) 
  10.                         .addLast(new CalculateRequestEncoder()) 
  11.                         .addLast(new CalculateResponseDecoder()) 
  12.                         .addLast(new RpcClientHandler()); 
  13.             } 
  14.         }) 
  15.         .connect(RpcConstant.ADDRESS, port) 
  16.         .syncUninterruptibly(); 

netty 中的 handler 泳道設計的非常優雅，讓我們的代碼可以非常靈活地進行拓展。

接下來我們看一下對應的實現。

RpcClientHandler

  1. package com.github.houbb.rpc.client.handler; 
  2.  
  3.  
  4. import com.github.houbb.log.integration.core.Log; 
  5. import com.github.houbb.log.integration.core.LogFactory; 
  6. import com.github.houbb.rpc.client.core.RpcClient; 
  7. import com.github.houbb.rpc.common.model.CalculateRequest; 
  8. import com.github.houbb.rpc.common.model.CalculateResponse; 
  9. import io.netty.channel.ChannelHandlerContext; 
  10. import io.netty.channel.SimpleChannelInboundHandler; 
  11.  
  12.  
  13. /** 
  14.  * <p> 客戶端處理類 </p> 
  15.  * 
  16.  * <pre> Created: 2019/10/16 11:30 下午  </pre> 
  17.  * <pre> Project: rpc  </pre> 
  18.  * 
  19.  * @author houbinbin 
  20.  * @since 0.0.2 
  21.  */ 
  22. public class RpcClientHandler extends SimpleChannelInboundHandler { 
  23.  
  24.  
  25.     private static final Log log = LogFactory.getLog(RpcClient.class); 
  26.  
  27.  
  28.     @Override 
  29.     public void channelActive(ChannelHandlerContext ctx) throws Exception { 
  30.         CalculateRequest request = new CalculateRequest(1, 2); 
  31.  
  32.  
  33.         ctx.writeAndFlush(request); 
  34.         log.info("[Client] request is :{}", request); 
  35.     } 
  36.  
  37.  
  38.     @Override 
  39.     protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception { 
  40.         CalculateResponse response = (CalculateResponse)msg; 
  41.         log.info("[Client] response is :{}", response); 
  42.     } 
  43.  
  44.  
  45. } 

這里比較簡單，channelActive 中我們直接發起調用，入參的對象為了簡單，此處固定寫死。

channelRead0 中監聽服務端的相應結果，并做日志輸出。

CalculateRequestEncoder

請求參數是一個對象，netty 是無法直接傳輸的，我們將其轉換為基本對象：

  1. package com.github.houbb.rpc.client.encoder; 
  2.  
  3.  
  4. import com.github.houbb.rpc.common.model.CalculateRequest; 
  5. import io.netty.buffer.ByteBuf; 
  6. import io.netty.channel.ChannelHandlerContext; 
  7. import io.netty.handler.codec.MessageToByteEncoder; 
  8.  
  9.  
  10. /** 
  11.  * @author binbin.hou 
  12.  * @since 0.0.3 
  13.  */ 
  14. public class CalculateRequestEncoder extends MessageToByteEncoder<CalculateRequest> { 
  15.  
  16.  
  17.     @Override 
  18.     protected void encode(ChannelHandlerContext ctx, CalculateRequest msg, ByteBuf out) throws Exception { 
  19.         int one = msg.getOne(); 
  20.         int two = msg.getTwo(); 
  21.  
  22.  
  23.         out.writeInt(one); 
  24.         out.writeInt(two); 
  25.     } 
  26.  
  27.  
  28. } 

CalculateResponseDecoder

針對服務端的響應，也是同理。

我們需要把基本的類型，封裝轉換為我們需要的對象。

  1. package com.github.houbb.rpc.client.decoder; 
  2.  
  3.  
  4. import com.github.houbb.rpc.common.model.CalculateResponse; 
  5. import io.netty.buffer.ByteBuf; 
  6. import io.netty.channel.ChannelHandlerContext; 
  7. import io.netty.handler.codec.ByteToMessageDecoder; 
  8.  
  9.  
  10. import java.util.List; 
  11.  
  12.  
  13. /** 
  14.  * 響應參數解碼 
  15.  * @author binbin.hou 
  16.  * @since 0.0.3 
  17.  */ 
  18. public class CalculateResponseDecoder extends ByteToMessageDecoder { 
  19.  
  20.  
  21.     @Override 
  22.     protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception { 
  23.         boolean success = in.readBoolean(); 
  24.         int sum = in.readInt(); 
  25.  
  26.  
  27.         CalculateResponse response = new CalculateResponse(success, sum); 
  28.         out.add(response); 
  29.     } 
  30.  
  31.  
  32. } 

服務端

設置處理類

RpcServer 中的處理類要稍微調整一下，其他的保持不變。

  1. ServerBootstrap serverBootstrap = new ServerBootstrap(); 
  2. serverBootstrap.group(workerGroup, bossGroup) 
  3.         .channel(NioServerSocketChannel.class) 
  4.         // 打印日志 
  5.         .handler(new LoggingHandler(LogLevel.INFO)) 
  6.         .childHandler(new ChannelInitializer<Channel>() { 
  7.             @Override 
  8.             protected void initChannel(Channel ch) throws Exception { 
  9.                 ch.pipeline() 
  10.                         .addLast(new CalculateRequestDecoder()) 
  11.                         .addLast(new CalculateResponseEncoder()) 
  12.                         .addLast(new RpcServerHandler()); 
  13.             } 
  14.         }) 
  15.         // 這個參數影響的是還沒有被accept 取出的連接 
  16.         .option(ChannelOption.SO_BACKLOG, 128) 
  17.         // 這個參數只是過一段時間內客戶端沒有響應，服務端會發送一個 ack 包，以判斷客戶端是否還活著。 
  18.         .childOption(ChannelOption.SO_KEEPALIVE, true); 

RpcServerHandler

一開始這里是空實現，我們來添加一下對應的實現。

  1. package com.github.houbb.rpc.server.handler; 
  2.  
  3.  
  4. import com.github.houbb.log.integration.core.Log; 
  5. import com.github.houbb.log.integration.core.LogFactory; 
  6. import com.github.houbb.rpc.common.model.CalculateRequest; 
  7. import com.github.houbb.rpc.common.model.CalculateResponse; 
  8. import com.github.houbb.rpc.common.service.Calculator; 
  9. import com.github.houbb.rpc.server.service.CalculatorService; 
  10. import io.netty.channel.ChannelHandlerContext; 
  11. import io.netty.channel.SimpleChannelInboundHandler; 
  12.  
  13.  
  14. /** 
  15.  * @author binbin.hou 
  16.  * @since 0.0.1 
  17.  */ 
  18. public class RpcServerHandler extends SimpleChannelInboundHandler { 
  19.  
  20.  
  21.     private static final Log log = LogFactory.getLog(RpcServerHandler.class); 
  22.  
  23.  
  24.     @Override 
  25.     public void channelActive(ChannelHandlerContext ctx) throws Exception { 
  26.         final String id = ctx.channel().id().asLongText(); 
  27.         log.info("[Server] channel {} connected " + id); 
  28.     } 
  29.  
  30.  
  31.     @Override 
  32.     protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception { 
  33.         final String id = ctx.channel().id().asLongText(); 
  34.  
  35.  
  36.         CalculateRequest request = (CalculateRequest)msg; 
  37.         log.info("[Server] receive channel {} request: {} from ", id, request); 
  38.  
  39.  
  40.         Calculator calculator = new CalculatorService(); 
  41.         CalculateResponse response = calculator.sum(request); 
  42.  
  43.  
  44.         // 回寫到 client 端 
  45.         ctx.writeAndFlush(response); 
  46.         log.info("[Server] channel {} response {}", id, response); 
  47.     } 
  48.  
  49.  
  50. } 

讀取到客戶端的訪問之后，我們獲取到計算的入參 CalculateRequest，然后調用 sum 方法，獲取到對應的 CalculateResponse，將結果通知客戶端。

CalculateRequestDecoder

這里和客戶端是一一對應的，我們首先把 netty 傳遞的基本類型轉換為 CalculateRequest 對象。

  1. package com.github.houbb.rpc.server.decoder; 
  2.  
  3.  
  4. import com.github.houbb.rpc.common.model.CalculateRequest; 
  5. import io.netty.buffer.ByteBuf; 
  6. import io.netty.channel.ChannelHandlerContext; 
  7. import io.netty.handler.codec.ByteToMessageDecoder; 
  8.  
  9.  
  10. import java.util.List; 
  11.  
  12.  
  13. /** 
  14.  * 請求參數解碼 
  15.  * @author binbin.hou 
  16.  * @since 0.0.3 
  17.  */ 
  18. public class CalculateRequestDecoder extends ByteToMessageDecoder { 
  19.  
  20.  
  21.     @Override 
  22.     protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception { 
  23.         int one = in.readInt(); 
  24.         int two = in.readInt(); 
  25.  
  26.  
  27.         CalculateRequest request = new CalculateRequest(one, two); 
  28.         out.add(request); 
  29.     } 
  30.  
  31.  
  32. } 

CalculateResponseEncoder

這里和客戶端類似，我們需要把 response 轉換為基本類型進行網絡傳輸。

  1. package com.github.houbb.rpc.server.encoder; 
  2.  
  3.  
  4. import com.github.houbb.rpc.common.model.CalculateResponse; 
  5. import io.netty.buffer.ByteBuf; 
  6. import io.netty.channel.ChannelHandlerContext; 
  7. import io.netty.handler.codec.MessageToByteEncoder; 
  8.  
  9.  
  10. /** 
  11.  * @author binbin.hou 
  12.  * @since 0.0.3 
  13.  */ 
  14. public class CalculateResponseEncoder extends MessageToByteEncoder<CalculateResponse> { 
  15.  
  16.  
  17.     @Override 
  18.     protected void encode(ChannelHandlerContext ctx, CalculateResponse msg, ByteBuf out) throws Exception { 
  19.         boolean success = msg.isSuccess(); 
  20.         int result = msg.getSum(); 
  21.         out.writeBoolean(success); 
  22.         out.writeInt(result); 
  23.     } 
  24.  
  25.  
  26. } 

CalculatorService

服務端對應的實現類。

  1. public class CalculatorService implements Calculator { 
  2.  
  3.  
  4.     @Override 
  5.     public CalculateResponse sum(CalculateRequest request) { 
  6.         int sum = request.getOne()+request.getTwo(); 
  7.  
  8.  
  9.         return new CalculateResponse(true, sum); 
  10.     } 
  11.  
  12.  
  13. } 

測試

服務端

啟動服務端：

  1. new RpcServer().start(); 

服務端啟動日志：

  1. [DEBUG] [2021-10-05 11:53:11.795] [main] [c.g.h.l.i.c.LogFactory.setImplementation] - Logging initialized using 'class com.github.houbb.log.integration.adaptors.stdout.StdOutExImpl' adapter. 
  2. [INFO] [2021-10-05 11:53:11.807] [Thread-0] [c.g.h.r.s.c.RpcServer.run] - RPC 服務開始啟動服務端 
  3. 十月 05, 2021 11:53:13 上午 io.netty.handler.logging.LoggingHandler channelRegistered 
  4. 信息: [id: 0xd399474f] REGISTERED 
  5. 十月 05, 2021 11:53:13 上午 io.netty.handler.logging.LoggingHandler bind 
  6. 信息: [id: 0xd399474f] BIND: 0.0.0.0/0.0.0.0:9527 
  7. 十月 05, 2021 11:53:13 上午 io.netty.handler.logging.LoggingHandler channelActive 
  8. 信息: [id: 0xd399474f, L:/0:0:0:0:0:0:0:0:9527] ACTIVE 
  9. [INFO] [2021-10-05 11:53:13.101] [Thread-0] [c.g.h.r.s.c.RpcServer.run] - RPC 服務端啟動完成，監聽【9527】端口 

客戶端

啟動客戶端：

  1. new RpcClient().start(); 

日志如下：

  1. [DEBUG] [2021-10-05 11:54:12.158] [main] [c.g.h.l.i.c.LogFactory.setImplementation] - Logging initialized using 'class com.github.houbb.log.integration.adaptors.stdout.StdOutExImpl' adapter. 
  2. [INFO] [2021-10-05 11:54:12.164] [Thread-0] [c.g.h.r.c.c.RpcClient.run] - RPC 服務開始啟動客戶端 
  3. 十月 05, 2021 11:54:13 上午 io.netty.handler.logging.LoggingHandler channelRegistered 
  4. 信息: [id: 0x4d75c580] REGISTERED 
  5. 十月 05, 2021 11:54:13 上午 io.netty.handler.logging.LoggingHandler connect 
  6. 信息: [id: 0x4d75c580] CONNECT: /127.0.0.1:9527 
  7. 十月 05, 2021 11:54:13 上午 io.netty.handler.logging.LoggingHandler channelActive 
  8. 信息: [id: 0x4d75c580, L:/127.0.0.1:54030 - R:/127.0.0.1:9527] ACTIVE 
  9. [INFO] [2021-10-05 11:54:13.403] [Thread-0] [c.g.h.r.c.c.RpcClient.run] - RPC 服務啟動客戶端完成，監聽端口：9527 
  10. 十月 05, 2021 11:54:13 上午 io.netty.handler.logging.LoggingHandler write 
  11. 信息: [id: 0x4d75c580, L:/127.0.0.1:54030 - R:/127.0.0.1:9527] WRITE: 8B 
  12.          +-------------------------------------------------+ 
  13.          |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f | 
  14. +--------+-------------------------------------------------+----------------+ 
  15. |00000000| 00 00 00 01 00 00 00 02                         |........        | 
  16. +--------+-------------------------------------------------+----------------+ 
  17. 十月 05, 2021 11:54:13 上午 io.netty.handler.logging.LoggingHandler flush 
  18. 信息: [id: 0x4d75c580, L:/127.0.0.1:54030 - R:/127.0.0.1:9527] FLUSH 
  19. [INFO] [2021-10-05 11:54:13.450] [nioEventLoopGroup-2-1] [c.g.h.r.c.c.RpcClient.channelActive] - [Client] request is :CalculateRequest{one=1, two=2} 
  20. 十月 05, 2021 11:54:13 上午 io.netty.handler.logging.LoggingHandler channelRead 
  21. 信息: [id: 0x4d75c580, L:/127.0.0.1:54030 - R:/127.0.0.1:9527] READ: 5B 
  22.          +-------------------------------------------------+ 
  23.          |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f | 
  24. +--------+-------------------------------------------------+----------------+ 
  25. |00000000| 01 00 00 00 03                                  |.....           | 
  26. +--------+-------------------------------------------------+----------------+ 
  27. 十月 05, 2021 11:54:13 上午 io.netty.handler.logging.LoggingHandler channelReadComplete 
  28. 信息: [id: 0x4d75c580, L:/127.0.0.1:54030 - R:/127.0.0.1:9527] READ COMPLETE 
  29. [INFO] [2021-10-05 11:54:13.508] [nioEventLoopGroup-2-1] [c.g.h.r.c.c.RpcClient.channelRead0] - [Client] response is :CalculateResponse{success=true, sum=3} 

可以看到，輸出了對應的請求參數和響應結果。

當然，此時服務端也有對應的新增日志：

  1. 十月 05, 2021 11:54:13 上午 io.netty.handler.logging.LoggingHandler channelRead 
  2. 信息: [id: 0xd399474f, L:/0:0:0:0:0:0:0:0:9527] READ: [id: 0xbc9f5927, L:/127.0.0.1:9527 - R:/127.0.0.1:54030] 
  3. 十月 05, 2021 11:54:13 上午 io.netty.handler.logging.LoggingHandler channelReadComplete 
  4. 信息: [id: 0xd399474f, L:/0:0:0:0:0:0:0:0:9527] READ COMPLETE 
  5. [INFO] [2021-10-05 11:54:13.432] [nioEventLoopGroup-2-1] [c.g.h.r.s.h.RpcServerHandler.channelActive] - [Server] channel {} connected 00e04cfffe360988-00001d34-00000001-2a80d950d8166c0c-bc9f5927 
  6. [INFO] [2021-10-05 11:54:13.495] [nioEventLoopGroup-2-1] [c.g.h.r.s.h.RpcServerHandler.channelRead0] - [Server] receive channel 00e04cfffe360988-00001d34-00000001-2a80d950d8166c0c-bc9f5927 request: CalculateRequest{one=1, two=2} from  
  7. [INFO] [2021-10-05 11:54:13.505] [nioEventLoopGroup-2-1] [c.g.h.r.s.h.RpcServerHandler.channelRead0] - [Server] channel 00e04cfffe360988-00001d34-00000001-2a80d950d8166c0c-bc9f5927 response CalculateResponse{success=true, sum=3} 

 

責任編輯：姜華 來源： 今日頭條
Java 語言 Java 基礎
相關推薦

2021-10-14 08:39:17

Java Netty Java 基礎

2021-10-21 08:21:10

Java Reflect Java 基礎

2021-10-27 08:10:15

Java 客戶端 Java 基礎

2021-10-13 08:21:52

Java websocket Java 基礎

2021-10-20 08:05:18

Java 序列化 Java 基礎

2009-08-21 15:36:41

服務端與客戶端

2009-08-21 15:54:40

服務端與客戶端

2021-10-29 08:07:30

Java timeout Java 基礎

2023-11-17 09:13:36

2010-03-18 17:47:07

Java 多客戶端通信

2024-03-06 14:58:52

客戶端微服務架構

2009-08-21 15:59:22

服務端與客戶端通信

2009-08-21 16:14:52

服務端與客戶端通信

2011-09-09 09:44:23

WCF

2023-03-06 08:01:56

MySQLCtrl + C

2010-05-31 10:11:32

瘦客戶端

2023-04-03 08:13:05

MySQLCtrl + C

2014-06-01 11:03:13

VDI零客戶端

2010-11-19 14:22:04

oracle服務端

2013-05-09 09:33:59

點贊
收藏

51CTO技術棧公眾號

主站蜘蛛池模板： 色毛片| 日本精品一区二区 | 日韩 国产 在线 | 一区二区在线不卡 | 久久久久久久久一区 | 另类 综合 日韩 欧美 亚洲 | 国产女人叫床高潮大片免费 | 亚洲综合色视频在线观看 | 日韩中文字幕 | 欧美日韩精品影院 | 日本在线播放一区二区 | 一区二区三区国产 | 欧美综合一区二区 | 日本三级电影在线观看视频 | 欧美亚洲国产一区 | 欧美不卡在线 | 亚洲在线久久 | 国产精品99久久久久久久久 | 日韩欧美精品一区 | 中文字幕在线中文 | 56pao在线| 色综合一区 | 在线免费观看亚洲 | 亚洲电影专区 | 韩日一区二区三区 | 日韩中文字幕 | 91最新视频 | 人人干在线| 精品国产乱码久久久久久闺蜜 | 精品免费国产视频 | 日本免费在线 | 欧美日一区二区 | 久久久久久久久久久久一区二区 | 高清久久久 | 毛片一区 | 国产精品99久久久久久久久 | 亚洲美女一区 | 黄色一级免费观看 | 免费一区二区三区 | 久久久www成人免费无遮挡大片 | 麻豆av电影网 |