Skip to content
This repository has been archived by the owner on Aug 2, 2024. It is now read-only.

Commit

Permalink
Update readme
Browse files Browse the repository at this point in the history
  • Loading branch information
liuzhengyang committed Jan 17, 2017
1 parent 00985d0 commit bbd9255
Showing 1 changed file with 279 additions and 15 deletions.
294 changes: 279 additions & 15 deletions README.md
View file
Original file line number Diff line number Diff line change
Expand Up @@ -45,8 +45,8 @@ Client invoke => message object => encode to bytes => tranport through net => se
# IO传输

## TCP粘包、拆包
tcp传输过程中的任何一个节点都可能会将数据包拆分或合并，最终保证的是数据到达终点的顺序是一致的。由于TCP只关心字节数组流，并不知晓上层的数据格式。所以需要应用层来处理数据是否完整的问题，一般在数据协议上会采用一个字段来表示数据的长度，知道了消息的长度，就可以解决粘包的问题。对于拆包问题，当读到的数据长度比数据长度小时，要继续等待数据。Netty中提供了`LengthFieldBasedFrameDecoder`这个类帮助我们简化粘包、拆包问题，如我们定义协议为表示数据长度的4字节 + 数据，数据长度不包括长度字段本身，假设传输数据"0101"会被转化为 0x 00 00 00 04 01 01。
解码时,`LengthFieldBasedFrameDecoder`中的decode方法中的
tcp传输过程中的任何一个节点都可能会将数据包拆分或合并，最终保证的是数据到达终点的顺序是一致的。由于TCP只关心字节数组流，并不知晓上层的数据格式。所以需要应用层来处理数据是否完整的问题，一般在数据协议上会采用一个字段来表示数据的长度，知道了消息的长度，就可以解决粘包的问题。对于拆包问题，当读到的数据长度比数据长度小时，要继续等待数据。Netty中提供了`LengthFieldBasedFrameDecoder`这个类帮助我们简化粘包、拆包问题，如我们定义协议为表示数据长度的4字节 + 数据，数据长度不包括长度字段本身，假设传输数据"0101"被转化为 0x 00 00 00 04 01 01。可以使用LengthFieldBasedFrameDecoder(0, 4, 0, 4)来进行解码。
解码遇到拆包时,`LengthFieldBasedFrameDecoder`中的decode方法中的
```
int frameLengthInt = (int) frameLength;
if (in.readableBytes() < frameLengthInt) {
Expand All @@ -56,15 +56,279 @@ int frameLengthInt = (int) frameLength;
会返回null,数据会在`ByteToMessageDecoder`中累积，直到数据累积充足，解码后返回对象，由后续的Handler处理。

# 服务的注册发布和监听
类似于域名访问的问题，我们无需记住一个http服务后的服务器是哪些，它们的变更对我们都是透明的。对应RPC服务，经常需要使用集群来保证服务的稳定性和共同提高系统的性能。为此需要提供一个注册中心，当服务器启动时进行服务注册，服务器宕机时注册中心能够检测到并将其从服务注册中心删除。客户端要访问一个服务时先到注册中心获取服务列表，然后建立连接进行访问，并且当服务列表发生变化时会收到通知。
类似于域名访问的问题，我们无需记住一个http服务后的服务器是哪些，它们的变更对我们都是透明的。对应RPC服务，经常需要使用集群来保证服务的稳定性和共同提高系统的性能。为此需要提供一个注册中心，当服务器启动时进行服务注册，服务器宕机时注册中心能够检测到并将其从服务注册中心删除。客户端要访问一个服务时先到注册中心获取服务列表，缓存到本地，然后建立连接进行访问，并且当服务列表发生变化时会收到通知并修改本地缓存。
![注册发布](http://oek9m2h2f.bkt.clouddn.com/RpcPubSub.png)

# 服务路由
有注册中心后，调用方调用服务提供者时可以动态的获取调用方的地址等信息进行选择，类似域名的机制。这样增加了一层抽象，避免了写死ip等问题。又一次说明了`Any problem in computer science can be solved by adding another level of indirection`。当有多个服务提供者时，调用方需要在其中选择一个进行调用，常见的调用策略有随机、Round-Robin(轮询)、权重比例等。采取权重方式可以通过服务调用的耗时、异常数量进行动态调用权重，也可以进行人工调整。当我们需要更复杂的控制策略是，可以通过脚本编写策略，并可以动态修改。

# IO调用方式
## nio和bio
bio指的是传统的阻塞io，在Java中使用方式是Socket、ServerSocket、InputStream、OutputStream的组合。当读数据是，如果没有数据可读，该线程会被切换为阻塞状态，直到数据可读等待处理器调度运行，会进行两次上下文切换和两次用户态内核态切换，并且这样一个线程同时只能处理一个连接，线程是比较宝贵的资源，除了创建和销毁线程的开销外，JVM中每个线程都会有1MB左右的栈大小，这样一线程一连接的方式无法应对单机数万的情况，这时的内存消耗和上下文切换的成本都非常高。Nio指non blocking io，即非阻塞io，当数据不可读时会返回一个错误而不是阻塞，在Java中，常用Selector、SocketChannel、ServerSocketChannel、ByteBuffer的组合实现nio服务，在一个Selector上可以监听多个连接的是否可读、可写、可连接等状态，这样一个线程就可以同时处理很多个连接，能够提高系统连接能力。
下面的经典的几张图可以说明bio和nio的区别
![bio](http://oek9m2h2f.bkt.clouddn.com/io:bio)
![nio](http://oek9m2h2f.bkt.clouddn.com/nio.jpg)
## 同步和异步
同步指发出一个请求后是否阻塞并一直等待结果返回，而异步可以在发送请求后先去执行其他任务，在一段时间后再获取结果或通过注册监听器设置回调。在Java中一般是通过Future或者一些Listener来实现异步调用。如ExecutorService.submit()方法返回一个Future，调用Future的get时会阻塞，可以在get时设置超时时间。Guava和Netty中的Future实现可以设置Listener在结果成功或失败时进行回调。

# 实现
下面利用一些好用的框架帮助我们快速的实现一个RPC。 源代码在 *[simple-rpc](https://github.com/liuzhengyang/simple-rpc)*
* netty 负责数据传输部分，netty作为异步事件驱动的高性能IO框架，使用方便久经考验，比手工编写nio代码方便不易出错。
* protostuff 负责序列化和反序列化。google的 protobuf需要编写IDL文件然后生成，好处是能够生成各个语言的代码。但是开发起来很繁琐，使用protostuff免去编写IDL文件并生成的痛苦。


## 请求和返回的抽象
```
@Data
public class Request {
private long requestId;
private Class<?> clazz;
private String method;
private Class<?>[] parameterTypes;
private Object[] params;
private long requestTime;
}
@Data
public class Response {
private long requestId;
private Object response;
}
# 消息协议
当前采用简单的在消息体前加上4byte的消息长度值
## 使用示例
// 需要先启动一个zookeeper作为服务注册发现中心
```

## 编解码部分
```
public class RequestCodec extends ByteToMessageCodec<Request>{
protected void encode(ChannelHandlerContext ctx, Request msg, ByteBuf out) throws Exception {
byte[] bytes = Serializer.serialize(msg);
int length = bytes.length;
out.writeInt(length);
ByteBuf byteBuf = out.writeBytes(bytes);
}
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
int length = in.readInt();
byte[] buffer = new byte[length];
in.readBytes(buffer);
Request request = Serializer.deserialize(Request.class, buffer);
out.add(request);
}
}
public class ResponseCodec extends ByteToMessageCodec<Response>{
private static final Logger LOGGER = LoggerFactory.getLogger(ResponseCodec.class);
protected void encode(ChannelHandlerContext ctx, Response msg, ByteBuf out) throws Exception {
LOGGER.info("Encode {}", msg);
byte[] bytes = Serializer.serialize(msg);
int length = bytes.length;
out.writeInt(length);
ByteBuf byteBuf = out.writeBytes(bytes);
}
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
int length = in.readInt();
byte[] buffer = new byte[length];
in.readBytes(buffer);
Response response = Serializer.deserialize(Response.class, buffer);
out.add(response);
LOGGER.info("Decode Result: {}", response);
}
}
```


## Client
```
public class RpcClientHandler extends SimpleChannelInboundHandler<Response>{
private static final Logger LOGGER = LoggerFactory.getLogger(RpcClientHandler.class);
protected void channelRead0(ChannelHandlerContext ctx, Response msg) throws Exception {
LOGGER.info("Receive {}", msg);
BlockingQueue<Response> blockingQueue = RpcClient.responseMap.get(msg.getRequestId());
blockingQueue.put(msg);
}
}
public class RpcClient {
private static AtomicLong atomicLong = new AtomicLong();
private String serverIp;
private int port;
private boolean started;
private Channel channel;
EventLoopGroup eventLoopGroup = new NioEventLoopGroup();
public static ConcurrentMap<Long, BlockingQueue<Response>> responseMap = new ConcurrentHashMap<Long, BlockingQueue<Response>>();
public RpcClient(String serverIp, int port) {
this.serverIp = serverIp;
this.port = port;
}
public void init() {
Bootstrap bootstrap = new Bootstrap();
bootstrap.channel(NioSocketChannel.class)
.group(eventLoopGroup)
.handler(new ChannelInitializer<Channel>() {
protected void initChannel(Channel ch) throws Exception {
ch.pipeline().addLast(new LoggingHandler(LogLevel.INFO))
.addLast(new ResponseCodec())
.addLast(new RpcClientHandler())
.addLast(new RequestCodec())
;
}
});
try {
ChannelFuture f = bootstrap.connect(serverIp, port).sync();
this.channel = f.channel();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
public Response sendMessage(Class<?> clazz, Method method, Object[] args) {
Request request = new Request();
request.setRequestId(atomicLong.incrementAndGet());
request.setMethod(method.getName());
request.setParams(args);
request.setClazz(clazz);
request.setParameterTypes(method.getParameterTypes());
this.channel.writeAndFlush(request);
BlockingQueue<Response> blockingQueue = new ArrayBlockingQueue<Response>(1);
responseMap.put(request.getRequestId(), blockingQueue);
try {
return blockingQueue.take();
} catch (InterruptedException e) {
e.printStackTrace();
return null;
}
}
public <T> T newProxy(final Class<T> serviceInterface) {
Object o = Proxy.newProxyInstance(RpcClient.class.getClassLoader(), new Class[]{serviceInterface}, new InvocationHandler() {
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
return sendMessage(serviceInterface, method, args).getResponse();
}
});
return (T) o;
}
public void destroy() {
try {
this.channel.close().sync();
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
eventLoopGroup.shutdownGracefully();
}
}
}
```

## Server
```
public class RpcServer {
private static final Logger LOGGER = LoggerFactory.getLogger(RpcServer.class);
private String ip;
private int port;
private boolean started = false;
private Channel channel;
private Object serviceImpl;
private EventLoopGroup bossGroup = new NioEventLoopGroup();
private EventLoopGroup workerGroup = new NioEventLoopGroup();
public RpcServer(int port, Object serviceImpl) {
this.port = port;
this.serviceImpl = serviceImpl;
}
public void init() {
ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.childHandler(new ChannelInitializer<SocketChannel>() {
protected void initChannel(SocketChannel ch) throws Exception {
ch.pipeline()
.addLast(new LoggingHandler(LogLevel.INFO))
.addLast(new RequestCodec())
.addLast(new RpcServerHandler(serviceImpl))
.addLast(new ResponseCodec())
;
}
});
try {
ChannelFuture sync = bootstrap.bind(port).sync();
LOGGER.info("Server Started At {}", port);
started = true;
this.channel = sync.channel();
sync.channel().closeFuture().sync();
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
}
public class RpcServerHandler extends SimpleChannelInboundHandler<Request> {
private Object service;
public RpcServerHandler(Object serviceImpl) {
this.service = serviceImpl;
}
protected void channelRead0(ChannelHandlerContext ctx, Request msg) throws Exception {
String methodName = msg.getMethod();
Object[] params = msg.getParams();
Class<?>[] parameterTypes = msg.getParameterTypes();
long requestId = msg.getRequestId();
Method method = service.getClass().getDeclaredMethod(methodName, parameterTypes);
method.setAccessible(true);
Object invoke = method.invoke(service, params);
Response response = new Response();
response.setRequestId(requestId);
response.setResponse(invoke);
ctx.pipeline().writeAndFlush(response);
}
}
```

## 序列化部分
```
public class Serializer {
public static byte[] serialize(Object obj){
RuntimeSchema schema = RuntimeSchema.createFrom(obj.getClass());
LinkedBuffer buffer = LinkedBuffer.allocate(LinkedBuffer.DEFAULT_BUFFER_SIZE);
return ProtostuffIOUtil.toByteArray(obj, schema, buffer);
}
public static <T> T deserialize(Class<T> clazz, byte[] bytes) {
try {
T t = clazz.newInstance();
RuntimeSchema schema = RuntimeSchema.createFrom(clazz);
ProtostuffIOUtil.mergeFrom(bytes, t, schema);
return t;
} catch (InstantiationException e) {
e.printStackTrace();
} catch (IllegalAccessException e) {
e.printStackTrace();
}
return null;
}
}
# SimpleRpc使用示例
*需要先启动一个zookeeper作为服务注册发现中心*
```
// 服务接口
public interface IHello {
Expand Down Expand Up @@ -93,14 +357,14 @@ public class HelloImpl implements IHello {
// 客户端代码
// beanJavaConfig方式
@Bean
public CountService countService() {
RpcClientWithLB rpcClientWithLB = new RpcClientWithLB("fyes-counter");
rpcClientWithLB.setZkConn("127.0.0.1:2181");
rpcClientWithLB.init();
CountService countService = rpcClientWithLB.newProxy(CountService.class);
return countService;
}
public CountService countService() {
RpcClientWithLB rpcClientWithLB = new RpcClientWithLB("fyes-counter");
rpcClientWithLB.setZkConn("127.0.0.1:2181");
rpcClientWithLB.init();
CountService countService = rpcClientWithLB.newProxy(CountService.class);
return countService;
}
// 服务端发布
// xml配置方式
<bean class="com.github.liuzhengyang.simplerpc.ServerFactoryBean" init-method="start">
Expand Down

0 comments on commit bbd9255

Please sign in to comment.