手写RPC框架（十五）

实现SPI机制，新增负载均衡算法，大升级改造解决线程安全隐患，构建一个自己根据霍夫曼编码实现的diyZip工具类（未实现）

• SPI机制
  该机制实际上就是和我们通过注解选择对应接口实现类的功能一致，我就用他来实现简单的压缩接口，其余还是使用注解开发

  • SPI工具类

    package compress;
    
    
    import exception.RpcException;
    
    import java.util.ServiceLoader;
    
    //通过SPI机制获取对应需要的解压缩工具类  而不是用注解类获取
    public class SPICompressUtils {
        public static CompressType getUtils() throws RpcException {
            ServiceLoader loader = ServiceLoader.load(CompressType.class);
            for (CompressType compressType : loader) {
                return compressType;
            }
            throw new RpcException("没有去配置对应的解压缩方法");
        }
    }
    

  • 使用时，须要在 META-INF/services 下建立和服务的 全限定名 相同的文件，而后在该文件中写入 服务提供者 的全限定名
    

  • 实现机制

    为某个接口寻找服务实现的机制。有点类似IOC的思想，在外面实现。

• 新增负载均衡算法 一致性哈希

  • 实现一个最简单的一致性哈希 经过判断每次zookeeper都是一样的，我们进行修改不同的线程独享一个zookeeper

    • 修改代码

      private static String connectString = RpcConstants.ZOOKEEPER_ADDRESS;
      private static int sessionTimeout = RpcConstants.ZOOKEEPER_SESSION_TIMEOUT;
      private static ZooKeeper zooKeeper;
      private static ThreadLocal zooKeeperThreadLocal = ThreadLocal.withInitial(()->{
          try {
              return new ZooKeeper(connectString, sessionTimeout, new Watcher() {
                  @Override
                  public void process(WatchedEvent watchedEvent) {
      
                  }
              });
          } catch (IOException e) {
              e.printStackTrace();
              return null;
          }
      });
      

    • 实现结果
      

  • 实现哈希一致性负载均衡算法，根据zookeeper的哈希值进行分配，而服务器地址就根据对应的服务器的哈希值进行分配

    • 代码

      package loadbalance;
      
      import org.apache.zookeeper.KeeperException;
      import org.apache.zookeeper.ZooKeeper;
      
      import java.util.HashMap;
      import java.util.List;
      
      //一致性哈希 进行负载均衡计算
      public class ConsistentLoadBalance implements LoadBalance{
          @Override
          public String loadBalance(ZooKeeper zooKeeper, String path) throws InterruptedException, KeeperException {
              List children = zooKeeper.getChildren(path, false, null);
              if (children.isEmpty())
              {
                  System.out.println("当前没有服务器提供该服务 请联系工作人员");
              }
              //我这个属于的是简单版的一致性哈希
              int zkHashCode = zooKeeper.hashCode();
              return children.get(zkHashCode%children.size());
      
              //细致版一致性哈希 太麻烦了
              
          }
      }
      

• 根据上面的问题引出了线程安全问题 实现大改造 保证线程安全

  • 出现的bug
    

  • 进行线程安全改造 不能只适用于单线程

  • 一点点追到代码中去 看到可能会影响线程安全的就进行改正测验

    发现问题，在添加消费者处理器的时候，由于处理升级为了全局变量，导致每次这个处理器可能设好对应的参数，又被另一个线程初始化了，导致空指针异常。
    解决方法：使得对应的处理器，可以每次调用都新创建或者是说设置线程独有ThreadLocal，当然这要注意每次调用完要记得删除，不然会出现问题。
    

    修改代码

    package consumer.netty;
    
    
    import annotation.HeartBeatTool;
    import codec.AddCodec;
    import consumer.netty_client_handler.NettyClientHandler24;
    import heartbeat.HeartBeat;
    import io.netty.bootstrap.Bootstrap;
    import io.netty.channel.ChannelInitializer;
    import io.netty.channel.ChannelPipeline;
    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 io.netty.handler.timeout.IdleStateHandler;
    
    import java.lang.reflect.Method;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    
    
    //2.4版本主要进行大量序列化工具的集合 然后进行方便式的序列化和反序列化
    
    //实际客户端启动类  进行操作
    //不确定能返回什么 所以判断是对象
    public class NettyClient24 {
    
        //线程池 实现异步调用
        private static ExecutorService executor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
        private static HeartBeatTool heartBeatToolAnnotation = HeartBeat.class.getAnnotation(HeartBeatTool.class);
        // static NettyClientHandler24 clientHandler;//跟他没关系 因为每次都新建一个
        private static final ThreadLocal nettyClientHandlerThreadLocal = ThreadLocal.withInitial(()->new NettyClientHandler24());
    
        public static Object callMethod(String hostName, int port, Object param, Method method) throws Exception {
    
            NettyClientHandler24 clientHandler = nettyClientHandlerThreadLocal.get();
            //建立客户端监听
            Bootstrap bootstrap = new Bootstrap();
            EventLoopGroup workGroup = new NioEventLoopGroup();
    
            try {
                bootstrap.group(workGroup)
                        .channel(NioSocketChannel.class)
                        .handler(new ChannelInitializer() {
                            @Override
                            protected void initChannel(SocketChannel socketChannel) throws Exception {
                                ChannelPipeline pipeline = socketChannel.pipeline();
    
                                //加编解码器的逻辑，根据对应的注解进行编码器的添加 这里面有实现对应的逻辑 //
                                AddCodec.addCodec(pipeline,method,true);
    
                                
                                //时间和实不实现 根据注解 判断是否开启
                                //记住后面一定是要有一个处理器 用来处理触发事件
                                if (heartBeatToolAnnotation.isOpenFunction())
                                {
                                    pipeline.addLast(new IdleStateHandler(
                                            heartBeatToolAnnotation.readerIdleTimeSeconds(),
                                            heartBeatToolAnnotation.writerIdleTimeSeconds(),
                                            heartBeatToolAnnotation.allIdleTimeSeconds())
                                    );
                                }
                                pipeline.addLast(clientHandler);
                            }
                        });
    
                //进行连接
                bootstrap.connect(hostName, port).sync();
    
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            //我是有多个地方进行调用的 不能只连接一个
            // initClient(hostName,port,method);
            clientHandler.setParam(param);
            clientHandler.setMethod(method);
            //接下来这就有关系到调用 直接调用
            Object response = executor.submit(clientHandler).get();
            nettyClientHandlerThreadLocal.remove(); //一个handler不能加到两个管道中 你说是吧
            return response;
    
        }
    }
    
    

• 根据霍夫曼编码实现diyZip工具类（暂未实现 我会回来的）

  • 因为霍夫曼编解码需要一个map而多个线程肯定不能共享一个map会造成线程不安全，所以我设置了一个ThreadLocal单独一个线程一个map用完就销毁
  • 未完成 待完成