devbox
繁依Fanyi0
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

简单的rpc实现_@rpcprovider 注解

作者：繁依Fanyi0 | 2024-04-14 23:10:35

赞

踩

@rpcprovider 注解

学习summer的github，主要用于记录方便后续查看代码

  1. package com.summer.simplerpc.annotation;
  2.  
  3. import org.springframework.stereotype.Component;
  4.  
  5. import java.lang.annotation.ElementType;
  6. import java.lang.annotation.Retention;
  7. import java.lang.annotation.RetentionPolicy;
  8. import java.lang.annotation.Target;
  9.  
  10. /**
  11.  * RPC consumer注解
  12.  *
  13.  * @author summer
  14.  * @version $Id: SimpleRpcProviderBean.java, v 0.1 2022年01月16日 11:53 AM summer Exp $
  15.  */
  16. @Retention(RetentionPolicy.RUNTIME)
  17. //注解打在属性上
  18. @Target(ElementType.FIELD)
  19. @Component
  20. public @interface SimpleRpcConsumer {
  21.     /**
  22.      * 服务版本号
  23.      * @return
  24.      */
  25.     String serviceVersion() default "1.0.0";
  26.  
  27.     /**
  28.      * 注册中心类型-默认zk
  29.      * @return
  30.      */
  31.     String registerType() default "zookeeper";
  32.  
  33.     /**
  34.      * 注册中心地址
  35.      * @return
  36.      */
  37.     String registerAddress() default "127.0.0.1:2181";
  38. }
  1. package com.summer.simplerpc.annotation;
  2.  
  3. import org.springframework.stereotype.Component;
  4.  
  5. import java.lang.annotation.ElementType;
  6. import java.lang.annotation.Retention;
  7. import java.lang.annotation.RetentionPolicy;
  8. import java.lang.annotation.Target;
  9.  
  10. /**
  11.  * RPC provider注解
  12.  *
  13.  * @author summer
  14.  * @version $Id: SimpleRpcProviderBean.java, v 0.1 2022年01月16日 11:53 AM summer Exp $
  15.  */
  16. @Retention(RetentionPolicy.RUNTIME)
  17. //注解打在类上
  18. @Target(ElementType.TYPE)
  19. @Component
  20. public @interface SimpleRpcProvider {
  21.     Class<?> serviceInterface() default Object.class;
  22.     String serviceVersion() default "1.0.0";
  23. }
  1. package com.summer.simplerpc.consumer;
  2.  
  3. import lombok.extern.slf4j.Slf4j;
  4. import org.springframework.beans.factory.config.BeanFactoryPostProcessor;
  5. import org.springframework.context.annotation.Bean;
  6. import org.springframework.context.annotation.Configuration;
  7.  
  8. /**
  9.  * rpc consumer starter
  10.  *
  11.  * @author summer
  12.  * @version $Id: SimplerConsumerAutoConfiguration.java, v 0.1 2022年01月16日 6:19 PM summer Exp $
  13.  */
  14. @Configuration
  15. @Slf4j
  16. public class SimplerConsumerAutoConfiguration {
  17.  
  18.     @Bean
  19.     public static BeanFactoryPostProcessor initRpcConsumer() throws Exception {
  20.         return new SimpleRpcConsumerPostProcessor();
  21.     }
  22. }
  1. package com.summer.simplerpc.consumer;
  2.  
  3. import com.summer.simplerpc.registry.ServiceRegistry;
  4. import com.summer.simplerpc.registry.cache.ServiceProviderCache;
  5. import com.summer.simplerpc.registry.cache.ServiceProviderLocalCache;
  6. import com.summer.simplerpc.registry.zk.ZkServiceRegistry;
  7. import lombok.extern.slf4j.Slf4j;
  8. import org.springframework.beans.factory.FactoryBean;
  9.  
  10. import java.lang.reflect.Proxy;
  11.  
  12. /**
  13.  * 生成rpc consumer代理bean的FactoryBean
  14.  *
  15.  * @author summer
  16.  * @version $Id: SimpleRpcConsumerFactoryBean.java, v 0.1 2022年01月18日 8:58 AM summer Exp $
  17.  */
  18. @Slf4j
  19. public class SimpleRpcConsumerFactoryBean implements FactoryBean {
  20.  
  21.     /**
  22.      * 调用的服务接口类
  23.      */
  24.     private Class<?> interfaceClass;
  25.  
  26.     /**
  27.      * 服务版本号
  28.      */
  29.     private String serviceVersion;
  30.  
  31.     /**
  32.      * 注册中心类型
  33.      */
  34.     private String registryType;
  35.  
  36.     /**
  37.      * 注册中心地址
  38.      */
  39.     private String registryAddress;
  40.  
  41.     /**
  42.      * 实际的bean
  43.      */
  44.     private Object object;
  45.  
  46.     /**
  47.      * init方法，通过动态代理生成bean
  48.      *
  49.      * @throws Exception
  50.      */
  51.     public void init() throws Exception {
  52.         ServiceProviderCache serviceProviderCache = new ServiceProviderLocalCache();
  53.         ServiceRegistry zkServiceRegistry = new ZkServiceRegistry(registryAddress, serviceProviderCache);
  54.  
  55.         //动态代理
  56.         this.object = Proxy.newProxyInstance(
  57.                 interfaceClass.getClassLoader(),
  58.                 new Class<?>[] {interfaceClass},
  59.                 new SimpleRpcInvokeHandler<>(this.serviceVersion, zkServiceRegistry));
  60.         log.info("SimpleRpcConsumerFactoryBean getObject {}", interfaceClass.getName());
  61.     }
  62.  
  63.     /**
  64.      * 返回创建的bean实例
  65.      *
  66.      * @return
  67.      * @throws Exception
  68.      */
  69.     @Override
  70.     public Object getObject() throws Exception {
  71.         return this.object;
  72.     }
  73.  
  74.     /**
  75.      * 创建的bean实例的类型
  76.      *
  77.      * @return
  78.      */
  79.     @Override
  80.     public Class<?> getObjectType() {
  81.         return interfaceClass;
  82.     }
  83.  
  84.     /**
  85.      * 创建的bean实例的作用域
  86.      *
  87.      * @return
  88.      */
  89.     @Override
  90.     public boolean isSingleton() {
  91.         return true;
  92.     }
  93.  
  94.     public void setInterfaceClass(Class<?> interfaceClass) {
  95.         this.interfaceClass = interfaceClass;
  96.     }
  97.  
  98.     public void setServiceVersion(String serviceVersion) {
  99.         this.serviceVersion = serviceVersion;
  100.     }
  101.  
  102.     public void setRegistryType(String registryType) {
  103.         this.registryType = registryType;
  104.     }
  105.  
  106.     public void setRegistryAddress(String registryAddress) {
  107.         this.registryAddress = registryAddress;
  108.     }
  109. }
  1. package com.summer.simplerpc.consumer;
  2.  
  3. import com.summer.simplerpc.io.RPCDecoder;
  4. import com.summer.simplerpc.io.RPCEncoder;
  5. import com.summer.simplerpc.model.SimpleRpcRequest;
  6. import com.summer.simplerpc.model.SimpleRpcResponse;
  7. import com.summer.simplerpc.registry.ServiceRegistry;
  8. import com.summer.simplerpc.registry.model.ServiceMetaConfig;
  9. import com.summer.simplerpc.util.ServiceUtils;
  10. import io.netty.bootstrap.Bootstrap;
  11. import io.netty.channel.*;
  12. import io.netty.channel.nio.NioEventLoopGroup;
  13. import io.netty.channel.socket.SocketChannel;
  14. import io.netty.channel.socket.nio.NioSocketChannel;
  15. import lombok.extern.slf4j.Slf4j;
  16.  
  17. /**
  18.  * consumer netty handler
  19.  *
  20.  * @author summer
  21.  * @version $Id: SimpleRpcConsumerNettyHandler.java, v 0.1 2022年01月19日 8:23 AM summer Exp $
  22.  */
  23. @Slf4j
  24. public class SimpleRpcConsumerNettyHandler extends SimpleChannelInboundHandler<SimpleRpcResponse> {
  25.  
  26.     /**
  27.      * 注册中心
  28.      */
  29.     private ServiceRegistry serviceRegistry;
  30.  
  31.     /**
  32.      * netty EventLoopGroup
  33.      */
  34.     private EventLoopGroup eventLoopGroup = new NioEventLoopGroup(4);
  35.  
  36.     /**
  37.      * netty channel
  38.      */
  39.     private Channel channel;
  40.  
  41.     /**
  42.      * rpc response
  43.      */
  44.     private SimpleRpcResponse rpcResponse;
  45.  
  46.     /**
  47.      * lock
  48.      */
  49.     private final Object lock = new Object();
  50.  
  51.     /**
  52.      * 构造函数
  53.      *
  54.      * @param serviceRegistry
  55.      */
  56.     public SimpleRpcConsumerNettyHandler(ServiceRegistry serviceRegistry) {
  57.         this.serviceRegistry = serviceRegistry;
  58.     }
  59.  
  60.     /**
  61.      * 发起RPC网络调用请求
  62.      *
  63.      * @param simpleRpcRequest 请求参数
  64.      * @return
  65.      */
  66.     public SimpleRpcResponse sendRpcRequest(SimpleRpcRequest simpleRpcRequest) {
  67.         try {
  68.             Bootstrap bootstrap = new Bootstrap();
  69.             bootstrap.group(eventLoopGroup)
  70.                     .channel(NioSocketChannel.class)
  71.                     .handler(new ChannelInitializer<SocketChannel>() {
  72.                         @Override
  73.                         protected void initChannel(SocketChannel socketChannel) throws Exception {
  74.                             socketChannel.pipeline()
  75.                                     .addLast(new RPCEncoder())
  76.                                     .addLast(new RPCDecoder())
  77.                                     //通过.class获取此类型的实例（https://www.cnblogs.com/penglee/p/3993033.html）
  78.                                     .addLast(SimpleRpcConsumerNettyHandler.this);
  79.                         }
  80.                     });
  81.  
  82.             String key = ServiceUtils.buildServiceKey(simpleRpcRequest.getClassName(), simpleRpcRequest.getServiceVersion());
  83.             ServiceMetaConfig serviceMetaConfig = this.serviceRegistry.discovery(key);
  84.             if (serviceMetaConfig == null) {
  85.                 log.error("sendRpcRequest fail,serviceMetaConfig not found");
  86.                 throw new Exception("serviceMetaConfig not found in registry");
  87.             }
  88.  
  89.             log.info("sendRpcRequest begin,serviceMetaConfig=" + serviceMetaConfig.toString() + ",key=" + key);
  90.             final ChannelFuture channelFuture = bootstrap.connect(serviceMetaConfig.getAddress(), serviceMetaConfig.getPort())
  91.                     .sync();
  92.             channelFuture.addListener((ChannelFutureListener)args0 -> {
  93.                if (channelFuture.isSuccess()) {
  94.                    log.info("rpc invoke success,");
  95.                } else {
  96.                    log.info("rpc invoke fail," + channelFuture.cause().getStackTrace());
  97.                    eventLoopGroup.shutdownGracefully();
  98.                }
  99.             });
  100.  
  101.             this.channel = channelFuture.channel();
  102.             this.channel.writeAndFlush(simpleRpcRequest).sync();
  103.  
  104.             synchronized (this.lock) {
  105.                 log.info("sendRpcRequest lock.wait");
  106.                 this.lock.wait();
  107.             }
  108.  
  109.             log.info("get rpc response=" + rpcResponse.toString());
  110.             return this.rpcResponse;
  111.         } catch (Exception e) {
  112.             log.error("sendRpcRequest exception,", e);
  113.             return null;
  114.         } finally {
  115.             //关闭相关连接
  116.             if (this.channel != null) {
  117.                 this.channel.close();
  118.             }
  119.             if (this.eventLoopGroup != null) {
  120.                 this.eventLoopGroup.shutdownGracefully();
  121.             }
  122.         }
  123.     }
  124.  
  125.     @Override
  126.     protected void channelRead0(ChannelHandlerContext channelHandlerContext, SimpleRpcResponse simpleRpcResponse) throws Exception {
  127.         this.rpcResponse = simpleRpcResponse;
  128.  
  129.         log.info("rpc consumer netty handler,channelRead0,rpcResponse=" + rpcResponse);
  130.  
  131.         //收到远程网络的rpc response，通知调用端
  132.         synchronized (lock) {
  133.             log.info("channelRead0 simpleRpcResponse lock.notifyAll");
  134.             lock.notifyAll();
  135.         }
  136.     }
  137. }
  1. package com.summer.simplerpc.consumer;
  2.  
  3. import com.summer.simplerpc.annotation.SimpleRpcConsumer;
  4. import lombok.extern.slf4j.Slf4j;
  5. import org.springframework.beans.BeansException;
  6. import org.springframework.beans.factory.BeanClassLoaderAware;
  7. import org.springframework.beans.factory.config.BeanDefinition;
  8. import org.springframework.beans.factory.config.BeanFactoryPostProcessor;
  9. import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
  10. import org.springframework.beans.factory.support.BeanDefinitionBuilder;
  11. import org.springframework.beans.factory.support.BeanDefinitionRegistry;
  12. import org.springframework.core.annotation.AnnotationUtils;
  13. import org.springframework.util.ClassUtils;
  14. import org.springframework.util.ReflectionUtils;
  15. import org.springframework.util.StringUtils;
  16.  
  17. import java.lang.reflect.Field;
  18. import java.util.HashMap;
  19. import java.util.Map;
  20.  
  21. /**
  22.  * rpc consumer具体处理器。负责扫描代码中含@SimpleRpcConsumer注解的属性，进行代理实现远端网络调用。
  23.  *
  24.  * @author summer
  25.  * @version $Id: SimpleRpcConsumerPostProcessor.java, v 0.1 2022年01月18日 8:28 AM summer Exp $
  26.  */
  27. @Slf4j
  28. public class SimpleRpcConsumerPostProcessor implements BeanFactoryPostProcessor, BeanClassLoaderAware {
  29.  
  30.     /**
  31.      * classloader
  32.      */
  33.     private ClassLoader classLoader;
  34.  
  35.     /**
  36.      * 保存BeanDefinition列表
  37.      */
  38.     private Map<String, BeanDefinition> beanDefinitions = new HashMap<>();
  39.  
  40.     @Override
  41.     public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) throws BeansException {
  42.         log.info("SimpleRpcConsumerPostProcessor postProcessBeanFactory begin");
  43.         //遍历bean，改些打了SimpleRpcConsumer注解的属性
  44.         for (String beanName : beanFactory.getBeanDefinitionNames()) {
  45.             BeanDefinition beanDefinition = beanFactory.getBeanDefinition(beanName);
  46.             if (StringUtils.isEmpty(beanDefinition.getBeanClassName())) {
  47.                 continue;
  48.             }
  49.             Class<?> clazz = ClassUtils.resolveClassName(beanDefinition.getBeanClassName(), this.classLoader);
  50.             ReflectionUtils.doWithFields(clazz, this::processConsumerBeanDefinition);
  51.         }
  52.  
  53.         //将BeanDefinition重新注入spring容器
  54.         BeanDefinitionRegistry beanDefinitionRegistry = (BeanDefinitionRegistry)beanFactory;
  55.         for (Map.Entry<String, BeanDefinition> entry : this.beanDefinitions.entrySet()) {
  56.             log.info("register BeanDefinition[" + entry.getKey() + "," + entry.getValue() + "]");
  57.             beanDefinitionRegistry.registerBeanDefinition(entry.getKey(), entry.getValue());
  58.         }
  59.     }
  60.  
  61.     /**
  62.      * 带有rpc消费者注解的bean定义的重处理
  63.      *
  64.      * @param field 属性
  65.      */
  66.     private void processConsumerBeanDefinition(Field field) {
  67.         SimpleRpcConsumer simpleRpcConsumer = AnnotationUtils.getAnnotation(field, SimpleRpcConsumer.class);
  68.  
  69.         //筛选出打了rpc consumer注解的属性
  70.         if (simpleRpcConsumer == null) {
  71.             return;
  72.         }
  73.  
  74.         log.info("processConsumerBeanDefinition,find a simpleRpcConsumer field:" + field.toString());
  75.         BeanDefinitionBuilder beanDefinitionBuilder = BeanDefinitionBuilder.genericBeanDefinition(SimpleRpcConsumerFactoryBean.class);
  76.         beanDefinitionBuilder.setInitMethodName("init");
  77.         beanDefinitionBuilder.addPropertyValue("interfaceClass", field.getType());
  78.         beanDefinitionBuilder.addPropertyValue("serviceVersion", simpleRpcConsumer.serviceVersion());
  79.         beanDefinitionBuilder.addPropertyValue("registryType", simpleRpcConsumer.registerType());
  80.         beanDefinitionBuilder.addPropertyValue("registryAddress", simpleRpcConsumer.registerAddress());
  81.  
  82.         BeanDefinition beanDefinition = beanDefinitionBuilder.getBeanDefinition();
  83.  
  84.         log.info("processConsumerBeanDefinition,find a simpleRpcConsumer field,result beanDefinition:" + field.toString());
  85.  
  86.         beanDefinitions.put(field.getName(), beanDefinition);
  87.     }
  88.  
  89.     /**
  90.      * 获取classloader
  91.      *
  92.      * @param classLoader
  93.      */
  94.     @Override
  95.     public void setBeanClassLoader(ClassLoader classLoader) {
  96.         this.classLoader = classLoader;
  97.     }
  98. }
  1. package com.summer.simplerpc.consumer;
  2.  
  3. import com.summer.simplerpc.model.SimpleRpcRequest;
  4. import com.summer.simplerpc.model.SimpleRpcResponse;
  5. import com.summer.simplerpc.registry.ServiceRegistry;
  6. import lombok.extern.slf4j.Slf4j;
  7.  
  8. import java.lang.reflect.InvocationHandler;
  9. import java.lang.reflect.Method;
  10. import java.util.UUID;
  11.  
  12. /**
  13.  * RPC调用动态代理handler实现
  14.  *
  15.  * @author summer
  16.  * @version $Id: SimpleRpcInvokeHandler.java, v 0.1 2022年01月18日 9:19 AM summer Exp $
  17.  */
  18. @Slf4j
  19. public class SimpleRpcInvokeHandler<T> implements InvocationHandler {
  20.  
  21.     /**
  22.      * 服务版本号
  23.      */
  24.     private String serviceVersion;
  25.  
  26.     /**
  27.      * 注册中心
  28.      */
  29.     private ServiceRegistry serviceRegistry;
  30.  
  31.     /**
  32.      * 默认构造函数
  33.      */
  34.     public SimpleRpcInvokeHandler() {
  35.  
  36.     }
  37.  
  38.     public SimpleRpcInvokeHandler(String serviceVersion, ServiceRegistry serviceRegistry) {
  39.         this.serviceVersion = serviceVersion;
  40.         this.serviceRegistry = serviceRegistry;
  41.     }
  42.  
  43.     @Override
  44.     public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
  45.         SimpleRpcRequest simpleRpcRequest = new SimpleRpcRequest();
  46.         simpleRpcRequest.setBizNO(UUID.randomUUID().toString());
  47.         simpleRpcRequest.setClassName(method.getDeclaringClass().getName());
  48.         simpleRpcRequest.setServiceVersion(this.serviceVersion);
  49.         simpleRpcRequest.setMethodName(method.getName());
  50.         simpleRpcRequest.setParamTypes(method.getParameterTypes());
  51.         simpleRpcRequest.setParamValues(args);
  52.  
  53.         log.info("begin simpleRpcRequest=" + simpleRpcRequest.toString());
  54.  
  55.         SimpleRpcConsumerNettyHandler simpleRpcConsumerNettyHandler = new SimpleRpcConsumerNettyHandler(this.serviceRegistry);
  56.         SimpleRpcResponse simpleRpcResponse = simpleRpcConsumerNettyHandler.sendRpcRequest(simpleRpcRequest);
  57.  
  58.         log.info("result simpleRpcResponse=" + simpleRpcResponse);
  59.         return simpleRpcResponse.getData();
  60.     }
  61. }
  1. package com.summer.simplerpc.io;
  2.  
  3. import com.caucho.hessian.io.HessianInput;
  4. import com.caucho.hessian.io.HessianOutput;
  5. import lombok.extern.slf4j.Slf4j;
  6.  
  7. import java.io.ByteArrayInputStream;
  8. import java.io.ByteArrayOutputStream;
  9. import java.io.IOException;
  10.  
  11. /**
  12.  * 序列化工具类（Hessian序列化协议）
  13.  *
  14.  * @author summer
  15.  * @version $Id: HessianUtils.java, v 0.1 2022年01月16日 5:00 PM summer Exp $
  16.  */
  17. @Slf4j
  18. public class HessianUtils {
  19.  
  20.     /**
  21.      * 序列化
  22.      *
  23.      * @param object
  24.      * @param <T>
  25.      * @return
  26.      */
  27.     public final static <T> byte[] serialize(T object) {
  28.         ByteArrayOutputStream byteArrayOutputStream = null;
  29.         HessianOutput hessianOutput = null;
  30.  
  31.         try {
  32.             byteArrayOutputStream = new ByteArrayOutputStream();
  33.             hessianOutput = new HessianOutput(byteArrayOutputStream);
  34.             hessianOutput.writeObject(object);
  35.             return byteArrayOutputStream.toByteArray();
  36.         } catch (IOException ioe) {
  37.             log.error("serialize io exception,object=" + object, ioe);
  38.         } finally {
  39.             if (byteArrayOutputStream != null) {
  40.                 try {
  41.                     byteArrayOutputStream.close();
  42.                 } catch (IOException ioe) {
  43.                     log.error("serialize byteArrayOutputStream close io exception,object=" + object, ioe);
  44.                 }
  45.             }
  46.  
  47.             if (hessianOutput != null) {
  48.                 try {
  49.                     hessianOutput.close();
  50.                 } catch (IOException ioe) {
  51.                     log.error("serialize hessianOutput close io exception,object=" + object, ioe);
  52.                 }
  53.             }
  54.         }
  55.  
  56.         return null;
  57.     }
  58.  
  59.     /**
  60.      * 反序列化
  61.      *
  62.      * @param bytes
  63.      * @param <T>
  64.      * @return
  65.      */
  66.     public final static <T> T deserialize(byte[] bytes) {
  67.         ByteArrayInputStream byteArrayInputStream = null;
  68.         HessianInput hessianInput = null;
  69.         try {
  70.             byteArrayInputStream = new ByteArrayInputStream(bytes);
  71.             hessianInput = new HessianInput(byteArrayInputStream);
  72.             return (T)hessianInput.readObject();
  73.         } catch (IOException ioe) {
  74.             log.error("deserialize io exception,bytes=" + bytes, ioe);
  75.         } finally {
  76.             if (byteArrayInputStream != null) {
  77.                 try {
  78.                     byteArrayInputStream.close();
  79.                 } catch (IOException ioe) {
  80.                     log.error("deserialize byteArrayOutputStream close io exception,bytes=" + bytes, ioe);
  81.                 }
  82.             }
  83.             if (hessianInput != null) {
  84.                 try {
  85.                     hessianInput.close();
  86.                 } catch (Exception ioe) {
  87.                     log.error("deserialize hessianInput close io exception,bytes=" + bytes, ioe);
  88.                 }
  89.             }
  90.         }
  91.         return null;
  92.     }
  93. }
  1. package com.summer.simplerpc.io;
  2.  
  3. import io.netty.buffer.ByteBuf;
  4. import io.netty.channel.ChannelHandlerContext;
  5. import io.netty.handler.codec.ByteToMessageDecoder;
  6. import lombok.extern.slf4j.Slf4j;
  7.  
  8. import java.util.List;
  9.  
  10. /**
  11.  * 自定义解码器，解析出RPC请求对象
  12.  *
  13.  * @author summer
  14.  * @version $Id: RPCDecoder.java, v 0.1 2022年01月16日 5:22 PM summer Exp $
  15.  */
  16. @Slf4j
  17. public class RPCDecoder extends ByteToMessageDecoder {
  18.  
  19.     @Override
  20.     public final void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
  21.         //入参校验
  22.         if (in.readableBytes() < 4) {
  23.             log.error("decode fail,input ByteBuf illegal,in.readableBytes=" + in.readableBytes());
  24.             return;
  25.         }
  26.  
  27.         in.markReaderIndex();
  28.         //读取长度内容
  29.         int dataLen = in.readInt();
  30.         //剩余可读内容小于预定长度
  31.         if (in.readableBytes() < dataLen) {
  32.             log.error("decode fail,input ByteBuf illegal,in.readableBytes {0} less than dataLen {1}", in.readableBytes(), dataLen);
  33.             return;
  34.         }
  35.  
  36.         //读取实际内容
  37.         byte[] actualDataBytes = new byte[dataLen];
  38.         in.readBytes(actualDataBytes);
  39.         //反序列化
  40.         Object dataObj = HessianUtils.deserialize(actualDataBytes);
  41.         if (dataObj == null) {
  42.             log.error("decode fail,input ByteBuf illegal,dataObj null,actualDataBytes={0}", actualDataBytes);
  43.             return;
  44.         }
  45.         out.add(dataObj);
  46.     }
  47. }
  1. package com.summer.simplerpc.io;
  2.  
  3. import io.netty.buffer.ByteBuf;
  4. import io.netty.channel.ChannelHandlerContext;
  5. import io.netty.handler.codec.MessageToByteEncoder;
  6. import lombok.extern.slf4j.Slf4j;
  7.  
  8. /**
  9.  * 自动以编码器，将rpc返回结果编码为字节流
  10.  *
  11.  * @author summer
  12.  * @version $Id: RPCEncoder.java, v 0.1 2022年01月16日 5:29 PM summer Exp $
  13.  */
  14. @Slf4j
  15. public class RPCEncoder extends MessageToByteEncoder {
  16.  
  17.     @Override
  18.     protected void encode(ChannelHandlerContext channelHandlerContext, Object o, ByteBuf byteBuf) throws Exception {
  19.         byte[] data = HessianUtils.serialize(o);
  20.         if (data == null) {
  21.             log.error("encode fail,result data null,result object=" + o);
  22.         }
  23.  
  24.         byteBuf.writeInt(data.length);
  25.         byteBuf.writeBytes(data);
  26.     }
  27. }
  1. package com.summer.simplerpc.model;
  2.  
  3. import lombok.Data;
  4. import org.springframework.boot.context.properties.ConfigurationProperties;
  5. import org.springframework.stereotype.Component;
  6.  
  7. /**
  8.  * RPC通用配置信息，提供用户自定义的功能
  9.  *
  10.  * @author summer
  11.  * @version $Id: RpcCommonProperty.java, v 0.1 2022年01月16日 6:09 PM summer Exp $
  12.  */
  13. @Data
  14. @Component
  15. @ConfigurationProperties(prefix = "summer.simplerpc")
  16. public class RpcCommonProperty {
  17.  
  18.     /**
  19.      * 服务提供方地址
  20.      */
  21.     private String serviceAddress;
  22.  
  23.     /**
  24.      * 注册中心类型
  25.      */
  26.     private String registryType;
  27.  
  28.     /**
  29.      * 注册中心地址
  30.      */
  31.     private String registryAddress;
  32. }
  1. package com.summer.simplerpc.model;
  2.  
  3. import lombok.Data;
  4.  
  5. import java.io.Serializable;
  6.  
  7. /**
  8.  * RPC请求领域模型
  9.  *
  10.  * @author summer
  11.  * @version $Id: SimpleRpcRequest.java, v 0.1 2022年01月16日 5:37 PM summer Exp $
  12.  */
  13. @Data
  14. public class SimpleRpcRequest implements Serializable {
  15.  
  16.     private static final long serialVersionUID = -6523563004185159591L;
  17.  
  18.     /**
  19.      * 业务流水号
  20.      */
  21.     private String bizNO;
  22.  
  23.     /**
  24.      * 服务类名
  25.      */
  26.     private String className;
  27.  
  28.     /**
  29.      * 服务方法名
  30.      */
  31.     private String methodName;
  32.  
  33.     /**
  34.      * 服务版本号
  35.      */
  36.     private String serviceVersion;
  37.  
  38.     /**
  39.      * 参数类型列表
  40.      */
  41.     private Class<?>[] paramTypes;
  42.  
  43.     /**
  44.      * 参数值列表
  45.      */
  46.     private Object[] paramValues;
  47. }
  1. package com.summer.simplerpc.model;
  2.  
  3. import lombok.Data;
  4.  
  5. import java.io.Serializable;
  6.  
  7. /**
  8.  * rpc处理结果
  9.  *
  10.  * @author summer
  11.  * @version $Id: SimpleRpcResponse.java, v 0.1 2022年01月16日 5:52 PM summer Exp $
  12.  */
  13. @Data
  14. public class SimpleRpcResponse implements Serializable {
  15.  
  16.     private static final long serialVersionUID = 7306531831668743451L;
  17.  
  18.     /**
  19.      * 业务流水号
  20.      */
  21.     private String bizNO;
  22.  
  23.     /**
  24.      * 错误结果提示消息
  25.      */
  26.     private String msg;
  27.  
  28.     /**
  29.      * 实际结果
  30.      */
  31.     private Object data;
  32.  
  33. }
  1. package com.summer.simplerpc.provider;
  2.  
  3. import com.google.common.util.concurrent.ThreadFactoryBuilder;
  4. import com.summer.simplerpc.annotation.SimpleRpcProvider;
  5. import com.summer.simplerpc.io.RPCDecoder;
  6. import com.summer.simplerpc.io.RPCEncoder;
  7. import com.summer.simplerpc.registry.ServiceRegistry;
  8. import com.summer.simplerpc.registry.model.ServiceMetaConfig;
  9. import com.summer.simplerpc.util.ServiceUtils;
  10. import io.netty.bootstrap.ServerBootstrap;
  11. import io.netty.channel.ChannelFuture;
  12. import io.netty.channel.ChannelInitializer;
  13. import io.netty.channel.ChannelOption;
  14. import io.netty.channel.EventLoopGroup;
  15. import io.netty.channel.nio.NioEventLoopGroup;
  16. import io.netty.channel.socket.SocketChannel;
  17. import io.netty.channel.socket.nio.NioServerSocketChannel;
  18. import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
  19. import lombok.extern.slf4j.Slf4j;
  20. import org.springframework.beans.BeansException;
  21. import org.springframework.beans.factory.InitializingBean;
  22. import org.springframework.beans.factory.config.BeanPostProcessor;
  23.  
  24. import java.util.Map;
  25. import java.util.concurrent.*;
  26.  
  27. /**
  28.  * rpc provider功能实现。
  29.  *
  30.  * 负责扫描服务provider注解bean，注册服务到注册中心，启动netty监听。
  31.  * 提供RPC请求实际处理。
  32.  *
  33.  * @author summer
  34.  * @version $Id: SimpleRpcProviderBean.java, v 0.1 2022年01月16日 12:19 PM summer Exp $
  35.  */
  36. @Slf4j
  37. public class SimpleRpcProviderBean implements InitializingBean, BeanPostProcessor {
  38.  
  39.     /**
  40.      * 地址
  41.      */
  42.     private String          address;
  43.  
  44.     /**
  45.      * 服务注册中心
  46.      */
  47.     private ServiceRegistry serviceRegistry;
  48.  
  49.     /**
  50.      * 服务提供bean的缓存map
  51.      */
  52.     private Map<String, Object> providerBeanMap = new ConcurrentHashMap<>(64);
  53.  
  54.     /**
  55.      * 处理实际rpc请求的线程池
  56.      */
  57.     private static ThreadPoolExecutor rpcThreadPoolExecutor;
  58.  
  59.     private static ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat("simplerpc-provider-pool-%d").build();
  60.  
  61.     /**
  62.      * netty相关
  63.      */
  64.     private EventLoopGroup bossGroup   = null;
  65.     private EventLoopGroup workerGroup = null;
  66.  
  67.     /**
  68.      * 构造函数
  69.      *
  70.      * @param address 地址
  71.      * @param serviceRegistry 服务注册中心
  72.      */
  73.     public SimpleRpcProviderBean(String address, ServiceRegistry serviceRegistry) {
  74.         this.address = address;
  75.         this.serviceRegistry = serviceRegistry;
  76.     }
  77.  
  78.     @Override
  79.     public void afterPropertiesSet() throws Exception {
  80.         //启动netty服务监听
  81.         new Thread(() -> {
  82.             try {
  83.                 startNettyServer();
  84.             } catch (InterruptedException e) {
  85.                 log.error("startNettyServer exception,", e);
  86.             }
  87.         }).start();
  88.     }
  89.  
  90.     /**
  91.      * 提交rpc处理任务
  92.      *
  93.      * @param task 任务
  94.      */
  95.     public static void submit(Runnable task) {
  96.         if (rpcThreadPoolExecutor == null) {
  97.             synchronized (SimpleRpcProviderBean.class) {
  98.                 if (rpcThreadPoolExecutor == null) {
  99.                     rpcThreadPoolExecutor = new ThreadPoolExecutor(100, 100,
  100.                             600L, TimeUnit.SECONDS, new ArrayBlockingQueue<>(1000),
  101.                             threadFactory);
  102.                 }
  103.             }
  104.         }
  105.         rpcThreadPoolExecutor.submit(task);
  106.     }
  107.  
  108.     /**
  109.      * 启动netty服务监听
  110.      *
  111.      * @throws InterruptedException
  112.      */
  113.     private void startNettyServer() throws InterruptedException {
  114.         if (workerGroup != null && bossGroup != null) {
  115.             return;
  116.         }
  117.  
  118.         log.info("startNettyServer begin");
  119.  
  120.         bossGroup = new NioEventLoopGroup();
  121.         workerGroup = new NioEventLoopGroup();
  122.  
  123.         ServerBootstrap serverBootstrap = new ServerBootstrap();
  124.         serverBootstrap.group(bossGroup, workerGroup)
  125.                 .channel(NioServerSocketChannel.class)
  126.                 .childHandler(new ChannelInitializer<SocketChannel>() {
  127.                     @Override
  128.                     protected void initChannel(SocketChannel socketChannel) {
  129.                         socketChannel.pipeline()
  130.                                 .addLast(new LengthFieldBasedFrameDecoder(65535,0,4,0,0))
  131.                                 .addLast(new RPCDecoder())
  132.                                 .addLast(new RPCEncoder())
  133.                                 .addLast(new SimpleRpcProviderNettyHandler(providerBeanMap))
  134.                         ;
  135.                     }
  136.                 })
  137.                 .option(ChannelOption.SO_BACKLOG, 512)
  138.                 .childOption(ChannelOption.SO_KEEPALIVE, true);
  139.  
  140.         String[] array = address.split(":");
  141.         String host = array[0];
  142.         int port = Integer.parseInt(array[1]);
  143.  
  144.         //启动服务
  145.         ChannelFuture future = serverBootstrap.bind(host, port).sync();
  146.  
  147.         log.info(String.format("startNettyServer,host=%s,port=%s", host, port));
  148.  
  149.         future.channel().closeFuture().sync();
  150.     }
  151.  
  152.     @Override
  153.     public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
  154.         return bean;
  155.     }
  156.  
  157.     @Override
  158.     public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
  159.         //获取bean上的注解
  160.         SimpleRpcProvider simpleRpcProvider = bean.getClass().getAnnotation(SimpleRpcProvider.class);
  161.         if (simpleRpcProvider == null) {
  162.             //无注解直接return原始的bean
  163.             return bean;
  164.         }
  165.  
  166.         //缓存保存
  167.         String serviceName = simpleRpcProvider.serviceInterface().getName();
  168.         String version = simpleRpcProvider.serviceVersion();
  169.         providerBeanMap.put(ServiceUtils.buildServiceKey(serviceName, version), bean);
  170.  
  171.         log.info("postProcessAfterInitialization find a simpleRpcProvider[" + serviceName + "," + version + "]");
  172.  
  173.         //将服务注册到注册中心
  174.         String[] addressArray = address.split(ServiceUtils.SPLIT_CHAR);
  175.         String host = addressArray[0];
  176.         String port = addressArray[1];
  177.  
  178.         ServiceMetaConfig serviceMetaConfig = new ServiceMetaConfig();
  179.         serviceMetaConfig.setAddress(host);
  180.         serviceMetaConfig.setName(serviceName);
  181.         serviceMetaConfig.setVersion(version);
  182.         serviceMetaConfig.setPort(Integer.parseInt(port));
  183.  
  184.         try {
  185.             serviceRegistry.register(serviceMetaConfig);
  186.             log.info("register service success,serviceMetaConfig=" + serviceMetaConfig.toString());
  187.         } catch (Exception e) {
  188.             log.error("register service fail,serviceMetaConfig=" + serviceMetaConfig.toString(), e);
  189.         }
  190.  
  191.         return bean;
  192.     }
  193. }
  1. package com.summer.simplerpc.provider;
  2.  
  3. import com.summer.simplerpc.model.SimpleRpcRequest;
  4. import com.summer.simplerpc.model.SimpleRpcResponse;
  5. import com.summer.simplerpc.util.ServiceUtils;
  6. import io.netty.channel.ChannelFutureListener;
  7. import io.netty.channel.ChannelHandlerContext;
  8. import io.netty.channel.SimpleChannelInboundHandler;
  9. import lombok.extern.slf4j.Slf4j;
  10. import org.springframework.cglib.reflect.FastClass;
  11.  
  12. import java.util.Map;
  13.  
  14. /**
  15.  * RPC核心处理逻辑handler
  16.  *
  17.  * @author summer
  18.  * @version $Id: SimpleRpcProviderNettyHandler.java, v 0.1 2022年01月16日 5:36 PM summer Exp $
  19.  */
  20. @Slf4j
  21. public class SimpleRpcProviderNettyHandler extends SimpleChannelInboundHandler<SimpleRpcRequest> {
  22.  
  23.     /**
  24.      * 提供rpc服务的实例缓存map
  25.      */
  26.     private Map<String, Object> handlerMap;
  27.  
  28.     /**
  29.      * 构造函数
  30.      *
  31.      * @param handlerMap
  32.      */
  33.     public SimpleRpcProviderNettyHandler(Map<String, Object> handlerMap) {
  34.         this.handlerMap = handlerMap;
  35.     }
  36.  
  37.     @Override
  38.     protected void channelRead0(ChannelHandlerContext channelHandlerContext, SimpleRpcRequest simpleRpcRequest) throws Exception {
  39.         SimpleRpcProviderBean.submit(() -> {
  40.             log.debug("Receive rpc request {}", simpleRpcRequest.getBizNO());
  41.             SimpleRpcResponse simpleRpcResponse = new SimpleRpcResponse();
  42.             simpleRpcResponse.setBizNO(simpleRpcRequest.getBizNO());
  43.             try {
  44.                 Object result = doHandle(simpleRpcRequest);
  45.                 simpleRpcResponse.setData(result);
  46.             } catch (Throwable throwable) {
  47.                 simpleRpcResponse.setMsg(throwable.toString());
  48.                 log.error("handle rpc request error", throwable);
  49.             }
  50.             channelHandlerContext.writeAndFlush(simpleRpcResponse).addListener(
  51.                     (ChannelFutureListener) channelFuture ->
  52.                             log.info("return response for request " + simpleRpcRequest.getBizNO() + ",simpleRpcResponse=" + simpleRpcResponse));
  53.         });
  54.     }
  55.  
  56.     /**
  57.      * 通过反射，执行实际的rpc请求
  58.      * @param simpleRpcRequest
  59.      * @return
  60.      */
  61.     private Object doHandle(SimpleRpcRequest simpleRpcRequest) throws Exception {
  62.         String key = ServiceUtils.buildServiceKey(simpleRpcRequest.getClassName(), simpleRpcRequest.getServiceVersion());
  63.         if (handlerMap == null || handlerMap.get(key) == null) {
  64.             log.error("doHandle,the provider {0} not exist,", simpleRpcRequest.getClassName(), simpleRpcRequest.getServiceVersion());
  65.             throw new RuntimeException("the provider not exist");
  66.         }
  67.  
  68.         log.info("doHandle,simpleRpcRequest=" + simpleRpcRequest.toString());
  69.  
  70.         Object provider = handlerMap.get(key);
  71.  
  72.         //通过动态代理执行实际的调用
  73.         FastClass fastClass = FastClass.create(provider.getClass());
  74.         return fastClass.invoke(fastClass.getIndex(simpleRpcRequest.getMethodName(), simpleRpcRequest.getParamTypes()),
  75.                 provider, simpleRpcRequest.getParamValues());
  76.     }
  77. }
  1. package com.summer.simplerpc.provider;
  2.  
  3. import com.summer.simplerpc.model.RpcCommonProperty;
  4. import com.summer.simplerpc.registry.ServiceRegistry;
  5. import com.summer.simplerpc.registry.cache.ServiceProviderCache;
  6. import com.summer.simplerpc.registry.cache.ServiceProviderLocalCache;
  7. import com.summer.simplerpc.registry.zk.ZkServiceRegistry;
  8. import lombok.extern.slf4j.Slf4j;
  9. import org.springframework.context.annotation.Bean;
  10. import org.springframework.context.annotation.Configuration;
  11.  
  12. /**
  13.  * rpc provider starter
  14.  *
  15.  * @author summer
  16.  * @version $Id: SimplerRpcProviderAutoConfiguration.java, v 0.1 2022年01月16日 6:19 PM summer Exp $
  17.  */
  18. @Configuration
  19. @Slf4j
  20. public class SimplerRpcProviderAutoConfiguration {
  21.  
  22.     @Bean
  23.     public SimpleRpcProviderBean initRpcProvider() throws Exception {
  24.         RpcCommonProperty rpcCommonProperty = new RpcCommonProperty();
  25.         rpcCommonProperty.setServiceAddress("127.0.0.1:50001");
  26.         rpcCommonProperty.setRegistryAddress("127.0.0.1:2181");
  27.  
  28.         log.info("===================SimplerRpcProviderAutoConfiguration init，rpcCommonProperty=" + rpcCommonProperty.toString());
  29.         ServiceProviderCache serviceProviderCache = new ServiceProviderLocalCache();
  30.         ServiceRegistry zkServiceRegistry = new ZkServiceRegistry(rpcCommonProperty.getRegistryAddress(), serviceProviderCache);
  31.  
  32.         return new SimpleRpcProviderBean(rpcCommonProperty.getServiceAddress(), zkServiceRegistry);
  33.     }
  34. }
  1. package com.summer.simplerpc.registry;
  2.  
  3. import com.summer.simplerpc.registry.model.ServiceMetaConfig;
  4.  
  5. /**
  6.  * 注册中心服务接口定义
  7.  *
  8.  * @author summer
  9.  * @version $Id: ServiceRegistry.java, v 0.1 2022年01月16日 10:56 AM summer Exp $
  10.  */
  11. public interface ServiceRegistry {
  12.  
  13.     /**
  14.      * 注册服务
  15.      *
  16.      * @param serviceMetaConfig 服务元数据配置
  17.      * @throws Exception
  18.      */
  19.     void register(ServiceMetaConfig serviceMetaConfig) throws Exception;
  20.  
  21.     /**
  22.      * 取消注册服务
  23.      *
  24.      * @param serviceMetaConfig 服务元数据配置
  25.      * @throws Exception
  26.      */
  27.     void unRegister(ServiceMetaConfig serviceMetaConfig) throws Exception;
  28.  
  29.     /**
  30.      * 服务发现
  31.      *
  32.      * @param serviceName 服务名
  33.      * @return
  34.      * @throws Exception
  35.      */
  36.     ServiceMetaConfig discovery(String serviceName) throws Exception;
  37. }
  1. package com.summer.simplerpc.registry.model;
  2.  
  3. import lombok.Data;
  4.  
  5. /**
  6.  * 服务元数据配置领域模型
  7.  *
  8.  * @author summer
  9.  * @version $Id: ServiceMetaConfig.java, v 0.1 2022年01月16日 10:58 AM summer Exp $
  10.  */
  11. @Data
  12. public class ServiceMetaConfig {
  13.  
  14.     /**
  15.      * 服务名
  16.      */
  17.     private String name;
  18.  
  19.     /**
  20.      * 服务版本
  21.      */
  22.     private String version;
  23.  
  24.     /**
  25.      * 服务地址
  26.      */
  27.     private String address;
  28.  
  29.     /**
  30.      * 服务端口
  31.      */
  32.     private Integer port;
  33. }
  1. package com.summer.simplerpc.registry.zk;
  2.  
  3. import com.summer.simplerpc.registry.cache.ServiceProviderCache;
  4. import com.summer.simplerpc.registry.model.ServiceMetaConfig;
  5. import com.summer.simplerpc.registry.ServiceRegistry;
  6. import com.summer.simplerpc.util.ServiceUtils;
  7. import org.apache.curator.framework.CuratorFramework;
  8. import org.apache.curator.framework.CuratorFrameworkFactory;
  9. import org.apache.curator.retry.ExponentialBackoffRetry;
  10. import org.apache.curator.x.discovery.*;
  11. import org.apache.curator.x.discovery.details.JsonInstanceSerializer;
  12. import org.apache.curator.x.discovery.strategies.RoundRobinStrategy;
  13.  
  14. /**
  15.  * 服务注册中心-zk实现
  16.  *
  17.  * @author summer
  18.  * @version $Id: ZkServiceRegistry.java, v 0.1 2022年01月16日 11:07 AM summer Exp $
  19.  */
  20. public class ZkServiceRegistry implements ServiceRegistry {
  21.  
  22.     /**
  23.      * zk base path
  24.      */
  25.     private final static String ZK_BASE_PATH = "/simplerpc";
  26.  
  27.     /**
  28.      * serviceProvider锁
  29.      */
  30.     private final Object lock = new Object();
  31.  
  32.     /**
  33.      * zk framework client
  34.      */
  35.     private CuratorFramework client;
  36.  
  37.     /**
  38.      * 服务发现
  39.      */
  40.     private ServiceDiscovery<ServiceMetaConfig> serviceDiscovery;
  41.  
  42.     /**
  43.      * serviceProvider缓存
  44.      */
  45.     private ServiceProviderCache serviceProviderCache;
  46.  
  47.     /**
  48.      * 构造函数
  49.      *
  50.      * @param address 地址
  51.      */
  52.     public ZkServiceRegistry(String address, ServiceProviderCache serviceProviderCache) throws Exception {
  53.         this.client = CuratorFrameworkFactory.newClient(address, new ExponentialBackoffRetry(1000, 3));
  54.         this.client.start();
  55.  
  56.         this.serviceProviderCache = serviceProviderCache;
  57.  
  58.         JsonInstanceSerializer<ServiceMetaConfig> serializer = new JsonInstanceSerializer<>(ServiceMetaConfig.class);
  59.         serviceDiscovery = ServiceDiscoveryBuilder.builder(ServiceMetaConfig.class)
  60.                 .client(client)
  61.                 .serializer(serializer)
  62.                 .basePath(ZK_BASE_PATH)
  63.                 .build();
  64.         serviceDiscovery.start();
  65.     }
  66.  
  67.     @Override
  68.     public void register(ServiceMetaConfig serviceMetaConfig) throws Exception {
  69.         ServiceInstanceBuilder<ServiceMetaConfig> serviceInstanceBuilder = ServiceInstance.builder();
  70.         ServiceInstance<ServiceMetaConfig> serviceInstance = serviceInstanceBuilder
  71.                 .name(ServiceUtils.buildServiceKey(serviceMetaConfig.getName(), serviceMetaConfig.getVersion()))
  72.                 .address(serviceMetaConfig.getAddress())
  73.                 .port(serviceMetaConfig.getPort())
  74.                 .payload(serviceMetaConfig)
  75.                 .uriSpec(new UriSpec("{scheme}://{address}:{port}"))
  76.                 .build();
  77.  
  78.         serviceDiscovery.registerService(serviceInstance);
  79.     }
  80.  
  81.     @Override
  82.     public void unRegister(ServiceMetaConfig serviceMetaConfig) throws Exception {
  83.         ServiceInstanceBuilder<ServiceMetaConfig> serviceInstanceBuilder = ServiceInstance.builder();
  84.         ServiceInstance<ServiceMetaConfig> serviceInstance = serviceInstanceBuilder
  85.                 .name(ServiceUtils.buildServiceKey(serviceMetaConfig.getName(), serviceMetaConfig.getVersion()))
  86.                 .address(serviceMetaConfig.getAddress())
  87.                 .port(serviceMetaConfig.getPort())
  88.                 .payload(serviceMetaConfig)
  89.                 .uriSpec(new UriSpec("{scheme}://{address}:{port}"))
  90.                 .build();
  91.  
  92.         serviceDiscovery.unregisterService(serviceInstance);
  93.     }
  94.  
  95.     @Override
  96.     public ServiceMetaConfig discovery(String serviceName) throws Exception {
  97.         //先读缓存
  98.         ServiceProvider<ServiceMetaConfig> serviceProvider = serviceProviderCache.queryCache(serviceName);
  99.  
  100.         //缓存miss，需要调serviceDiscovery
  101.         if (serviceProvider == null) {
  102.             synchronized (lock) {
  103.                 serviceProvider = serviceDiscovery.serviceProviderBuilder()
  104.                         .serviceName(serviceName)
  105.                         .providerStrategy(new RoundRobinStrategy<>())
  106.                         .build();
  107.                 serviceProvider.start();
  108.  
  109.                 //更新缓存
  110.                 serviceProviderCache.updateCache(serviceName, serviceProvider);
  111.             }
  112.         }
  113.  
  114.         ServiceInstance<ServiceMetaConfig> serviceInstance = serviceProvider.getInstance();
  115.         return serviceInstance != null ? serviceInstance.getPayload() : null;
  116.     }
  117. }
  1. package com.summer.simplerpc.registry.cache;
  2.  
  3. import com.summer.simplerpc.registry.model.ServiceMetaConfig;
  4. import org.apache.curator.x.discovery.ServiceProvider;
  5.  
  6. /**
  7.  *
  8.  * @author summer
  9.  * @version $Id: ServiceProviderCache.java, v 0.1 2022年01月16日 11:41 AM summer Exp $
  10.  */
  11. public interface ServiceProviderCache {
  12.  
  13.     /**
  14.      * 查询缓存
  15.      * @param serviceName
  16.      * @return
  17.      */
  18.     ServiceProvider<ServiceMetaConfig> queryCache(String serviceName);
  19.  
  20.     /**
  21.      * 更新缓存
  22.      *
  23.      * @param serviceName 服务名
  24.      * @param serviceProvider 服务provider
  25.      * @return
  26.      */
  27.     void updateCache(String serviceName, ServiceProvider<ServiceMetaConfig> serviceProvider);
  28. }
  1. package com.summer.simplerpc.registry.cache;
  2.  
  3. import com.summer.simplerpc.registry.model.ServiceMetaConfig;
  4. import org.apache.curator.x.discovery.ServiceProvider;
  5.  
  6. import java.util.Map;
  7. import java.util.concurrent.ConcurrentHashMap;
  8.  
  9. /**
  10.  * 本地缓存实现
  11.  *
  12.  * @author summer
  13.  * @version $Id: ServiceProviderLocalCache.java, v 0.1 2022年01月16日 11:43 AM summer Exp $
  14.  */
  15. public class ServiceProviderLocalCache implements ServiceProviderCache {
  16.  
  17.     /**
  18.      * 本地缓存map
  19.      */
  20.     private Map<String, ServiceProvider<ServiceMetaConfig>> serviceProviderMap = new ConcurrentHashMap<>();
  22.     @Override
  23.     public ServiceProvider<ServiceMetaConfig> queryCache(String serviceName) {
  24.         return serviceProviderMap.get(serviceName);
  25.     }
  26.  
  27.     @Override
  28.     public void updateCache(String serviceName, ServiceProvider<ServiceMetaConfig> serviceProvider) {
  29.         serviceProviderMap.put(serviceName, serviceProvider);
  30.     }
  31. }
  1. package com.summer.simplerpc.util;
  2.  
  3. /**
  4.  * 服务相关通用工具类
  5.  *
  6.  * @author summer
  7.  * @version $Id: ServiceUtils.java, v 0.1 2022年01月16日 11:28 AM summer Exp $
  8.  */
  9. public class ServiceUtils {
  10.  
  11.     /**
  12.      * 分隔符
  13.      */
  14.     public final static String SPLIT_CHAR = ":";
  15.  
  16.     /**
  17.      * 服务唯一标识key组装
  18.      *
  19.      * @param serviceName 服务名
  20.      * @param serviceVersion 服务版本
  21.      * @return
  22.      */
  23.     public final static String buildServiceKey(String serviceName, String serviceVersion) {
  24.         return String.join(SPLIT_CHAR, serviceName, serviceVersion);
  25.     }
  26. }
  1. package com.summer.simplerpc;
  2.  
  3. import org.springframework.boot.SpringApplication;
  4. import org.springframework.boot.autoconfigure.SpringBootApplication;
  5.  
  6. @SpringBootApplication
  7. public class SimplerpcApplication {
  8.  
  9.     public static void main(String[] args) {
  10.         SpringApplication.run(SimplerpcApplication.class, args);
  11.     }
  12.  
  13. }
  1. summer.simplerpc.service.address=127.0.0.1:50001
  2. summer.simplerpc.service.registry.type=zookeeper
  3. summer.simplerpc.service.registry.address=127.0.0.1:2181
  1. <?xml version="1.0" encoding="UTF-8"?>
  2. <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  3.          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
  4.     <modelVersion>4.0.0</modelVersion>
  5.     <parent>
  6.         <groupId>org.springframework.boot</groupId>
  7.         <artifactId>spring-boot-starters</artifactId>
  8.         <version>1.5.2.RELEASE</version>
  9.     </parent>
  10.     <groupId>com.summer</groupId>
  11.     <artifactId>simplerpc-starter</artifactId>
  12.     <version>0.0.1-SNAPSHOT</version>
  13.     <name>simplerpc-starter</name>
  14.     <description>Demo project for Spring Boot</description>
  15.     <properties>
  16.         <java.version>1.8</java.version>
  17.         <spring-boot.version>2.2.1.RELEASE</spring-boot.version>
  18.     </properties>
  19.     <dependencies>
  20.         <dependency>
  21.             <groupId>org.springframework.boot</groupId>
  22.             <artifactId>spring-boot-dependencies</artifactId>
  23.             <version>${spring-boot.version}</version>
  24.             <type>pom</type>
  25.             <scope>import</scope>
  26.         </dependency>
  27.  
  28.         <dependency>
  29.             <groupId>org.springframework.boot</groupId>
  30.             <artifactId>spring-boot-starter-web</artifactId>
  31.             <version>${spring-boot.version}</version>
  32.         </dependency>
  33.  
  34.         <dependency>
  35.             <groupId>org.springframework.boot</groupId>
  36.             <artifactId>spring-boot-configuration-processor</artifactId>
  37.             <version>${spring-boot.version}</version>
  38.         </dependency>
  39.  
  40.         <dependency>
  41.             <groupId>org.projectlombok</groupId>
  42.             <artifactId>lombok</artifactId>
  43.             <version>1.16.22</version>
  44.         </dependency>
  45.  
  46.         <dependency>
  47.             <groupId>org.apache.curator</groupId>
  48.             <artifactId>curator-x-discovery-server</artifactId>
  49.             <version>2.9.1</version>
  50.             <exclusions>
  51.                 <exclusion>
  52.                     <groupId>log4j</groupId>
  53.                     <artifactId>log4j</artifactId>
  54.                 </exclusion>
  55.             </exclusions>
  56.         </dependency>
  57.  
  58.         <dependency>
  59.             <groupId>io.netty</groupId>
  60.             <artifactId>netty-all</artifactId>
  61.             <version>4.1.45.Final</version>
  62.         </dependency>
  63.  
  64.         <dependency>
  65.             <groupId>cglib</groupId>
  66.             <artifactId>cglib</artifactId>
  67.             <version>3.1</version>
  68.             <exclusions>
  69.                 <exclusion>
  70.                     <groupId>org.ow2.asm</groupId>
  71.                     <artifactId>asm</artifactId>
  72.                 </exclusion>
  73.             </exclusions>
  74.         </dependency>
  75.  
  76.         <dependency>
  77.             <groupId>com.caucho</groupId>
  78.             <artifactId>hessian</artifactId>
  79.             <version>4.0.38</version>
  80.         </dependency>
  81.  
  82.         <dependency>
  83.             <groupId>org.springframework.boot</groupId>
  84.             <artifactId>spring-boot-starter-test</artifactId>
  85.             <scope>test</scope>
  86.             <exclusions>
  87.                 <exclusion>
  88.                     <groupId>net.minidev</groupId>
  89.                     <artifactId>accessors-smart</artifactId>
  90.                 </exclusion>
  91.             </exclusions>
  92.         </dependency>
  93.  
  94.     </dependencies>
  95.  
  96.     <build>
  97.         <plugins>
  98.             <plugin>
  99.                 <groupId>org.springframework.boot</groupId>
  100.                 <artifactId>spring-boot-maven-plugin</artifactId>
  101.             </plugin>
  102.         </plugins>
  103.     </build>
  104.  
  105. </project>

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/繁依Fanyi0/article/detail/424583
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号