基于netty实现rpc框架-spring boot客户端
2020-04-23 16:06:17来源:博客园 阅读 ()
基于netty实现rpc框架-spring boot客户端
上篇讲了RPC服务端的实现。原理就是解析netty通道数据拿到类、方法及入参等信息,然后通过java反射机制调用本地接口返回结果。没有用到很复杂的技术。
这篇我们将客户端的实现。说白了客户端的任务很简单:一是建立socket长连接。二是封装发送服务端需要的数据包。三是处理返回结果。
demo地址
https://gitee.com/syher/grave-netty
RPC实现
同样定义注解扫描service接口。
@Retention(RetentionPolicy.RUNTIME) @Target({ElementType.TYPE}) @Documented @Import({NettyClientScannerRegistrar.class, NettyClientApplicationContextAware.class}) public @interface NettyClientScan { String[] basePackages(); Class<? extends NettyFactoryBean> factoryBean() default NettyFactoryBean.class; }
该注解用于spring boot启动类上,参数basePackages指定接口所在的包路径。
@SpringBootApplication @NettyClientScan(basePackages = { "com.braska.grave.netty.api.service" }) public class GraveNettyClientApplication { public static void main(String[] args) { SpringApplication.run(GraveNettyClientApplication.class, args); } }
NettyServerScannerRegistrar类注册bean。
public class NettyClientScannerRegistrar implements ImportBeanDefinitionRegistrar, ResourceLoaderAware { @Override public void registerBeanDefinitions(AnnotationMetadata importingClassMetadata, BeanDefinitionRegistry registry) { // spring bean注册 NettyClientInterfaceScanner scanner = new NettyClientInterfaceScanner(registry); AnnotationAttributes annoAttrs = AnnotationAttributes.fromMap(importingClassMetadata.getAnnotationAttributes(NettyClientScan.class.getName())); Class<? extends NettyFactoryBean> nettyFactoryBeanClass = annoAttrs.getClass("factoryBean"); if (!NettyFactoryBean.class.equals(nettyFactoryBeanClass)) { scanner.setNettyFactoryBean(BeanUtils.instantiateClass(nettyFactoryBeanClass)); } List<String> basePackages = new ArrayList<String>(); for (String pkg : annoAttrs.getStringArray("basePackages")) { if (StringUtils.hasText(pkg)) { basePackages.add(pkg); } } scanner.doScan(StringUtils.toStringArray(basePackages)); } }
NettyClientInterfaceScanner类使用jdk动态代理basePackages路径下的接口。
public class NettyClientInterfaceScanner extends ClassPathBeanDefinitionScanner { private NettyFactoryBean nettyFactoryBean = new NettyFactoryBean(); @Override public Set<BeanDefinitionHolder> doScan(String... basePackages) { Set<BeanDefinitionHolder> beanDefinitions = super.doScan(basePackages); if (beanDefinitions.isEmpty()) { } else { processBeanDefinitions(beanDefinitions); } return beanDefinitions; } private void processBeanDefinitions( Set<BeanDefinitionHolder> beanDefinitions) { GenericBeanDefinition definition; for (BeanDefinitionHolder holder : beanDefinitions) { definition = (GenericBeanDefinition) holder.getBeanDefinition(); // 为对象属性赋值(这一块我也还不太明白) definition.getConstructorArgumentValues().addGenericArgumentValue(definition.getBeanClassName()); // 这里的nettyFactoryBean是生成Bean实例的工厂,不是Bean本身 definition.setBeanClass(this.nettyFactoryBean.getClass()); definition.setAutowireMode(AbstractBeanDefinition.AUTOWIRE_BY_TYPE); } } }
NettyFactoryBean
public class NettyFactoryBean<T> implements FactoryBean<T> { private Class<T> nettyInterface; public NettyFactoryBean() {} public NettyFactoryBean(Class<T> nettyInterface) { this.nettyInterface = nettyInterface; } @Override public T getObject() throws Exception { // 通过jdk动态代理创建实例 return (T) Proxy.newProxyInstance(nettyInterface.getClassLoader(), new Class[]{nettyInterface}, c.getInstance()); } @Override public Class<?> getObjectType() { return this.nettyInterface; } @Override public boolean isSingleton() { return true; } }
关键来了,NettyInterfaceInvoker类负责数据包封装及发送。
public class NettyInterfaceInvoker implements InvocationHandler { private RequestSender sender; // 静态内部类做单例模式 private static class SINGLETON { private static final NettyInterfaceInvoker invoker = new NettyInterfaceInvoker(); private static NettyInterfaceInvoker setSender(RequestSender sender) { invoker.sender = sender; return invoker; } } public static NettyInterfaceInvoker getInstance() { return SINGLETON.invoker; } public static NettyInterfaceInvoker setSender(RequestSender sender) { return SINGLETON.setSender(sender); } @Override public Object invoke(Object proxy, Method method, Object[] args) throws Throwable { // 数据包封装,包含类名、方法名及参数等信息。 Request request = new Request(); request.setClassName(method.getDeclaringClass().getName()); request.setMethodName(method.getName()); request.setParameters(args); request.setParameterTypes(method.getParameterTypes()); request.setId(UUID.randomUUID().toString()); // 数据发送 Object result = sender.send(request); Class<?> returnType = method.getReturnType(); // 处理返回数据 Response response = JSON.parseObject(result.toString(), Response.class); if (response.getCode() == 1) { throw new Exception(response.getError()); } if (returnType.isPrimitive() || String.class.isAssignableFrom(returnType)) { return response.getData(); } else if (Collection.class.isAssignableFrom(returnType)) { return JSONArray.parseArray(response.getData().toString(), Object.class); } else if (Map.class.isAssignableFrom(returnType)) { return JSON.parseObject(response.getData().toString(), Map.class); } else { Object data = response.getData(); return JSONObject.parseObject(data.toString(), returnType); } } }
接着我们来看看RequestSender怎么处理数据的。
public interface RequestSender { Channel connect(SocketAddress address) throws InterruptedException; Object send(Request request) throws InterruptedException; }
RequestSender本身只是一个接口。他的实现类有:
public class NettyClientApplicationContextAware extends ChannelInitializer<SocketChannel> implements RequestSender, ApplicationContextAware, InitializingBean { private static final Logger logger = Logger.getLogger(NettyClientApplicationContextAware.class.getName()); private String remoteAddress; private Bootstrap bootstrap; private EventLoopGroup group; private NettyChannelManager manager; private NettyClientHandler handler; @Override public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { this.remoteAddress = applicationContext.getEnvironment().getProperty("remoteAddress"); this.bootstrap = new Bootstrap(); this.group = new NioEventLoopGroup(1); this.bootstrap.group(group). channel(NioSocketChannel.class). option(ChannelOption.TCP_NODELAY, true). option(ChannelOption.SO_KEEPALIVE, true). handler(this); this.manager = new NettyChannelManager(this); this.handler = new NettyClientHandler(manager, remoteAddress); } @Override public void afterPropertiesSet() throws Exception { // socket连接入口。 this.manager.refresh(Lists.newArrayList(remoteAddress)); } @Override public Object send(Request request) throws InterruptedException { Channel channel = manager.take(); if (channel != null && channel.isActive()) { SynchronousQueue<Object> queue = this.handler.sendRequest(request, channel); Object result = queue.take(); return JSONArray.toJSONString(result); } else { Response res = new Response(); res.setCode(1); res.setError("未正确连接到服务器.请检查相关配置信息!"); return JSONArray.toJSONString(res); } } @Override protected void initChannel(SocketChannel channel) throws Exception { ChannelPipeline pipeline = channel.pipeline(); pipeline.addLast(new IdleStateHandler(0, 0, 30)); pipeline.addLast(new JSONEncoder()); pipeline.addLast(new JSONDecoder()); // 管道处理器 pipeline.addLast(this.handler); } @Override public Channel connect(SocketAddress address) throws InterruptedException { ChannelFuture future = bootstrap.connect(address); // 建立长连接,提供失败重连。 future.addListener(new ConnectionListener(this.manager, this.remoteAddress)); Channel channel = future.channel();//future.sync().channel(); return channel; } public void destroy() { this.group.shutdownGracefully(); } }
NettyClientHandler类处理管道事件。与服务端不通,这个管道处理器是继承ChannelInboundHandlerAdapter类。
@ChannelHandler.Sharable public class NettyClientHandler extends ChannelInboundHandlerAdapter { private static final Logger logger = Logger.getLogger(NettyServerHandler.class.getName()); private ConcurrentHashMap<String, SynchronousQueue<Object>> queueMap = new ConcurrentHashMap<>(); private NettyChannelManager manager; private String remoteAddress; public NettyClientHandler(NettyChannelManager manager, String remoteAddress) { this.manager = manager; this.remoteAddress = remoteAddress; } @Override public void channelInactive(ChannelHandlerContext ctx) { InetSocketAddress address = (InetSocketAddress) ctx.channel().remoteAddress(); logger.info("与netty服务器断开连接." + address); ctx.channel().close(); manager.remove(ctx.channel()); // 掉线重连 final EventLoop eventLoop = ctx.channel().eventLoop(); eventLoop.schedule(() -> { manager.refresh(Lists.newArrayList(remoteAddress)); }, 1L, TimeUnit.SECONDS); } @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { // 处理服务端返回的数据 Response response = JSON.parseObject(msg.toString(), Response.class); String requestId = response.getRequestId(); SynchronousQueue<Object> queue = queueMap.get(requestId); queue.put(response); queueMap.remove(requestId); } public SynchronousQueue<Object> sendRequest(Request request, Channel channel) { // 使用阻塞队列处理客户端请求 SynchronousQueue<Object> queue = new SynchronousQueue<>(); queueMap.put(request.getId(), queue); channel.writeAndFlush(request); return queue; } public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception { logger.info("发送心跳消息..."); if (evt instanceof IdleStateEvent) { IdleStateEvent event = (IdleStateEvent) evt; if (event.state() == IdleState.ALL_IDLE) { Request request = new Request(); request.setMethodName("heartBeat"); ctx.channel().writeAndFlush(request); } } else { super.userEventTriggered(ctx, evt); } } }
这样,RPC的客户端就写好了,其中主要涉及到的关键内容就是netty实例及管道处理器、jdk动态代理、还有一个阻塞队列。
结合上篇RPC服务端。一个完整的RPC框架就搭建完了。
当然,有些地方处理的还是比较粗糙。后续有修改以git代码为准。
原文链接:https://www.cnblogs.com/braska/p/12759062.html
如有疑问请与原作者联系
标签:
版权申明:本站文章部分自网络,如有侵权,请联系:west999com@outlook.com
特别注意:本站所有转载文章言论不代表本站观点,本站所提供的摄影照片,插画,设计作品,如需使用,请与原作者联系,版权归原作者所有
- DES/3DES/AES 三种对称加密算法实现 2020-06-11
- SpringBoot + Vue + ElementUI 实现后台管理系统模板 -- 后 2020-06-10
- Spring Boot 实现定时任务的 4 种方式 2020-06-10
- JSP+SSH+Mysql+DBCP实现的租车系统 2020-06-09
- Java实现的三种字符串反转 2020-06-09
IDC资讯: 主机资讯 注册资讯 托管资讯 vps资讯 网站建设
网站运营: 建站经验 策划盈利 搜索优化 网站推广 免费资源
网络编程: Asp.Net编程 Asp编程 Php编程 Xml编程 Access Mssql Mysql 其它
服务器技术: Web服务器 Ftp服务器 Mail服务器 Dns服务器 安全防护
软件技巧: 其它软件 Word Excel Powerpoint Ghost Vista QQ空间 QQ FlashGet 迅雷
网页制作: FrontPages Dreamweaver Javascript css photoshop fireworks Flash