基于netty实现rpc框架-spring boot服务端
Braska。 人气:0demo地址
https://gitee.com/syher/grave-netty
RPC介绍
首先了解一下RPC:远程过程调用。简单点说就是本地应用可以调用远程服务器的接口。那么通过什么方式调用远程接口呢?说白了RPC只是一种概念。他的调用可以基于HTTP实现,也可以基于TCP/IP实现。甚至私人定制的通讯协议。
当然,私人定制通讯协议成本过高且不具备通用性。我们不做展开讨论(其实我也展不开。。。)。那为什么不使用HTTP协议呢?受限于HTTP协议层级过高,数据传输效率不如TCP/IP。所以RPC远程调用一般采用TCP/IP实现。即调用socket方法。
RPC实现原理
1. 客户端发起远程服务调用。
2. 客户端将类信息、调用方法和入参信息通过socket通道发送给服务端。
3. 服务端解析数据包,调用本地接口。
5.将执行结果通过socket返回给客户端。
6.客户端拿到并解析返回结果。
RPC实现
java如何实现一个rpc框架,其实就是按照上面的原理再做一些详细的补充。比如通过动态代理封装客户端的数据包、通过反射机制实现服务端实现类的调用等等。
今天,我们先基于spring boot + netty 做rpc服务端的实现。
首先,做一个注解用于标识接口提供rpc调用。
@Target(ElementType.TYPE) @Retention(RetentionPolicy.RUNTIME) public @interface Service { String name() default ""; }
该注解用于提供服务的实现类上。
public interface INettyService { String getString(); }
其实现类:
package com.braska.grave.netty.server.service; @Service // 该注解为自定义rpc服务注解 public class NettyService implements INettyService { @Override public String getString() { return "welcome to use netty rpc."; } }
接着,定义一个注解用来扫描指定包名下的Service注解。
@Retention(RetentionPolicy.RUNTIME) @Target({ElementType.TYPE}) @Documented @Import({NettyServerScannerRegistrar.class, NettyServerApplicationContextAware.class}) public @interface NettyServerScan { String[] basePackages(); }
该注解用于spring boot启动类上,参数basePackages指定服务所在的包路径。
@SpringBootApplication @NettyServerScan(basePackages = { "com.braska.grave.netty.server.service" }) public class GraveNettyServerApplication { public static void main(String[] args) { SpringApplication.run(GraveNettyServerApplication.class, args); } }
NettyServerScannerRegistrar类处理服务的spring bean注册。
public class NettyServerScannerRegistrar implements BeanFactoryAware, ImportBeanDefinitionRegistrar, ResourceLoaderAware { @Override public void registerBeanDefinitions(AnnotationMetadata importingClassMetadata, BeanDefinitionRegistry registry) {
// 创建扫描器实例 NettyServerInterfaceScanner scanner = new NettyServerInterfaceScanner(registry); if (this.resourceLoader != null) { scanner.setResourceLoader(this.resourceLoader); } AnnotationAttributes annoAttrs = AnnotationAttributes.fromMap(importingClassMetadata.getAnnotationAttributes(NettyServerScan.class.getName())); List<String> basePackages = new ArrayList<String>(); for (String pkg : annoAttrs.getStringArray("basePackages")) { if (StringUtils.hasText(pkg)) { basePackages.add(pkg); } } // 只扫描指定的注解。 scanner.setAnnotationClass(Service.class); scanner.registerFilters();
// 将basePackages里面的通过@Service注解的类注册成spring bean。 scanner.doScan(StringUtils.toStringArray(basePackages)); } }
NettyServerApplicationContextAware类,暴露socket server端口。
public class NettyServerApplicationContextAware implements ApplicationContextAware, InitializingBean { private static final Logger logger = Logger.getLogger(NettyServerApplicationContextAware.class.getName());
// 存储接口与实现类的映射,其中key是接口名。value是实现类的bean。 private Map<String, Object> serviceMap = new HashMap<>();
// 服务worker。包含netty socket服务端生命周期及读写。 ServerWorker runner; @Override public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { String address = applicationContext.getEnvironment().getProperty("remoteAddress"); Map<String, Object> beans = applicationContext.getBeansWithAnnotation(Service.class); for (Object serviceBean : beans.values()) { Class<?> clazz = serviceBean.getClass(); Class<?>[] interfaces = clazz.getInterfaces(); for (Class<?> inter : interfaces) { String interfaceName = inter.getName(); serviceMap.put(interfaceName, serviceBean); } }
// 创建netty worker对象 runner = new ServerWorker(address, serviceMap); } @Override public void afterPropertiesSet() throws Exception {
// 创建netty socketServer及通道处理器 runner.open(); } }
ServerWorker类的open方法。
public class ServerWorker extends ChannelInitializer {
// socket ip:port private String remoteAddress;
// 实现类的beanMap private Map<String, Object> serviceMap;
// netty channel处理器 NettyServerHandler handler;public void open() { try { int parallel = Runtime.getRuntime().availableProcessors() * 2; ServerBootstrap bootstrap = new ServerBootstrap(); this.bossGroup = new NioEventLoopGroup(); // todo 使用线程池,提高并发能力 this.workerGroup = new NioEventLoopGroup(parallel); bootstrap.group(bossGroup, workerGroup) .channel(NioServerSocketChannel.class) .option(ChannelOption.SO_BACKLOG, 1024) .childOption(ChannelOption.SO_KEEPALIVE, true) .childOption(ChannelOption.TCP_NODELAY, true) .childHandler(this); String[] hostAndPort = this.remoteAddress.split(":"); if (hostAndPort == null || hostAndPort.length != 2) { throw new RuntimeException("remoteAddress is error."); } ChannelFuture cf = bootstrap.bind(hostAndPort[0], Integer.parseInt(hostAndPort[1])).sync(); // todo 信息写入注册中心 // registry.register(serverAddress); logger.info("netty 服务器启动.监听端口:" + hostAndPort[1]); // 等待服务端监听端口关闭 cf.channel().closeFuture().sync(); } catch (Exception e) { logger.log(Level.SEVERE, "netty server open failed.", e); this.bossGroup.shutdownGracefully(); this.workerGroup.shutdownGracefully(); } } @Override protected void initChannel(Channel channel) throws Exception { ChannelPipeline pipeline = channel.pipeline(); pipeline.addLast(new IdleStateHandler(0, 0, 60)); pipeline.addLast(new JSONEncoder()); pipeline.addLast(new JSONDecoder()); pipeline.addLast(this.handler); } }
NettyServerHandler服务端channel处理器,继承ChannelInboundHandlerAdapter。
@ChannelHandler.Sharable public class NettyServerHandler extends ChannelInboundHandlerAdapter { private Map<String, Object> serviceMap; public NettyServerHandler(Map<String, Object> serviceMap) { this.serviceMap = serviceMap; } @Override public void channelRead(ChannelHandlerContext ctx, Object msg) {
// 解析客户端发送过来的数据。包含类名、方法名、入参等信息。 Request request = JSON.parseObject(msg.toString(), Request.class); Response response = new Response(); response.setRequestId(request.getId()); try {
// 调用本地实现类 Object res = this.handler(request); response.setData(res); } catch (Exception e) { response.setCode(-1); response.setError(e.getMessage()); logger.log(Level.SEVERE, "请求调用失败", e); }
// 返回处理结果给客户端 ctx.writeAndFlush(response); } private Object handler(Request request) throws Exception { String className = request.getClassName();
// 通过className从beanMap映射中找到托管给spring的bean实现类。 Object serviceBean = serviceMap.get(className); String methodName = request.getMethodName(); Object[] parameters = request.getParameters();
// 通过反射机制调用实现类。并返回调用结果。 return MethodUtils.invokeMethod(serviceBean, methodName, parameters); } }
至此,rpc服务端的实现就完成了。
一路看下来,服务端的代码实现还是比较简单的。核心代码只有两个类:ServerWorker和NettyServerHandler。其余的都是对spring bean注册的支持。
加载全部内容