概述
一、dubbo简介
实现一个简易版的dubbo,首先看下dubbo是什么
dubbo是阿里开源的rpc框架,目前是apache顶级开源项目,可以用来构建微服务。社区已经到了3.0版本,生态活跃,原来是java写的,现在有go版本,支持云原生。
研究dubbo的目的
- dubbo浑身是宝,用到了zk/javassit/netty/spring/spi等技术,可以说看懂了dubbo源码,对自己是十分有帮助的。不仅能学习优秀的代码设计,理解solid原则,更能深刻理解rpc的实现原理,了解各种技术的运用。
- 当前目前有不少大公司在使用,dubbo也是经常被问的面试题,包括dubbo spi/dubbo通信流程,对于简历也是十分加分的。
- 手写dubbo,比看懂dubbo要难得多,让自己透彻理解dubbo的设计细节,内功修炼更上一层楼,以后再去看其他源码会非常快。
二、架构设计
博主在看懂dubbo源码的基础上,自己动手实现dubbo最基础的功能,包括服务注册、服务发现、服务负载均衡、网络通信、与spring boot集成等,以便更好掌握dubbo的实现原理。实现架构如下:
Api层主要是面向框架使用者,通过注解就能使用简易版dubbo。并实现了在spring boot中进行自动配置。
目前微服务都是基于spring boot搭建的,大部分公司都从spring迁移到spring boot。整个实例都是基于spring boot来构建的,基于自动化配置搞个starter,配置下yaml就能使用miniDubbo,无需配置xml。
服务发现基于zookeeper实现,所有发现的实例,暂存在注册目录中。
Zookeeper是一个树状的结构,有三种节点类型,永久节点(需要自己删除)、临时节点(有会话的概念,当zookeeper监听到会话断开时,会删除临时节点)、有序节点。在dubbo中主要用的是临时节点,用于实时感知服务实例上下线。
路径格式:
/miniDubbo/${group}/${className}/providers/${ip:port}
集群失败策略:主要有failover/failsafe/failback。
也就是当consumer调用某个provider失败时,采取的策略,failover表示失败转移,重试下一台实例;failsafe表示不重试,不抛出异常,返回null。
模拟dubbo协议,这里通过netty实现网络通信,底层使用JSON序列化,通过长度 + body的协议防止粘包。
Netty是java领域的一套高性能的nio异步网络编程框架,底层也是基于java nio,已经封装好了网络编程,实现网络通信非常方便。在dubbo中用于provider-consumer建立长连接,避免重复建立连接。
整体思路是:consumer调用接口方法时,对接口进行JDK代理,代理逻辑是通过服务发现找到一台可用实例,将请求的类名、方法名、入参进行打包、编码,通过网络请求发送给provider。在服务端拿到请求的类名、方法名、入参后,通过反射调用服务实例,并将结果的返回给consumer。
三、开发工具
需要安装如下工具。
- IDEA+JDK8+MAVEN
- zookeeper
四、一步步实现
这里给出部分代码,具体参考github:miniDubbo
4.1 客户端消费实现
4.1.1 扫描Reference注解,注入dubbo依赖
定义一个Reference注解,这个注解必须是运行时注解,且至少可用在属性上。
package com.jessin.practice.dubbo.processor; import java.lang.annotation.Documented; import java.lang.annotation.ElementType; import java.lang.annotation.Retention; import java.lang.annotation.RetentionPolicy; import java.lang.annotation.Target; /** * 消费端注解 * @Author: jessin * @Date: 19-11-25 下午9:48 */ @Target({ ElementType.CONSTRUCTOR, ElementType.METHOD, ElementType.PARAMETER, ElementType.FIELD, ElementType.ANNOTATION_TYPE}) // 必须写为runtime,否则获取不到 @Retention(RetentionPolicy.RUNTIME) @Documented public @interface Reference { String group() default ""; String version() default "1.0.0"; String timeout() default "3000"; String failStrategy() default "failover"; String retryCount() default "3"; }
在spring bean实例化后初始化前,对应的是InstantiationAwareBeanPostProcessor#postProcessProperties,这个是bean实例化后第一个扩展点,且在aware方法之前,进行依赖反射注入,并设置动态代理,代理实现该属性对应的接口。
package com.jessin.practice.dubbo.processor; import com.jessin.practice.dubbo.config.InterfaceConfig; import com.jessin.practice.dubbo.config.MiniDubboProperties; import java.lang.reflect.Field; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.BeansException; import org.springframework.beans.PropertyValues; import org.springframework.beans.factory.config.InstantiationAwareBeanPostProcessorAdapter; /** * @Author: jessin * @Date: 19-11-25 下午9:49 */ @Slf4j public class ReferenceBeanPostProcessor extends InstantiationAwareBeanPostProcessorAdapter { private MiniDubboProperties miniDubboProperties; public ReferenceBeanPostProcessor(MiniDubboProperties miniDubboProperties) { this.miniDubboProperties = miniDubboProperties; } @Override public PropertyValues postProcessProperties(PropertyValues pvs, Object bean, String beanName) throws BeansException { Field[] fields = bean.getClass().getDeclaredFields(); for (Field field : fields) { boolean isExist = field.isAnnotationPresent(Reference.class); if (isExist) { try { if (!field.getType().isInterface()) { throw new RuntimeException("dubbo依赖不是接口:" + field.getType().getName()); } Reference ref = field.getAnnotation(Reference.class); log.info("尝试注入接口代理,bean的{}属性为:{}", beanName, ref); // 私有属性,必须设置为可访问 field.setAccessible(true); field.set(bean, JdkDynamicProxy.createProxy(field.getType(), transform(ref), miniDubboProperties)); } catch (IllegalAccessException e) { log.error("设置jdk实例出错啦:{}", field); } } } return pvs; } private InterfaceConfig transform(Reference ref) { InterfaceConfig interfaceConfig = new InterfaceConfig(); interfaceConfig.setGroup(ref.group()); interfaceConfig.setVersion(ref.version()); interfaceConfig.setTimeout(ref.timeout()); interfaceConfig.setFailStrategy(ref.failStrategy()); interfaceConfig.setRetryCount(ref.retryCount()); return interfaceConfig; } }
动态代理的逻辑非常简单,基于注册目录找到可用的DubboInvoker,并发起网络请求,传输请求id、类名、入参、超时时间、版本等信息。
package com.jessin.practice.dubbo.processor; import com.alibaba.fastjson.JSONObject; import com.jessin.practice.dubbo.config.InterfaceConfig; import com.jessin.practice.dubbo.config.MiniDubboProperties; import com.jessin.practice.dubbo.invoker.FailfastClusterInvoker; import com.jessin.practice.dubbo.invoker.RpcInvocation; import com.jessin.practice.dubbo.registry.RegistryDirectory; import com.jessin.practice.dubbo.transport.Response; import java.lang.reflect.InvocationHandler; import java.lang.reflect.Method; import java.lang.reflect.Proxy; import lombok.extern.slf4j.Slf4j; /** * @Author: jessin * @Date: 19-11-25 下午9:54 */ @Slf4j public class JdkDynamicProxy<T> implements InvocationHandler { private String clazzName; private Object proxy; private RegistryDirectory registryDirectory; private FailfastClusterInvoker failfastClusterInvoker; private InterfaceConfig interfaceConfig; private MiniDubboProperties miniDubboProperties; public JdkDynamicProxy(Class<T> clazz, InterfaceConfig interfaceConfig, MiniDubboProperties miniDubboProperties) { proxy = Proxy.newProxyInstance(this.getClass().getClassLoader(), new Class[]{clazz}, this); this.clazzName = clazz.getName(); registryDirectory = new RegistryDirectory(clazzName, miniDubboProperties.getRegistry(), interfaceConfig); failfastClusterInvoker = new FailfastClusterInvoker(registryDirectory); this.interfaceConfig = interfaceConfig; } public static <T> Object createProxy(Class<T> clazz, InterfaceConfig interfaceConfig, MiniDubboProperties miniDubboProperties) { return new JdkDynamicProxy(clazz, interfaceConfig, miniDubboProperties).proxy; } /** * TODO 特殊方法不拦截。。 * @param proxy * @param method * @param args * @return * @throws Throwable */ @Override public Object invoke(Object proxy, Method method, Object[] args) throws Throwable { if ("toString".equals(method.getName())) { return this.toString(); } // todo group,attachment RpcInvocation rpcInvocation = new RpcInvocation(); rpcInvocation.setInterfaceName(clazzName); rpcInvocation.setParameterType(method.getParameterTypes()); rpcInvocation.setArgs(args); rpcInvocation.setMethodName(method.getName()); rpcInvocation.setVersion(interfaceConfig.getVersion()); Class returnType = method.getReturnType(); log.info("jdk调用:{},代理类为:{},返回类型:{}", rpcInvocation, proxy, returnType); // todo 通过接口配置决定用哪种策略 Response response = (Response)failfastClusterInvoker.invoke(rpcInvocation); if (returnType == Void.class) { return null; } JSONObject jsonObject = (JSONObject)response.getResult(); return jsonObject.toJavaObject(returnType); } }
4.1.2 使用注册目录刷新服务实例
主要是创建zk连接,订阅该类的zk路径,处理zk子节点事件,维护子节点上线和下线,并构造DubboInvoker,建立网络连接。该service接口的所有可用实例,都会存在一个注册目录中,并动态刷新。
package com.jessin.practice.dubbo.registry; import com.google.common.collect.Lists; import com.jessin.practice.dubbo.config.InterfaceConfig; import com.jessin.practice.dubbo.invoker.DubboInvoker; import java.util.ArrayList; import java.util.List; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; import java.util.stream.Collectors; import lombok.extern.slf4j.Slf4j; /** * * 注册项目录,所有的dubboInvoker都保存到这里,实现zk listener,监听zk路径变化,当zk变化时,构造DubboInvoker。 * * 每个service应该有一个,同一个zk集群同一台机器应该只建立一个共享连接 * @Author: jessin * @Date: 19-11-25 下午10:34 */ @Slf4j public class RegistryDirectory { private Map<String, DubboInvoker> ipAndPort2InvokerMap = new ConcurrentHashMap<>(); private CuratorZookeeperClient curatorZookeeperClient; private InterfaceConfig interfaceConfig; private String providerPath; /** * TODO 创建zk连接,监听zk路径创建DubboInvoker * @param path */ public RegistryDirectory(String path, String registry, InterfaceConfig interfaceConfig) { this.interfaceConfig = interfaceConfig; // 监听group/接口名/providers,有变化时通知RegistryDirectory,也就是调用notify(url, listener, urls); this.providerPath = "/miniDubbo/" + interfaceConfig.getGroup() + "/" + path + "/providers"; // TODO 创建zk连接,并创建RegistryDirectory,第一次时创建DubboInvoker // 判断zk/redis。 curatorZookeeperClient = RegistryManager.getCuratorZookeeperClient(registry); // todo 抽取subscribe方法 List<String> children = curatorZookeeperClient.addTargetChildListener(providerPath, new ChildListener() { @Override public void childChanged(String path, List<String> children) { log.info("监听到zk路径变化:{},children:{}", path, children); processChildren(children); } }); processChildren(children); } public void processChildren(List<String> children) { try { if (children == null || children.size() == 0) { // 可能是远程抖动,或者zookeeper出问题了,造成所有服务实例下线,这里还需要通过心跳检测。 log.info("监听到zk路径无子节点:{}", providerPath); children = Lists.newArrayList(); } List<String> added = children.stream() .filter(one -> !ipAndPort2InvokerMap.containsKey(one)) .collect(Collectors.toList()); List<String> finalChildren = children; List<String> deleted = ipAndPort2InvokerMap.keySet().stream() .filter(one -> !finalChildren.contains(one)) .collect(Collectors.toList()); log.info("监听到zk路径:{},子节点变化,新增zk节点:{},删除zk节点:{}", providerPath, added, deleted); added.forEach(ipAndPort -> { if (!ipAndPort2InvokerMap.containsKey(ipAndPort)) { ipAndPort2InvokerMap.put(ipAndPort, new DubboInvoker(ipAndPort, interfaceConfig)); } }); deleted.forEach(ipAndPort -> { ipAndPort2InvokerMap.get(ipAndPort).destroy(); ipAndPort2InvokerMap.remove(ipAndPort); }); } catch (Exception e) { log.error("处理zk事件出错", e); } } public List<DubboInvoker> getInvokerList() { return new ArrayList<>(ipAndPort2InvokerMap.values()); } }
4.1.3 Netty client网络通信传输,编解码器的实现
这里基于netty client 4.x api,也就是NioEventLoopGroup/NioSocketChannel等,快速开发网络连接功能,异步发送网络请求,处理各种网络请求:
package com.jessin.practice.dubbo.netty; import com.alibaba.fastjson.JSON; import io.netty.bootstrap.Bootstrap; import io.netty.channel.Channel; import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelInitializer; import io.netty.channel.ChannelOption; import io.netty.channel.EventLoopGroup; import io.netty.channel.nio.NioEventLoopGroup; import io.netty.channel.socket.SocketChannel; import io.netty.channel.socket.nio.NioSocketChannel; import lombok.extern.slf4j.Slf4j; /** * @Author: jessin * @Date: 19-11-25 下午10:17 */ @Slf4j public class NettyClient { private String ipAndPort; /** * worker可以共用 */ private EventLoopGroup worker = new NioEventLoopGroup(); private Channel socketChannel; private NettyClientHandler clientHandler = new NettyClientHandler(); public NettyClient(String ipAndPort) { this.ipAndPort = ipAndPort; connect(); } public void connect() { log.info("建立netty连接:{}", ipAndPort); Bootstrap bootstrap = new Bootstrap(); try { bootstrap.group(worker).channel(NioSocketChannel.class) .option(ChannelOption.SO_KEEPALIVE, true) .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000) .handler(new ChannelInitializer<SocketChannel>() { // TODO 注意pipeline的顺序 @Override protected void initChannel(SocketChannel ch) throws Exception { ch.pipeline() // clientHandler可以提为全局变量 .addLast(new BaseEncoder()) .addLast(new BaseDecoder()) .addLast(clientHandler); } }); // 客户端是connect String[] values = ipAndPort.split(":"); // TODO 考虑超时重连,心跳断开重连,底层转换为pipeline.connect() ChannelFuture channelFuture = bootstrap.connect(values[0], Integer.parseInt(values[1])).sync(); if (channelFuture.isSuccess()) { log.info("与服务端建立连接成功:{}", ipAndPort); } else { log.error("与服务端建立连接失败:{}", ipAndPort); } // 建立连接时保存下来,可能有需要连接多个客户端 this.socketChannel = channelFuture.channel(); } catch (Exception e) { log.error("与服务端建立连接失败:{}", ipAndPort, e); throw new RuntimeException("与服务端建立连接失败: " + ipAndPort, e); } } /** * 对外发送数据接口 * @param msg */ public void send(Object msg) { // TODO 必须用writeAndFlush才会真正发出去,同时必须序列化为字符串,才能被编码继续往下走 String jsonStr = JSON.toJSONString(msg); socketChannel.writeAndFlush(jsonStr); } public void close() { log.info("关闭访问服务的连接:{}", ipAndPort); socketChannel.close(); if (socketChannel != null && socketChannel.isActive()) { try { socketChannel.closeFuture().sync(); } catch (InterruptedException e) { e.printStackTrace(); } } worker.shutdownGracefully(); } }
编码器通过长度字段 + body实现,具体如下:
package com.jessin.practice.dubbo.netty; import io.netty.buffer.ByteBuf; import io.netty.channel.ChannelHandlerContext; import io.netty.handler.codec.MessageToByteEncoder; import org.slf4j.Logger; import org.slf4j.LoggerFactory; /** * todo protocol buf * @Author: jessin * @Date: 19-11-25 下午10:20 */ public class BaseEncoder extends MessageToByteEncoder<String> { private Logger logger = LoggerFactory.getLogger(getClass()); /** * Encode a message into a {@link ByteBuf}. This method will be called for each written message that can be handled * by this encoder. * * @param ctx the {@link ChannelHandlerContext} which this {@link MessageToByteEncoder} belongs to * @param msg the message to encode * @param out the {@link ByteBuf} into which the encoded message will be written * @throws Exception is thrown if an error occurs */ @Override protected void encode(ChannelHandlerContext ctx, String msg, ByteBuf out) throws Exception { logger.info("对消息:{}进行编码", msg); byte[] wordBytes = msg.getBytes("utf-8"); out.writeInt(wordBytes.length); out.writeBytes(wordBytes); } }
对应解码器:
package com.jessin.practice.dubbo.netty; /** * 发送的字节数 + 字节信息 * 按照这个格式进行装包和拆包,主要是会产生粘包的现象 * 也就是发送方按照abc, def, 发送 * 接收方收到的可能是a,bc,de,f,面向的是字节流,需要拆包解出命令 * * @author jessin * @create 19-11-25 下午10:20 **/ import io.netty.buffer.ByteBuf; import io.netty.channel.ChannelHandlerContext; import io.netty.handler.codec.ByteToMessageDecoder; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.List; public class BaseDecoder extends ByteToMessageDecoder { private Logger logger = LoggerFactory.getLogger(getClass()); private int totalBytes = -1; /** * Decode the from one {@link ByteBuf} to an other. This method will be called till either the input * {@link ByteBuf} has nothing to read when return from this method or till nothing was read from the input * {@link ByteBuf}. * * @param ctx the {@link ChannelHandlerContext} which this {@link ByteToMessageDecoder} belongs to * @param in the {@link ByteBuf} from which to read data * @param out the {@link List} to which decoded messages should be added * @throws Exception is thrown if an error occurs */ @Override protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception { int readableBytes = in.readableBytes(); if (totalBytes == -1 && readableBytes >= 4) { totalBytes = in.readInt(); } int secondReadableBytes = in.readableBytes(); if (totalBytes > 0 && secondReadableBytes >= totalBytes) { byte[] realData = new byte[totalBytes]; in.readBytes(realData); out.add(new String(realData, "utf-8")); totalBytes = -1; } logger.info("读取字节个数:{},剩余字节个数:{}", readableBytes, secondReadableBytes); } }
4.1.4 发起请求和接收响应
如上,发送接口通过JSON进行序列化。这里通过CountDownLatch实现客户端等待功能,当然也可以通过wait/notify实现。发送时设置自增id到请求体中,并保存到consumer本地map中,然后会调用CountDownLatch.await,并设置等待时间,进行阻塞等待结果。在收到netty 响应时,也就是netty channelRead有数据时,唤醒等待的调用线程。在channelRead中根据响应的请求id,由netty线程设置上响应结果,调用countDownLatch.countDown进行唤醒。
package com.jessin.practice.dubbo.netty; import com.alibaba.fastjson.JSON; import com.jessin.practice.dubbo.transport.DefaultFuture; import com.jessin.practice.dubbo.transport.Response; import io.netty.channel.ChannelDuplexHandler; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelPromise; import lombok.extern.slf4j.Slf4j; /** * @Author: jessin * @Date: 19-11-25 下午10:25 */ @Slf4j public class NettyClientHandler extends ChannelDuplexHandler { @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { log.info("客户端和服务端建立连接成功"); //ctx.writeAndFlush("{"id":1,"rpcInvocation":{"interfaceName":"com.jessin.practice.dubbo.service.UserService","methodName":"getUser","parameterType":[],"version":"1.0.0"}}"); } /** * 对响应进行处理 * @param ctx * @param msg * @throws Exception */ @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { log.info("收到服务端消息:" + msg); // result字段为JSONObject Response response = JSON.parseObject((String)msg, Response.class); // todo 返回list/map,带复杂key/value的是否有问题 DefaultFuture.setResponse(response); } /** * TODO 发送消息前进行拦截,oubound,只有channel.writeAndFlush()才能起作用,active里直接用ctx不起作用 * @param ctx * @param msg * @param promise * @throws Exception */ @Override public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception { log.info("收到写消息:" + msg); // 必须的!保证继续往下走,发送出去,其实就是ctx.write(msg, promise) super.write(ctx, msg, promise); // TODO promise模式,依赖地狱,以及Future回调模式(guava、java8) promise.addListener(future -> { // 监听发送回调,看是否发送成功 if (future.isSuccess()) { log.info("发送写消息:{},成功", msg); } else { log.info("发送消息失败:{}", msg); } }); } }
4.2 服务实例曝光到注册中心
4.2.1 在BeanDefinitionRegistryPostProcessor中处理@Service注解
服务端使用@Service注解:
package com.jessin.practice.dubbo.processor; import java.lang.annotation.Documented; import java.lang.annotation.ElementType; import java.lang.annotation.Retention; import java.lang.annotation.RetentionPolicy; import java.lang.annotation.Target; /** * @Author: jessin * @Date: 19-11-27 下午9:11 */ @Target({ ElementType.TYPE, ElementType.CONSTRUCTOR, ElementType.METHOD, ElementType.PARAMETER, ElementType.FIELD, ElementType.ANNOTATION_TYPE}) // 必须写为runtime,否则获取不到 @Retention(RetentionPolicy.RUNTIME) @Documented public @interface Service { String group() default ""; String version() default "1.0.0"; String timeout() default "3000"; }
需要扫描标记有@Service的类,然后注册到spring 容器,涉及到注册BeanDefinition的,一般用到的扩展点是BeanDefinitionRegistryPostProcessor#postProcessBeanDefinitionRegistry。这里为什么要自定义一个@Service注解,而不复用spring原有的@Service注解?是因为我们还要基于标记有@Service的这个bean做一些处理操作,包括对该bean曝光到本地,开启网络监听,注册到zk,这些动作需要封装到“另外一个bean”来完成,我们需要注入“另外一个bean”这个Definition,这里也就是ServiceBean,下文讲解。
package com.jessin.practice.dubbo.processor; import com.jessin.practice.dubbo.config.InterfaceConfig; import com.jessin.practice.dubbo.config.MiniDubboProperties; import java.util.Set; import org.springframework.beans.BeansException; import org.springframework.beans.factory.BeanClassLoaderAware; import org.springframework.beans.factory.config.BeanDefinition; import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; import org.springframework.beans.factory.config.RuntimeBeanReference; import org.springframework.beans.factory.support.BeanDefinitionRegistry; import org.springframework.beans.factory.support.BeanDefinitionRegistryPostProcessor; import org.springframework.beans.factory.support.RootBeanDefinition; import org.springframework.context.annotation.AnnotationBeanNameGenerator; import org.springframework.context.annotation.ClassPathBeanDefinitionScanner; import org.springframework.core.annotation.AnnotationUtils; import org.springframework.core.type.filter.AnnotationTypeFilter; import org.springframework.util.ClassUtils; /** * @Author: jessin * @Date: 19-11-27 下午9:24 */ public class ServiceBeanPostProcessor implements BeanDefinitionRegistryPostProcessor, BeanClassLoaderAware { private AnnotationBeanNameGenerator annotationBeanNameGenerator = new AnnotationBeanNameGenerator(); private MiniDubboProperties miniDubboProperties; private ClassLoader classLoader; public ServiceBeanPostProcessor(MiniDubboProperties miniDubboProperties) { this.miniDubboProperties = miniDubboProperties; } @Override public void postProcessBeanDefinitionRegistry(BeanDefinitionRegistry registry) throws BeansException { ClassPathBeanDefinitionScanner classPathBeanDefinitionScanner = new ClassPathBeanDefinitionScanner(registry, false); classPathBeanDefinitionScanner.addIncludeFilter(new AnnotationTypeFilter(Service.class)); // 将该包下的@Service注解全部扫描为bean Set<BeanDefinition> beanDefinitionSet = classPathBeanDefinitionScanner.findCandidateComponents(miniDubboProperties.getPackagePath()); for (BeanDefinition beanDefinition : beanDefinitionSet) { String beanName = annotationBeanNameGenerator.generateBeanName(beanDefinition, registry); registry.registerBeanDefinition(beanName, beanDefinition); BeanDefinition wrapper = new RootBeanDefinition(ServiceBean.class); wrapper.getPropertyValues().addPropertyValue("ref", new RuntimeBeanReference(beanName)); wrapper.getPropertyValues().addPropertyValue("miniDubboProperties", miniDubboProperties); Class beanClass = ClassUtils.resolveClassName(beanDefinition.getBeanClassName(), classLoader); Service service = AnnotationUtils.findAnnotation(beanClass, Service.class); wrapper.getPropertyValues().addPropertyValue("interfaceConfig", transform(service)); registry.registerBeanDefinition("dubbo_" + beanName, wrapper); } } @Override public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) throws BeansException { } private InterfaceConfig transform(Service ref) { InterfaceConfig interfaceConfig = new InterfaceConfig(); interfaceConfig.setGroup(ref.group()); interfaceConfig.setVersion(ref.version()); interfaceConfig.setTimeout(ref.timeout()); return interfaceConfig; } @Override public void setBeanClassLoader(ClassLoader classLoader) { this.classLoader = classLoader; } }
4.2.2 解析注解参数,注册服务到zookeeper
如上,我们在ServiceBean中,依赖标记有@Service的bean,同时将其曝光到本地,开启netty端口监听,注册到zk,具体如下,见afterPropertiesSet方法:
package com.jessin.practice.dubbo.processor; import com.jessin.practice.dubbo.config.InterfaceConfig; import com.jessin.practice.dubbo.config.MiniDubboProperties; import com.jessin.practice.dubbo.exporter.DubboExporter; import com.jessin.practice.dubbo.netty.NettyManager; import com.jessin.practice.dubbo.netty.NettyServer; import com.jessin.practice.dubbo.registry.CuratorZookeeperClient; import com.jessin.practice.dubbo.registry.RegistryManager; import com.jessin.practice.dubbo.utils.NetUtils; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.DisposableBean; import org.springframework.beans.factory.InitializingBean; /** * @Author: jessin * @Date: 19-11-27 下午10:31 */ @Slf4j public class ServiceBean implements InitializingBean, DisposableBean { private NettyServer nettyServer; /** * zk地址 */ private CuratorZookeeperClient curatorZookeeperClient; private Object ref; private MiniDubboProperties miniDubboProperties; private InterfaceConfig interfaceConfig; public MiniDubboProperties getMiniDubboProperties() { return miniDubboProperties; } public void setMiniDubboProperties(MiniDubboProperties miniDubboProperties) { this.miniDubboProperties = miniDubboProperties; } public InterfaceConfig getInterfaceConfig() { return interfaceConfig; } public void setInterfaceConfig(InterfaceConfig interfaceConfig) { this.interfaceConfig = interfaceConfig; } public Object getRef() { return ref; } public void setRef(Object ref) { this.ref = ref; } @Override public void afterPropertiesSet() throws Exception { Class[] interfaces = ref.getClass().getInterfaces(); if (interfaces.length <= 0) { throw new IllegalStateException(ref.getClass().getName() + "未实现接口"); } // todo 目前只能实现一个接口 String clazzName = interfaces[0].getName(); log.info("曝光key:{},ref:{}", clazzName, ref); // 暴露服务 todo 版本 DubboExporter.exportService(clazzName, ref); // 先开启,再注册 // 判断协议 if ("dubbo".equals(miniDubboProperties.getProtocol())) { // 开启netty server nettyServer = NettyManager.getNettyServer(miniDubboProperties.getPort()); } else { throw new RuntimeException("unknown communicate protocol:" + miniDubboProperties.getProtocol()); } // 判断什么类型的注册中心 curatorZookeeperClient = RegistryManager.getCuratorZookeeperClient(miniDubboProperties.getRegistry()); String providerPath = "/miniDubbo/" + interfaceConfig.getGroup() + "/" + clazzName + "/providers" + "/" + NetUtils.getServerIp() + ":" + miniDubboProperties.getPort(); // 注册zk,提炼register方法 curatorZookeeperClient.create(providerPath, true); } @Override public void destroy() throws Exception { curatorZookeeperClient.doClose(); nettyServer.close(); } }
4.2.3 开启netty server,接收请求
在接受到consumer请求后,解码,然后根据类名、方法名,找到对应的曝光服务,进行反射调用,将方法返回结果和请求id原样写出去,返回给客户端。具体如下:
package com.jessin.practice.dubbo.netty; import com.alibaba.fastjson.JSON; import com.alibaba.fastjson.JSONObject; import com.jessin.practice.dubbo.exception.DubboException; import com.jessin.practice.dubbo.exporter.DubboExporter; import com.jessin.practice.dubbo.invoker.RpcInvocation; import com.jessin.practice.dubbo.transport.Request; import com.jessin.practice.dubbo.transport.Response; import com.jessin.practice.dubbo.utils.ArgDeserializerUtils; import io.netty.bootstrap.ServerBootstrap; import io.netty.channel.ChannelDuplexHandler; import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelHandler; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelInitializer; import io.netty.channel.ChannelOption; import io.netty.channel.ChannelPromise; import io.netty.channel.EventLoopGroup; import io.netty.channel.nio.NioEventLoopGroup; import io.netty.channel.socket.SocketChannel; import io.netty.channel.socket.nio.NioServerSocketChannel; import java.lang.reflect.Method; import lombok.extern.slf4j.Slf4j; /** * @Author: jessin * @Date: 19-11-27 下午7:40 */ @Slf4j public class NettyServer { // todo 底层会启动2*cpu个数的NioEventLoop,轮询注册到对应的NioEventLoop运行 private EventLoopGroup boss = new NioEventLoopGroup(); private EventLoopGroup worker = new NioEventLoopGroup(); // 全局复用,是否需要考虑可共享? private ServerHandler serverHandler = new ServerHandler(); private int port; public NettyServer(int port) { ServerBootstrap serverBootstrap = new ServerBootstrap(); // boss线程池用于accept到达的请求,worker线程池对到达的请求进行读写 // child表示对到达的请求起作用,没有child表示对ServerSocketChannel起作用 // 服务端用NioServerSocketChannel ChannelFuture channelFuture; this.port = port; try { serverBootstrap.group(boss, worker) .channel(NioServerSocketChannel.class) .childOption(ChannelOption.SO_KEEPALIVE, true) // todo option最终设置到jdk sever channel上 .option(ChannelOption.SO_BACKLOG, 1024) .childHandler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel ch) throws Exception { // 对到达的请求进行读写操作,责任链模式,ChannelPipeline ch.pipeline() .addLast(new BaseDecoder()) .addLast(new BaseEncoder()) .addLast(serverHandler); } }); // todo bind时,会新建NioServerSocketChannel,并注册到NioEventLoop.selector中 // todo 底层转换为pipeline.bind(),最终调用serverSocketChannel.bind(socketAddress, 128); channelFuture = serverBootstrap.bind(port); // 下面会阻塞 channelFuture.sync(); log.info("服务器绑定端口:{}成功", port); // TODO 关闭时调用,客户端也得关闭 // channelFuture.channel().closeFuture().sync(); } catch (Exception e) { throw new RuntimeException("bind port error:" + port, e); } } /** * 允许注册到多个客户端SocketChannel中 */ @ChannelHandler.Sharable class ServerHandler extends ChannelDuplexHandler { @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { log.info("客户端:{}和服务端建立连接成功", ctx.channel().remoteAddress()); } @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { // 这里是String类型,已经解码了 Request request = JSONObject.parseObject((String)msg, Request.class); log.info("收到请求消息:{}", msg); RpcInvocation rpcInvocation = request.getRpcInvocation(); Object obj = DubboExporter.getService(rpcInvocation); if (obj == null) { throw new IllegalStateException("服务端未曝光接口:" + request); } Response response = new Response(); response.setId(request.getId()); try { log.info("开始反射调用:{}", msg); // todo 这里最好用线程池实现,不然会阻塞NioEventLoop Method method = obj.getClass().getMethod(rpcInvocation.getMethodName(), rpcInvocation.getParameterType()); Object[] originArgs = ArgDeserializerUtils.parseArgs(method, rpcInvocation.getParameterType(), rpcInvocation.getArgs()); log.info("入参:{}", originArgs); Object responseData = method.invoke(obj, originArgs); response.setResult(responseData); log.info("调用实例:{},方法:{},返回结果:{}", obj, method, response); } catch (Exception e) { log.error("调用dubbo异常:{}", rpcInvocation, e); response.setException(true); response.setResult(new DubboException("服务端调用接口异常", e)); } // TODO 通过原来客户端通道发送出去 // 这里会走编码吗?,必须写成String,或者改下Encoder ctx.writeAndFlush(JSON.toJSONString(response)); } @Override public void channelInactive(ChannelHandlerContext ctx) throws Exception { log.info("收到客户端退出的消息"); ctx.close(); } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { log.error("IO出错了...", cause); } @Override public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception { log.info("发起写请求:{}", msg); // TODO 写的一般都有这个? super.write(ctx, msg, promise); } } /** * dubbo shutdown hook */ public void close() { // TODO 这里是否有问题?? log.info("关闭端口:{}", port); boss.shutdownGracefully(); worker.shutdownGracefully(); } }
4.3 自动化配置实现
这里根据yaml中配置的开关,自动开启consumer/provider配置,需要注意的是,由于注入的@Service处理器是容器级别的后处理器,需要使用静态方法进行注入,避免过早初始化自动配置类,而且不能@autowirte 自动化属性,需要通过方法获取Environment,因为这个时候自动化属性类还没有对应的后处理器对其进行处理,拿到的属性是空的,需要自己做bind。
最后在Resource目录下,META-INF/spring.factories下,配置自动启动即可:
package com.jessin.practice.dubbo.config; import com.jessin.practice.dubbo.processor.ReferenceBeanPostProcessor; import com.jessin.practice.dubbo.processor.Service; import com.jessin.practice.dubbo.processor.ServiceBeanPostProcessor; import lombok.extern.slf4j.Slf4j; import org.springframework.boot.autoconfigure.condition.AnyNestedCondition; import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.boot.context.properties.EnableConfigurationProperties; import org.springframework.boot.context.properties.bind.Bindable; import org.springframework.boot.context.properties.bind.Binder; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Conditional; import org.springframework.context.annotation.Configuration; import org.springframework.core.env.Environment; /** * todo 自己调用自己,以及url支持,资源销毁 * @Author: jessin * @Date: 2021/10/26 9:27 PM */ @Configuration @ConditionalOnClass(Service.class) @EnableConfigurationProperties(MiniDubboProperties.class) @Slf4j public class MiniDubboAutoConfiguration { static { FastJsonConfig.config(); } /** * 由于BeanFactoryPostProcessor是提前获取的,这个时候CommonAnnotationBeanPostProcessor还没注册到beanFactory中, * serviceBeanPostProcessor注入的属性为空 */ // @Autowired // private MiniDubboProperties miniDubboProperties; // public MiniDubboAutoConfiguration() { // log.info("init MiniDubboAutoConfiguration"); // } /** * 由于这个Bean是BeanFactoryPostProcessor,提前获取时, * ConfigurationProperties的ConfigurationPropertiesBindingPostProcessor还没注入到beanFactory中, * 所以MiniDubboProperties属性没法注入 * 这里通过environment构造 * @param environment * @return */ @Bean @Conditional(ServerCondition.class) @ConditionalOnMissingBean public static ServiceBeanPostProcessor serviceBeanPostProcessor(Environment environment) { MiniDubboProperties miniDubboProperties = getMiniDubboProperties(environment); return new ServiceBeanPostProcessor(miniDubboProperties); } static class ServerCondition extends AnyNestedCondition { ServerCondition() { super(ConfigurationPhase.REGISTER_BEAN); } @ConditionalOnProperty(prefix = MiniDubboProperties.PREFIX, name = "type", havingValue = "both") static class HostProperty { } @ConditionalOnProperty(prefix = MiniDubboProperties.PREFIX, name = "type", havingValue = "server") static class JndiNameProperty { } } /** * 使用静态方法,防止造成自动化配置实例提前初始化,没有进行增强 * @param environment * @return */ @Bean @Conditional(ClientCondition.class) @ConditionalOnMissingBean public static ReferenceBeanPostProcessor referenceBeanPostProcessor(Environment environment) { MiniDubboProperties miniDubboProperties = getMiniDubboProperties(environment); return new ReferenceBeanPostProcessor(miniDubboProperties); } static class ClientCondition extends AnyNestedCondition { ClientCondition() { super(ConfigurationPhase.REGISTER_BEAN); } @ConditionalOnProperty(prefix = MiniDubboProperties.PREFIX, name = "type", havingValue = "both") static class HostProperty { } @ConditionalOnProperty(prefix = MiniDubboProperties.PREFIX, name = "type", havingValue = "client") static class JndiNameProperty { } } private static MiniDubboProperties getMiniDubboProperties(Environment environment) { MiniDubboProperties miniDubboProperties = Binder.get(environment) //首先要绑定配置器 //再将属性绑定到对象上 .bind(MiniDubboProperties.PREFIX, Bindable.of(MiniDubboProperties.class) ).get(); //再获取实例 return miniDubboProperties; } }
五、测试
5.1 编写api
这里设置了5个接口,覆盖了较多的场景,具体见:github
package com.jessin.practice.dubbo.service; import com.jessin.practice.dubbo.model.User; import com.jessin.practice.dubbo.model.UserParam; import java.util.ArrayList; import java.util.HashMap; import java.util.List; import java.util.Map; /** * @Author: jessin * @Date: 19-11-24 上午11:23 */ public interface UserService { User getUser(); User getUser(UserParam userParam); User getUser(int age); User getUser(UserParam userParam, int type, String note, int[] ages, List<Integer> list); User getUser(List<UserParam> list, ArrayList<UserParam> list2, Map<String, UserParam> userParamMap, HashMap<String, UserParam> userParamMap2); User getUser(List list, Map userParamMap); }
5.2 实现api,标记@Service
在miniDubboDemo中实现api,该工程基于spring boot实现。见miniDubboDemo
package com.example.demo.service; import com.jessin.practice.dubbo.model.User; import com.jessin.practice.dubbo.model.UserParam; import com.jessin.practice.dubbo.processor.Service; import com.jessin.practice.dubbo.service.UserService; import java.util.ArrayList; import java.util.HashMap; import java.util.List; import java.util.Map; /** * @Author: jessin * @Date: 19-11-27 下午11:33 */ @Service(group="myGroup") public class UserServiceImpl implements UserService { @Override public User getUser() { User user = new User(); user.setId(1L); user.setName("小明"); user.setAge(10); user.setNote("测试无参miniDubbo"); return user; } @Override public User getUser(UserParam userParam) { User user = new User(); user.setId(2L); user.setName("小红"); user.setAge(11); user.setNote("测试带参数UserParam"); return user; } @Override public User getUser(int age) { User user = new User(); user.setId(2L); user.setName("小红"); user.setAge(11); user.setNote("测试带参数int"); return user; } @Override public User getUser(UserParam userParam, int type, String note, int[] ages, List<Integer> list) { User user = new User(); user.setId(2L); user.setName("小红"); user.setAge(11); user.setNote("测试带多个参数"); return user; } @Override public User getUser(List<UserParam> list, ArrayList<UserParam> arrayList, Map<String, UserParam> map, HashMap<String, UserParam> hashMap) { User user = new User(); user.setId(2L); user.setName("小红"); user.setAge(11); user.setNote("测试带list/map泛型参数"); return user; } @Override public User getUser(List list, Map userParamMap) { User user = new User(); user.setId(2L); user.setName("小红"); user.setAge(11); user.setNote("测试带list/map无参"); return user; } }
5.3 编写controller接口,使用@Reference注入api依赖
也是在demo工程中,这里demo工程既作为provider,也作为consumer。
package com.example.demo.controller; import com.google.common.collect.Lists; import com.google.common.collect.Maps; import com.jessin.practice.dubbo.model.User; import com.jessin.practice.dubbo.model.UserParam; import com.jessin.practice.dubbo.processor.Reference; import com.jessin.practice.dubbo.service.UserService; import java.util.HashMap; import java.util.Map; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RequestParam; import org.springframework.web.bind.annotation.RestController; /** * TODO 泛化调用 * * @Author: jessin * @Date: 19-8-3 下午4:25 */ @RestController public class HelloController { private Logger log = LoggerFactory.getLogger(getClass()); @Reference(group = "myGroup") private UserService userService; /** * http://localhost:9999/practice/helloParam * * @return */ @RequestMapping("/hello") public User hello(UserParam userParam, @RequestParam int type) { log.info("test miniDubbo param"); if (type == 1) { return userService.getUser(); } else if (type == 2) { return userService.getUser(userParam); } else if (type == 3) { return userService.getUser(3); } else if (type == 4) { return userService.getUser(userParam, 3, "hello", new int[]{1, 2, 3}, Lists.newArrayList(3, 5, 7)); } else if (type == 5) { UserParam user2 = new UserParam(); user2.setId(1132); user2.setName("hello"); UserParam user3 = new UserParam(); user3.setId(353); user3.setName("world"); Map<String, UserParam> map = Maps.newHashMap(); map.put("key1", user2); HashMap<String, UserParam> hashMap = Maps.newHashMap(); hashMap.put("key2", user3); return userService.getUser(Lists.newArrayList(user2, user3), Lists.newArrayList(userParam), map, hashMap); } else { UserParam user2 = new UserParam(); user2.setId(1132); user2.setName("hello"); UserParam user3 = new UserParam(); user3.setId(353); user3.setName("world"); Map<String, UserParam> map = Maps.newHashMap(); map.put("key1", user2); HashMap<String, UserParam> hashMap = Maps.newHashMap(); hashMap.put("key2", user3); return userService.getUser(Lists.newArrayList(userParam, user2, user3), map); } } }
5.4 启动步骤
5.4.1 启动zk
博主的zk部署到云主机上,需保证2181端口已经启动:
ubuntu@VM-0-14-ubuntu:~$ lsof -i:2181 COMMAND PID USER FD TYPE DEVICE SIZE/OFF NODE NAME java 3904 ubuntu 27u IPv6 545176752 0t0 TCP VM-0-14-ubuntu:35306->1.15.130.58:2181 (ESTABLISHED) java 26361 ubuntu 45u IPv6 452967153 0t0 TCP *:2181 (LISTEN) java 26361 ubuntu 49u IPv6 545176753 0t0 TCP VM-0-14-ubuntu:2181->1.15.130.58:35306 (ESTABLISHED)
5.4.2 部署api和miniDubbo到本地仓库
部署api到本地仓库:api,部署miniDubbo到本地仓库:miniDubbo
mvn install
5.4.3 在demo yaml配置miniDubbo。
配置zk地址和扫描的服务端路径(只在角色为服务端生效)
mini-dubbo: package-path: "@Service注解所在包" registry: "your_zk_ip:2181"
5.4.4 编译demo为可执行jar
mvn package
5.4.5 开启两个服务端
这两个服务端都部署在云服务器上,参数说明:
server.port=9997,指定http端口
public.ip=1.15.130.58,可以指定当前注册到zk上的公网地址,如果用的是云主机的话,可以在云端部署两台服务实例,以便在本地能进行服务发现和调用。
mini-dubbo.type=server,表示角色是服务端
catalina.base=xxx,设置的是日志目录,改为你的即可。
启动第一个provider实例: 默认9999 http端口,20880 miniDubbo端口
java -Dmini-dubbo.type=server -Dcatalina.base=/home/ubuntu/program/java/miniDubboDemo/tomcat1 -Dpublic.ip=1.15.130.58 -jar target/demo-0.0.1-SNAPSHOT.jar
启动第二个provider实例: 默认9998 http端口,20881 miniDubbo端口
java -Dmini-dubbo.type=server -Dcatalina.base=/home/ubuntu/program/java/miniDubboDemo/tomcat2 -Dserver.port=9998 -Dmini-dubbo.port=20881 -Dpublic.ip=1.15.130.58 -jar target/demo-0.0.1-SNAPSHOT.jar
5.4.6 本地开启客户端进行调用
在本地启动客户端,并与两个provider建立连接:
java -Dmini-dubbo.type=client -Dcatalina.base=./tomcat3 -Dserver.port=9997 -jar target/demo-0.0.1-SNAPSHOT.jar```
通过客户端restapi触发miniDubbo调用,可以挂掉一台服务,两台服务,看看效果,并通过type调用不同的api方法
http://localhost:9997/practice/hello?type=2
得到结果:
客户端发起请求,可以看到调用的是20881这个服务:
服务端返回结果:
上面将20881销毁后,再次调用客户端发起请求,自动转移到20880这个服务:
将20880也销毁后,再次调用客户端发起请求,直接抛出no provider异常:
六、总结
6.1 已完成功能列表
- 基于zk服务注册和服务发现
- 基于netty + json序列化网络通信
- zk连接复用、netty client连接复用
- 与spring boot集成
6.2 TODO LIST
通用需求:
- spring容器销毁时,关闭占用的资源,如netty client/netty server
- 支持protobuf序列化
- 支持http协议通信
- netty超时重连,心跳断开重连,销毁dubbo invoker
- zk抖动导致所有服务实例下线优化
- 其他注册中心支持,如consul/redis
- Attach/tag实现
- 支持一个类多个版本实现
- zk重新连接时,需要重新注册关注的事件,恢复现场,对于服务端是重新注册;对于客户端是重新订阅
provider功能:
- 服务端优雅启动和优雅下线,防止流量过早进来,造成超时。在spring容器启动成功后,再注册到zk上。在spring容器销毁时,先从zk取消注册,最后再关闭客户端连接。
- 服务端请求用线程池实现,避免阻塞NioEventLoop
- 服务端支持曝光实现多个接口的一个类
consumer功能:
- netty client通过计数引用销毁资源
- 服务负载均衡算法:随机/轮询/加权
- 集群失败策略:failover/failsafe/failback
以上就是java实现简易版简易版dubbo的详细内容,更多关于dubbo的资料请关注靠谱客其它相关文章!
最后
以上就是无语枕头为你收集整理的java实现简易版简易版dubbo的全部内容,希望文章能够帮你解决java实现简易版简易版dubbo所遇到的程序开发问题。
如果觉得靠谱客网站的内容还不错,欢迎将靠谱客网站推荐给程序员好友。
发表评论 取消回复