本身用 Netty 實現一個簡單的 RPC

 

 目錄：

1. 需求
2. 設計
3. 實現
   1. 建立 maven 項目，導入 Netty 4.1.16。
   2. 項目目錄結構
   3. 設計接口
   4. 提供者相關實現
   5. 消費者相關實現
   6. 測試結果
4. 總結

源碼地址：github 地址

 

前言

衆所周知，dubbo 底層使用了 Netty 做爲網絡通信框架，而 Netty 的高性能咱們以前也分析過源碼，對他也算仍是比較瞭解了。今天咱們就本身用 Netty 實現一個簡單的 RPC 框架。

 

1. 需求

模仿 dubbo，消費者和提供者約定接口和協議，消費者遠程調用提供者，提供者返回一個字符串，消費者打印提供者返回的數據。底層網絡通訊使用 Netty 4.1.16。

2. 設計

1. 建立一個接口，定義抽象方法。用於消費者和提供者之間的約定。
2. 建立一個提供者，該類須要監聽消費者的請求，並按照約定返回數據。
3. 建立一個消費者，該類須要透明的調用本身不存在的方法，內部須要使用 Netty 請求提供者返回數據。

3. 實現

1. 建立 maven 項目，導入 Netty 4.1.16。

 1   <groupId>cn.thinkinjava</groupId>
 2   <artifactId>rpc-demo</artifactId>
 3   <version>1.0-SNAPSHOT</version>
 4 
 5   <dependencies>
 6     <dependency>
 7       <groupId>io.netty</groupId>
 8       <artifactId>netty-all</artifactId>
 9       <version>4.1.16.Final</version>
10   </dependency>

 

 

2. 項目目錄結構以下：

 

 

 

3. 設計接口

一個簡單的 hello world：

1 public interface HelloService {
2   String hello(String msg);
3 }

 

 

 

4. 提供者相關實現

4.1. 首先實現約定接口，用於返回客戶端數據：

1 /**
2  * 實現類
3  */
4 public class HelloServiceImpl implements HelloService {
5   public String hello(String msg) {
6     return msg != null ? msg + " -----> I am fine." : "I am fine.";
7   }
8 }

 

 

 

4.2. 實現 Netty 服務端和自定義 handler

啓動 Netty Server 代碼：

 1   private static void startServer0(String hostName, int port) {
 2     try {
 3       ServerBootstrap bootstrap = new ServerBootstrap();
 4       NioEventLoopGroup eventLoopGroup = new NioEventLoopGroup();
 5       bootstrap.group(eventLoopGroup)
 6           .channel(NioServerSocketChannel.class)
 7           .childHandler(new ChannelInitializer<SocketChannel>() {
 8             @Override
 9             protected void initChannel(SocketChannel ch) throws Exception {
10               ChannelPipeline p = ch.pipeline();
11               p.addLast(new StringDecoder());
12               p.addLast(new StringEncoder());
13               p.addLast(new HelloServerHandler());
14             }
15           });
16       bootstrap.bind(hostName, port).sync();
17     } catch (InterruptedException e) {
18       e.printStackTrace();
19     }
20   }

 

 

上面的代碼中添加了 String類型的編解碼 handler，添加了一個自定義 handler。

自定義 handler 邏輯以下：

 1 /**
 2  * 用於處理請求數據
 3  */
 4 public class HelloServerHandler extends ChannelInboundHandlerAdapter {
 5   @Override
 6   public void channelRead(ChannelHandlerContext ctx, Object msg) {
 7 
 8     // 如何符合約定，則調用本地方法，返回數據
 9     if (msg.toString().startsWith(ClientBootstrap.providerName)) {
10       String result = new HelloServiceImpl()
11           .hello(msg.toString().substring(msg.toString().lastIndexOf("#") + 1));
12       ctx.writeAndFlush(result);
13     }
14   }
15 }

 

 

這裏顯示判斷了是否符合約定（並無使用複雜的協議，只是一個字符串判斷），而後建立一個具體實現類，並調用方法寫回客戶端。

還須要一個啓動類：

1 public class ServerBootstrap {
2   public static void main(String[] args) {
3     NettyServer.startServer("localhost", 8088);
4   }
5 }

 

 

好，關於提供者的代碼就寫完了，主要就是建立一個 netty 服務端，實現一個自定義的 handler，自定義 handler 判斷是否符合之間的約定（算是協議吧），若是符合，就建立一個接口的實現類，並調用他的方法返回字符串。

 

5. 消費者相關實現

消費者有一個須要注意的地方，就是調用須要透明，也就是說，框架使用者不用關心底層的網絡實現。這裏咱們可使用 JDK 的動態代理來實現這個目的。

思路：客戶端調用代理方法，返回一個實現了 HelloService 接口的代理對象，調用代理對象的方法，返回結果。

咱們須要在代理中作手腳，當調用代理方法的時候，咱們須要初始化 Netty 客戶端，還須要向服務端請求數據，並返回數據。

 

5.1. 首先建立代理相關的類

 

 1 public class RpcConsumer {
 2 
 3   private static ExecutorService executor = Executors
 4       .newFixedThreadPool(Runtime.getRuntime().availableProcessors());
 5 
 6   private static HelloClientHandler client;
 7 
 8   /**
 9    * 建立一個代理對象
10    */
11   public Object createProxy(final Class<?> serviceClass,
12       final String providerName) {
13     return Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(),
14         new Class<?>[]{serviceClass}, (proxy, method, args) -> {
15           if (client == null) {
16             initClient();
17           }
18           // 設置參數
19           client.setPara(providerName + args[0]);
20           return executor.submit(client).get();
21         });
22   }
23 
24   /**
25    * 初始化客戶端
26    */
27   private static void initClient() {
28     client = new HelloClientHandler();
29     EventLoopGroup group = new NioEventLoopGroup();
30     Bootstrap b = new Bootstrap();
31     b.group(group)
32         .channel(NioSocketChannel.class)
33         .option(ChannelOption.TCP_NODELAY, true)
34         .handler(new ChannelInitializer<SocketChannel>() {
35           @Override
36           public void initChannel(SocketChannel ch) throws Exception {
37             ChannelPipeline p = ch.pipeline();
38             p.addLast(new StringDecoder());
39             p.addLast(new StringEncoder());
40             p.addLast(client);
41           }
42         });
43     try {
44       b.connect("localhost", 8088).sync();
45     } catch (InterruptedException e) {
46       e.printStackTrace();
47     }
48   }
49 }

View Code

 

 

 

該類有 2 個方法，建立代理和初始化客戶端。

初始化客戶端邏輯： 建立一個 Netty 的客戶端，並鏈接提供者，並設置一個自定義 handler，和一些 String 類型的編解碼器。

建立代理邏輯：使用 JDK 的動態代理技術，代理對象中的 invoke 方法實現以下：
若是 client 沒有初始化，則初始化 client，這個 client 既是 handler ，也是一個 Callback。將參數設置進 client ，使用線程池調用 client 的 call 方法並阻塞等待數據返回。

 

 

看看 HelloClientHandler 的實現：

 

 1 public class HelloClientHandler extends ChannelInboundHandlerAdapter implements Callable {
 2 
 3   private ChannelHandlerContext context;
 4   private String result;
 5   private String para;
 6 
 7   @Override
 8   public void channelActive(ChannelHandlerContext ctx) {
 9     context = ctx;
10   }
11 
12   /**
13    * 收到服務端數據，喚醒等待線程
14    */
15   @Override
16   public synchronized void channelRead(ChannelHandlerContext ctx, Object msg) {
17     result = msg.toString();
18     notify();
19   }
20 
21   /**
22    * 寫出數據，開始等待喚醒
23    */
24   @Override
25   public synchronized Object call() throws InterruptedException {
26     context.writeAndFlush(para);
27     wait();
28     return result;
29   }
30 
31   void setPara(String para) {
32     this.para = para;
33   }
34 }

View Code

 

 

 

該類緩存了 ChannelHandlerContext，用於下次使用，有兩個屬性：返回結果和請求參數。

當成功鏈接後，緩存 ChannelHandlerContext，當調用 call 方法的時候，將請求參數發送到服務端，等待。當服務端收到並返回數據後，調用 channelRead 方法，將返回值賦值個 result，並喚醒等待在 call 方法上的線程。此時，代理對象返回數據。

 

再看看設計的測試類：

 

 1 public class ClientBootstrap {
 2 
 3   public static final String providerName = "HelloService#hello#";
 4 
 5   public static void main(String[] args) throws InterruptedException {
 6 
 7     RpcConsumer consumer = new RpcConsumer();
 8     // 建立一個代理對象
 9     HelloService service = (HelloService) consumer
10         .createProxy(HelloService.class, providerName);
11     for (; ; ) {
12       Thread.sleep(1000);
13       System.out.println(service.hello("are you ok ?"));
14     }
15   }
16 }

 

 

 

測試類首先建立了一個代理對象，而後每隔一秒鐘調用代理的 hello 方法，並打印服務端返回的結果。

 

 

測試結果

 

 

 

 

 

總結

看了這麼久的 Netty 源碼，咱們終於實現了一個本身的 Netty 應用，雖然這個應用很簡單，甚至代碼寫的有些粗糙，但功能仍是實現了，RPC 的目的就是容許像調用本地服務同樣調用遠程服務，須要對使用者透明，因而咱們使用了動態代理。並使用 Netty 的 handler 發送數據和響應數據，完成了一次簡單的 RPC 調用。

固然，仍是那句話，代碼比較簡單，主要是思路，以及瞭解 RPC 底層的實現。

好吧。good luck！！！！