可参考的自定义实现rpc框架

 

基于netty的rpc实现过程：

 

消费者

    实现代理接口，在组装好数据之后实例化成代理

    设置好消费者的netty处理管道过滤器---尤其是异步处理

 

 

服务提供者

    发布服务

    设置服务提供方的netty管道过滤处理器---接受接口参数，结果处理器

 

消费者实例化的时之后，基于次消费者获取服务对象的代理对象，调用了服务代理的方法之后就会走invoke方法，调用封装好的消费端的netty--send将调用的服务，信息，参数信息传过去

 

服务提供端发布服务接口信息到一个内存（类的成员变量）中，通过netty管道中自定义的过滤器，接收消息，利用反射实例化处接口的对象，调用这个接口对象的方法

 

 消费端和服务端交互过程：

客户端的读在自定义的管道处理中（读服务方返回的结果），写在发送请求的方法(代理中)中（写请求的参数）

 

服务端读（读取客户端的请求参数），调用复写的handle方法反射出服务实例调用，写（写入返回结果）都在自定义的管道处理器中

 

 

    

rpc的异步原理：

 

futuer结合thradlocal

 

future  future在最终get的时候相当于同步，在发出请求此时是异步（主线程可继续往下），最终的get还是异步，

 

用future时最终的同步

例如

Object result=future.getResult(timeout, TimeUnit.MILLISECONDS);

 

还有一种思路就是结合threadlocal类，请求的时候直接发出，然后返回null,然后在需要的时候（主线程没有停止）通过这个线程从threadlocal中获取结果集

例如：

ResponseFuture

 

 

 

 

 

一个管道必加的处理器，编码，解码，读写处理器---进行消息的发送和接收

发布的时候根据端口（服务端不必写ip）,监听端口的请求

请求连接需要ip+端口

 

 

发送 接收就是操作管道  通过消费管道的自定义过滤器写  继承ChannelInboundHandlerAdapter  （实例化接收服务的时候连接，连接的时候就建立了管道）

 

 

连接获取管道；

public RpcConsumerImpl()

{

String ip=System.getProperty("SIP");

//String ip="127.0.0.1";

this.asyncMethods=new HashMap<String,ResponseCallbackListener>();

this.connection=new RpcNettyConnection(ip,8888);

this.connection.connect();

connection_list=new ArrayList<RpcConnection>();

int num=Runtime.getRuntime().availableProcessors()/3 -2;

for (int i = 0; i < num; i++) {

connection_list.add(new RpcNettyConnection(ip, 8888));

}

for (RpcConnection conn:connection_list) 

{

conn.connect();

}

 

}

 

 

获取管道发送：

public Object Send(RpcRequest request,boolean async) {

if(channel==null)

channel=getChannel(inetAddr.toString());

if(channel!=null)

{

final InvokeFuture<Object> future=new InvokeFuture<Object>();

futrues.put(request.getRequestId(), future);

future.setMethod(request.getMethodName());

ChannelFuture cfuture=channel.writeAndFlush(request);

cfuture.addListener(new ChannelFutureListener() {

@Override

public void operationComplete(ChannelFuture rfuture) throws Exception {

if(!rfuture.isSuccess()){

future.setCause(rfuture.cause());

}

}

});

resultFuture=new ResultFuture<Object>(timeout);

resultFuture.setRequestId(request.getRequestId());

try

{

if(async)//异步执行的话直接返回

{

ResponseFuture.setFuture(resultFuture);

return null;

}

Object result=future.getResult(timeout, TimeUnit.MILLISECONDS);

return result;

}

catch(RuntimeException e)

{

throw e;

}

finally

{

//这个结果已经收到

if(!async)

futrues.remove(request.getRequestId());

}

}

else

{

return null;

}

}

 

 

接收端：

通过接收管道的自定义过滤器，复写监听读   继承ChannelInboundHandlerAdapter

 

 

 

  request   实现了序列化

 

发布的时候根据端口（服务端不必写ip）,监听端口的请求

 

发布接受连接请求：

@Override

public void publish() {

handlerMap.put(interfaceclazz.getName(), classimplement);

// TODO Auto-generated method stub

EventLoopGroup bossGroup = new NioEventLoopGroup();

        EventLoopGroup workerGroup = new NioEventLoopGroup();

        try {

            ServerBootstrap serverBootstrap = new ServerBootstrap();

            // server端采用简洁的连写方式，client端才用分段普通写法。

           serverBootstrap.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class)

           .handler(new LoggingHandler(LogLevel.INFO))

           .childHandler(new ChannelInitializer<SocketChannel>() {

                           @Override

                           public void initChannel(SocketChannel ch)

                                    throws Exception {

                            ch.pipeline().addLast(new RpcEncoder(RpcResponse.class));

                            ch.pipeline().addLast(new RpcDecoder(RpcRequest.class));

//                        ch.pipeline().addLast(new FSTNettyEncode());

//                            ch.pipeline().addLast(new FSTNettyDecode());

                           ch.pipeline().addLast(new RpcRequestHandler(handlerMap));

                          }

                     })

                     .option(ChannelOption.SO_KEEPALIVE , true )

                     .childOption(ChannelOption.TCP_NODELAY, true)

                     .option(ChannelOption.SO_SNDBUF, 1024)

                     .option(ChannelOption.SO_RCVBUF, 2048);

           

           

           ChannelFuture f = serverBootstrap.bind(8888).sync();

           f.channel().closeFuture().sync();

       } catch (InterruptedException e) {

       } finally {

           workerGroup.shutdownGracefully();

           bossGroup.shutdownGracefully();

       }

}

 

 

 

 

读取发送端的参数:

 

 

 @Override

  public void channelRead(

      ChannelHandlerContext ctx, Object msg) throws Exception {

  RpcRequest request=(RpcRequest)msg;

  String host=ctx.channel().remoteAddress().toString();

  //更新上下文

  UpdateRpcContext(host,request.getContext());

  //TODO 获取接口名 函数名 参数    找到实现类   反射实现

  RpcResponse response = new RpcResponse();

  response.setRequestId(request.getRequestId());

  try 

  {

  Object result = handle(request);

  if(cacheName!=null&&cacheName.equals(result))

  {

  response.setAppResponse(cacheVaule);

  }

  else

  {

  response.setAppResponse(ByteObjConverter.ObjectToByte(result));

  cacheName=result;

  cacheVaule=ByteObjConverter.ObjectToByte(result);

  }

  } 

  catch (Throwable t) 

  {

  //response.setErrorMsg(t);

  response.setExption(Tool.serialize(t));

  response.setClazz(t.getClass());

  }

  ctx.writeAndFlush(response);

  }

 

 

 

 

 

 

 

 

 参考：

https://blog.csdn.net/zhujunxxxxx/article/details/48742529

 

https://github.com/zhujunxxxxx/