标签：信息 循环 ntp inpu exec cep 输出流 ++ 循环调用
主要内容
1. 使用zookeeper原生 API 实现分布式锁
2. 分析Curator实现分布式锁的原理
3. 实现带注册中心的RPC框架
使用zookeeper原生 API 实现分布式锁
Synchronized或者Lock
zookeeper 、redis、数据库
在使用ZooKeeper进行分布式锁的实现过程中，如何有效的避免“羊群效应( herd effect)”的出现。
有序临时节点
数据库写入log实现分布式锁
? 缺点：1.若当前线程挂掉容易出现死锁
? 2.容易出现单点故障
redis实现分布式锁
? 缺点 ：容易出现短期死锁，当前线程挂掉，但是由于redis的过期机制，所以可能多时间内死锁
zookeeper实现分布式锁
总结：相对来说zookeeper 更适合实现分布式锁，实现相对简单,可靠性相对比较高，性能比较好。
zookeeper如何实现分布式锁解决羊群效应
什么是惊群效应
zk的客户端可以在znode上添加一个watch，用来监听znode相关事件并被通知
惊群效应就是 一个特定的znode 改变的时候ZooKeper 触发了所有watches 的事件。
举个例子，如果有1000个客户端watch 一个znode的exists调用，当这个节点被创建的时候，将会有1000个通知被发送。这种由于一个被watch的znode变化，导致大量的通知需要被发送，将会导致在这个通知期间的其他操作提交的延迟。因此，只要可能，我们都强烈建议不要这么使用watch。仅仅有很少的客户端同时去watch一个znode比较好，理想的情况是只有1个。
案例分析
举个例子，有n 个clients 需要去拿到一个全局的lock.
一种简单的实现就是所有的client 去create 一个/lock znode.如果znode 已经存在，只是简单的watch 该znode 被删除。当该znode 被删除的时候，client收到通知并试图create /lock。这种策略下，就会存在上文所说的问题，每次变化都会通知所有的客户端。（惊群效应）
另外一种策略就是每个client去创建一个顺序的znode /lock/lock-.ZooKeeper 会自动添加顺序号/lock/lock-xxx.我们可以通过/lock getChildren 去拿到最小的顺序号。如果client不是最小的序列号，就再比自己小一点的znode上添加watch.
分布式锁实现
维基百科：分布式锁，是控制分布式系统之间同步访问共享资源的一种方式。在分布式系统中，常常需要协调他们的动作。如果不同的系统或是同一个系统的不同主机之间共享了一个或一组资源，那么访问这些资源的时候，往往需要互斥来防止彼此干扰来保证一致性，在这种情况下，便需要使用到分布式锁。
锁代码简单设计
基于ZooKeeper实现锁，一般都是创建EPHEMERAL_SEQUENTIAL子节点并比较序号实现的。参照Redis的分布式锁实现，也可以使用EPHEMERAL节点实现。
1 /** 2 * zk分布式实现 3 */ 4 public class DistributedLock implements Lock, Watcher { 5 /** 6 * 定义zookeeper的连接 7 */ 8 private ZooKeeper zk = null; 9 /** 10 * 定义一个根节点 11 */ 12 private String ROOT_LOCK = "/locks"; 13 /** 14 * 等待前一个锁 15 */ 16 private String WAIT_LOCK; 17 /** 18 * 表示当前锁 19 */ 20 private String CURRENT_LOCK; 21 22 23 private CountDownLatch countDownLatch; 24 25 public DistributedLock() { 26 try { 27 this.zk = new ZooKeeper("192.168.1.101:2181", 4000, this); 28 // 判断根节点是否存在 29 Stat stat = zk.exists(ROOT_LOCK, false); 30 if (null == stat) { 31 zk.create(ROOT_LOCK, "0".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); 32 } 33 } catch (IOException e) { 34 e.printStackTrace(); 35 } catch (InterruptedException e) { 36 e.printStackTrace(); 37 } catch (KeeperException e) { 38 e.printStackTrace(); 39 } 40 } 41 42 @Override 43 public void lock() { 44 // 如果获得锁成功 45 if (this.tryLock()) { 46 System.out.println(Thread.currentThread().getName() + "->" + CURRENT_LOCK + "-> 获得锁成功!"); 47 return; 48 } 49 try { 50 waitForLock(WAIT_LOCK); // 没有获得锁，则继续等待获得锁 51 } catch (KeeperException e) { 52 e.printStackTrace(); 53 } catch (InterruptedException e) { 54 e.printStackTrace(); 55 } 56 } 57 58 private boolean waitForLock(String prev) throws KeeperException, InterruptedException { 59 // 监听当前节点的上一个节点 60 Stat stat = zk.exists(prev, true); 61 if (null != stat) { 62 System.out.println(Thread.currentThread().getName() + "->等待" + ROOT_LOCK + "/" + prev + "释放"); 63 countDownLatch = new CountDownLatch(1); 64 countDownLatch.await(); 65 System.out.println(Thread.currentThread().getName() + "->获得锁成功"); 66 } 67 return true; 68 } 69 70 @Override 71 public void lockInterruptibly() throws InterruptedException { 72 } 73 74 @Override 75 public boolean tryLock() { 76 try { 77 // 创建临时有序节点 78 CURRENT_LOCK = zk.create(ROOT_LOCK + "/", "0".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL); 79 System.out.println(Thread.currentThread().getName() + "->" + 80 CURRENT_LOCK + "尝试竞争锁"); 81 // 获取根节点下的所有子节点 82 List<String> childrens = zk.getChildren(ROOT_LOCK, false); 83 // 定义一个集合对节点进行排序 84 SortedSet<String> sortedSet = new TreeSet<>(); 85 for (String children : childrens) { 86 sortedSet.add(ROOT_LOCK + "/" + children); 87 } 88 // 获取当前所有子节点中最小的值 89 String firstNode = sortedSet.first(); 90 SortedSet<String> lessThenMe = ((TreeSet<String>) sortedSet).headSet(CURRENT_LOCK); 91 // 通过当前节点和子节点中最小的节点进行比较，如果相等，表示获得锁成功 92 if (CURRENT_LOCK.equals(firstNode)) { 93 return true; 94 } 95 if (!lessThenMe.isEmpty()) { 96 // 获得比当前节点更小的最后一个节点，设置给WAIT_LOCK 97 WAIT_LOCK = lessThenMe.last(); 98 } 99 } catch (KeeperException e) { 100 e.printStackTrace(); 101 } catch (InterruptedException e) { 102 e.printStackTrace(); 103 } 104 return false; 105 } 106 107 @Override 108 public boolean tryLock(long time, TimeUnit unit) throws InterruptedException { 109 return false; 110 } 111 112 @Override 113 public void unlock() { 114 System.out.println(Thread.currentThread().getName() + "->释放锁" + CURRENT_LOCK); 115 try { 116 zk.delete(CURRENT_LOCK, -1); 117 CURRENT_LOCK = null; 118 zk.close(); 119 } catch (InterruptedException e) { 120 e.printStackTrace(); 121 } catch (KeeperException e) { 122 e.printStackTrace(); 123 } 124 } 125 126 @Override 127 public Condition newCondition() { 128 return null; 129 } 130 131 @Override 132 public void process(WatchedEvent watchedEvent) { 133 if (this.countDownLatch != null) { 134 this.countDownLatch.countDown(); 135 } 136 } 137 }
1 public static void main(String[] args) throws IOException { 2 final CountDownLatch countDownLatch = new CountDownLatch(10); 3 for (int i = 0; i < 10; i++) { 4 new Thread(() -> { 5 try { 6 countDownLatch.await(); 7 DistributedLock distributedLock = new DistributedLock(); 8 // 获得锁 9 distributedLock.lock(); 10 } catch (InterruptedException e) { 11 e.printStackTrace(); 12 } 13 }, "Thread-" + i).start(); 14 countDownLatch.countDown(); 15 } 16 System.in.read(); 17 }
1 阻塞锁和非阻塞锁
根据业务特点，普通分布式锁有两种需求：阻塞锁和非阻塞锁。
阻塞锁：多个系统同时调用同一个资源，所有请求被排队处理。已经得到分布式锁的系统，进入运行状态完成业务操作；没有得到分布式锁的线程进入阻塞状态等待，当获得相应的信号并获得分布式锁后，进入运行状态完成业务操作。
非阻塞锁：多个系统同时调用同一个资源，当某一个系统最先获取到锁，进入运行状态完成业务操作；其他没有得到分布式锁的系统，就直接返回，不做任何业务逻辑，可以给用户提示进行其他操作。
基于Zookeeper注册中心实现简易手写RPC框架
一，环境准备
1. Zookepper单点或集群环境，演示使用集群环境
2. Zookeeper的Curator客户端工具
3. 使用Maven进行项目构建
二，代码实现
服务端
1，jar包依赖
<dependency> <groupId>org.apache.zookeeper</groupId> <artifactId>zookeeper</artifactId> <version>3.4.8</version> </dependency> <dependency> <groupId>org.apache.curator</groupId> <artifactId>curator-framework</artifactId> <version>4.0.1</version> </dependency>
2、请求rpc时需要的参数
1 /** 2 * @Description: 请求rpc时需要的参数 3 * @Author: cong zhi 4 * @CreateDate: 2021/1/16 17:52 5 * @UpdateUser: cong zhi 6 * @UpdateDate: 2021/1/16 17:52 7 * @UpdateRemark: 修改内容 8 * @Version: 1.0 9 */ 10 public class RpcRequest implements Serializable { 11 12 private static final long serialVersionUID = 4186108850712886052L; 13 /** 14 * 类 15 */ 16 private String className; 17 /** 18 * 方法名 19 */ 20 private String methodName; 21 /** 22 * 参数 23 */ 24 private Object[] parameters; 25 26 private String version; 27 28 public String getVersion() { 29 return version; 30 } 31 32 public void setVersion(String version) { 33 this.version = version; 34 } 35 36 public String getClassName() { 37 return className; 38 } 39 40 public void setClassName(String className) { 41 this.className = className; 42 } 43 44 public String getMethodName() { 45 return methodName; 46 } 47 48 public void setMethodName(String methodName) { 49 this.methodName = methodName; 50 } 51 52 public Object[] getParameters() { 53 return parameters; 54 } 55 56 public void setParameters(Object[] parameters) { 57 this.parameters = parameters; 58 } 59 }
服务地址
1 /** 2 * zk 连接配置服务地址 3 * @Author: cong zhi 4 * @CreateDate: 2021/2/4 16:25 5 * @UpdateUser: cong zhi 6 * @UpdateDate: 2021/2/4 16:25 7 * @UpdateRemark: 修改内容 8 * @Version: 1.0 9 */ 10 public class ZkConfig { 11 12 public final static String CONNECTION_STR ="192.168.1.101:2181"; 13 14 public final static String ZK_REGISTER_PATH = "/registrys"; 15 }
自定义注解
1 /** 2 * 自定义注解 3 * @Author: cong zhi 4 * @CreateDate: 2021/2/7 10:51 5 * @UpdateUser: cong zhi 6 * @UpdateDate: 2021/2/7 10:51 7 * @UpdateRemark: 修改内容 8 * @Version: 1.0 9 */ 10 @Target(ElementType.TYPE) 11 @Retention(RetentionPolicy.RUNTIME) 12 public @interface RpcAnnotation { 13 14 /** 15 * 对外发布的服务接口地址 16 * @author cong zhi 17 * @date 2021/2/7 10:50 18 */ 19 Class<?> value(); 20 21 /** 22 * 版本 23 */ 24 String version() default ""; 25 }
3、注册服务接口
* 注册服务
? -- 注册服务名称为根节点下的持久化节点
? -- 注册服务对应的服务器地址为服务名称节点下的临时节点
? -- 每次服务重启时会清空临时节点（服务的分布式服务器），并重新注册临时节点
1 public class RegisterCenterImpl implements RegisterCenter { 2 3 private CuratorFramework curatorFramework; 4 // 通过静态块初始化Curator 5 { 6 curatorFramework = CuratorFrameworkFactory.builder() 7 .connectString(ZkConfig.CONNECTION_STR) 8 .sessionTimeoutMs(4000) 9 .retryPolicy(new ExponentialBackoffRetry(1000, 10)) 10 .build(); 11 curatorFramework.start(); 12 } 13 14 @Override 15 public void register(String serviceName, String serviceAddress) { 16 // 注册服务 17 String servicePath = ZkConfig.ZK_REGISTER_PATH + "/" + serviceName; 18 try { 19 // 判断 /registrys/product-service是否存在，不存在则创建 20 if (curatorFramework.checkExists().forPath(servicePath) == null) { 21 curatorFramework.create().creatingParentsIfNeeded() 22 .withMode(CreateMode.PERSISTENT).forPath(servicePath, "0".getBytes()); 23 } 24 // 获取完整路径 25 String addressPath = servicePath + "/" + serviceAddress; 26 // 创建临时节点 27 String rsNode = curatorFramework.create().withMode(CreateMode.EPHEMERAL) 28 .forPath(addressPath, "0".getBytes()); 29 System.out.println("服务注册成功" + rsNode); 30 31 } catch (Exception e) { 32 e.printStackTrace(); 33 } 34 }
4、绑定服务
* 通过构造初始化构造中心，服务发布地址已经服务绑定集合（Map）
1 * 远程调用的服务端入口，使用socket监听 2 * 3 * @Author: cong zhi 4 * @CreateDate: 2021/1/17 11:29 5 * @UpdateUser: cong zhi 6 * @UpdateDate: 2021/1/17 11:29 7 * @UpdateRemark: 修改内容 8 * @Version: 1.0 9 */ 10 public class RpcService { 11 /** 12 * 定义个线程池 13 */ 14 private static final ExecutorService executorService = Executors.newCachedThreadPool(); 15 16 /** 17 * 注册中心 18 */ 19 private RegisterCenter registerCenter; 20 /** 21 * 服务发布地址 22 */ 23 private String serviceAddress; 24 25 /** 26 * 存放服务名称和对象之间的关系 27 */ 28 Map<String, Object> handleMap = Maps.newHashMap(); 29 30 public RpcService(RegisterCenter registerCenter, String serviceAddress) { 31 this.registerCenter = registerCenter; 32 this.serviceAddress = serviceAddress; 33 }
* 对外服务接口
1 @RpcAnnotation(IUserService.class) 2 public class UserServiceImpl implements IUserService { 3 @Override 4 public String sayHello(String msg) { 5 return "Hello ,"+msg; 6 } 7 }
* 根据注解信息获取服务名称，服务版本号并绑定服务
/** * 绑定服务名称和服务对象 * @author cong zhi * @params * @param null * @date 2021/2/7 10:45 */ public void bind(Object... services) { for (Object service : services) { RpcAnnotation annotation = service.getClass().getAnnotation(RpcAnnotation.class); String serviceName = annotation.value().getName(); String version = annotation.version(); if (null != version && !version.equals("")) { serviceName = serviceName + "-" + version; } // 绑定接口名称对应的服务名 handleMap.put(serviceName, service); } }
5，发布服务端服务
? \* 根据服务地址启动服务，并注册服务到注册中心
1 /** 2 * 注册服务实例，服务注册后，其他客户端通过接口调用就可以调用服务端的实现 3 * 4 * @return 5 * @throws 6 * @author cong zhi 7 * @date 2021/1/17 12:14 8 */ 9 public void publisher() { 10 11 ServerSocket serverSocket = null; 12 try { 13 String[] addrs = serviceAddress.split(":"); 14 15 // 启动一个服务监听 16 serverSocket = new ServerSocket(Integer.parseInt(addrs[1])); 17 for (String interfaceName : handleMap.keySet()) { 18 registerCenter.register(interfaceName, serviceAddress); 19 System.out.println("注册服务成功: " + interfaceName + "->" + serviceAddress); 20 } 21 while (true) { 22 // 监听端口，是个阻塞的方法 23 Socket socket = serverSocket.accept(); 24 // 处理rpc请求，这里使用线程池来处理 25 executorService.execute(new ProcessorHandler(socket, handleMap)); 26 } 27 } catch (IOException e) { 28 e.printStackTrace(); 29 } finally { 30 if (serverSocket != null) { 31 try { 32 serverSocket.close(); 33 } catch (IOException e) { 34 e.printStackTrace(); 35 } 36 } 37 } 38 }
* main方法启动服务，通过启动两个服务实现模拟分布式，并验证负载均衡
1 public class ServiceDemo { 2 3 public static void main(String[] args) throws IOException { 4 5 IUserService userService = new UserServiceImpl(); 6 IUserService userService2 = new UserServiceImpl2(); 7 RegisterCenter register = new RegisterCenterImpl(); 8 // 初始化注册中心和服务端口信息 9 RpcService service = new RpcService(register, "127.0.0.1:8080"); 10 // 绑定服务 11 service.bind(userService,userService2); 12 // 发布并注册服务 13 service.publisher(); 14 System.in.read(); 15 } 16 }
* 服务启动执行结果
* zookeeper查询节点
客户端
1、jar包依赖
<dependency> <groupId>org.apache.zookeeper</groupId> <artifactId>zookeeper</artifactId> <version>3.4.8</version> </dependency> <dependency> <groupId>org.apache.curator</groupId> <artifactId>curator-framework</artifactId> <version>4.0.1</version> </dependency> <dependency> <groupId>org.apache.curator</groupId> <artifactId>curator-recipes</artifactId> <version>4.0.1</version> </dependency>
2，通过Zookeeper获取注册服务
* 通过构造方法初始化Curator客户端
/** * 实现类 * * @Author: cong zhi * @CreateDate: 2021/2/7 11:44 * @UpdateUser: cong zhi * @UpdateDate: 2021/2/7 11:44 * @UpdateRemark: 修改内容 * @Version: 1.0 */ public class ServiceDiscoveryImpl implements ServiceDiscovery { List<String> repos = Lists.newArrayList(); private String address; private CuratorFramework curatorFramework; public ServiceDiscoveryImpl(String address) { this.address = address; curatorFramework = CuratorFrameworkFactory.builder() .connectString(address) .sessionTimeoutMs(4000) .retryPolicy(new ExponentialBackoffRetry(1000, 10)) .build(); curatorFramework.start(); } // * 通过serviceName获取服务节点 @Override public String discovery(String serviceName) { String path = ZkConfig.ZK_REGISTER_PATH + "/" + serviceName; try { repos = curatorFramework.getChildren().forPath(path); } catch (Exception e) { throw new RuntimeException("获取子节点异常"); } // 动态发现服务节点变化 registerWatcher(path); // 负载均衡机制 LoadBanalce loadBanalce = new RandomLoadBanalce(); // 返回调用服务地址 return loadBanalce.selectHost(repos); } // * 服务监听代码 /** * 监听服务节点 * * @param path */ private void registerWatcher(final String path) { PathChildrenCache pathChildrenCache = new PathChildrenCache(curatorFramework, path, true); PathChildrenCacheListener pathChildrenCacheListener = new PathChildrenCacheListener() { @Override public void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent pathChildrenCacheEvent) throws Exception { repos = curatorFramework.getChildren().forPath(path); } }; // 将当前事件添加到当前节点上 pathChildrenCache.getListenable().addListener(pathChildrenCacheListener); try { pathChildrenCache.start(); } catch (Exception e) { throw new RuntimeException("注册PatchChild Watcher 异常:" + e); } }
* 通过随机数实现简易负载均衡代码
1 /** 2 * 负载均衡 3 * @Author: cong zhi 4 * @CreateDate: 2021/2/7 12:05 5 * @UpdateUser: cong zhi 6 * @UpdateDate: 2021/2/7 12:05 7 * @UpdateRemark: 修改内容 8 * @Version: 1.0 9 */ 10 public interface LoadBanalce { 11 12 String selectHost(List<String> repos); 13 }
* 抽象类构造模板方法
1 public abstract class AbstractLoadBanalce implements LoadBanalce { 2 3 protected abstract String doSelect(List<String> repos); 4 5 @Override 6 public String selectHost(List<String> repos) { 7 if (repos == null || repos.size() == 0) { 8 return null; 9 } 10 if (repos.size() == 1) { 11 return repos.get(0); 12 } 13 return doSelect(repos); 14 } 15 }
随机负载均衡
1 /** 2 * 随机负载均衡 3 * @Author: cong zhi 4 * @CreateDate: 2021/2/7 12:16 5 * @UpdateUser: cong zhi 6 * @UpdateDate: 2021/2/7 12:16 7 * @UpdateRemark: 修改内容 8 * @Version: 1.0 9 */ 10 public class RandomLoadBanalce extends AbstractLoadBanalce { 11 12 @Override 13 protected String doSelect(List<String> repos) { 14 int len = repos.size(); 15 Random random =new Random(); 16 17 return repos.get(random.nextInt(len)); 18 } 19 }
3，构建动态代理，发起服务请求
* 构建动态代理对象
1 /** 2 * @Description: 创建代理对象 3 * @Author: li cong zhi 4 * @CreateDate: 2021/1/16 17:52 5 * @UpdateUser: li cong zhi 6 * @UpdateDate: 2021/1/16 17:52 7 * @UpdateRemark: 修改内容 8 * @Version: 1.0 9 */ 10 public class RpcClientProxy { 11 12 private ServiceDiscovery serviceDiscovery; 13 14 15 public RpcClientProxy(ServiceDiscovery serviceDiscovery) { 16 this.serviceDiscovery = serviceDiscovery; 17 } 18 19 /*public <T> T clientProxy(final Class<T> interfaceCls, 20 final String host, final int port) { 21 // 使用动态代理 22 return (T) Proxy.newProxyInstance(interfaceCls.getClassLoader(), new Class[]{ 23 interfaceCls}, new RemoteInvokeHandler(host, port)); 24 }*/ 25 26 public <T> T clientProxy(final Class<T> interfaceCls, String version) { 27 // 使用动态代理 28 return (T) Proxy.newProxyInstance(interfaceCls.getClassLoader(), new Class[]{ 29 interfaceCls}, new RemoteInvokeHandler(serviceDiscovery, version)); 30 } 31 }
4，发送服务请求到客户端
* 通过负载均衡获取的服务地址构建套接字
1 /** 2 * @Description: socket传输 3 * @Author: cong zhi 4 * @CreateDate: 2021/1/16 18:09 5 * @UpdateUser: cong zhi 6 * @UpdateDate: 2021/1/16 18:09 7 * @UpdateRemark: 修改内容 8 * @Version: 1.0 9 */ 10 public class TcpTransport { 11 12 private String serviceAddress; 13 14 15 public TcpTransport(String serviceAddress) { 16 this.serviceAddress = serviceAddress; 17 } 18 19 public Socket newSocket() { 20 System.out.println("准备创建Socket连接"); 21 Socket socket; 22 try { 23 String[] arrs = serviceAddress.split(":"); 24 socket = new Socket(arrs[0], Integer.parseInt(arrs[1])); 25 return socket; 26 } catch (IOException e) { 27 throw new RuntimeException("Socket连接建立失败！"); 28 } 29 } 30 31 // * 通过套接字输出流构建对象输出流 32 // * 通过对象输出流写出服务请求参数 33 public Object send(RpcRequest request) { 34 Socket socket = null; 35 try { 36 socket = newSocket(); 37 ObjectOutputStream outputStream = new ObjectOutputStream(socket.getOutputStream()); 38 // 通过 ObjectOutputStream 将当前request传给服务端 39 outputStream.writeObject(request); 40 outputStream.flush(); 41 42 ObjectInputStream inputStream = new ObjectInputStream(socket.getInputStream()); 43 Object result = inputStream.readObject(); 44 inputStream.close(); 45 outputStream.close(); 46 return result; 47 } catch (Exception e) { 48 throw new RuntimeException("发起远程调用异常"); 49 } finally { 50 if (socket != null) { 51 try { 52 socket.close(); 53 } catch (IOException ioException) { 54 ioException.printStackTrace(); 55 } 56 } 57 } 58 } 59 }
5，InvocationHandler实现
* 通过构造方法初始化注册中心对象和版本号
1 public class RemoteInvokeHandler implements InvocationHandler { 2 3 private ServiceDiscovery serviceDiscovery; 4 5 private String version; 6 7 public RemoteInvokeHandler(ServiceDiscovery serviceDiscovery, String version) { 8 this.serviceDiscovery = serviceDiscovery; 9 this.version = version; 10 } 11 12 /** 13 * 增强的InvocationHandler,接口调用方法的时候实际是调用socket进行传输 14 * 动态代理实现方法调用并提交服务请求发送 15 */ 16 @Override 17 public Object invoke(Object proxy, Method method, Object[] args) throws Throwable { 18 // 将远程调用需要的接口类、方法名、参数信息封装成RPCRequest 19 RpcRequest request = new RpcRequest(); 20 request.setClassName(method.getDeclaringClass().getName()); 21 request.setMethodName(method.getName()); 22 request.setParameters(args); 23 request.setVersion(version); 24 25 // 通过socket发送RPCRequest给服务端并获取结果返回 00:42 26 // 根据接口名称得到对应的服务地址 27 String serviceAddress = serviceDiscovery.discovery(request.getClassName()); 28 29 TcpTransport tcpTransport = new TcpTransport(serviceAddress); 30 31 return tcpTransport.send(request); 32 } 33 }
客户端接收响应
1，根据套接字输入流构建对象输入流（class RPCTransPort）
1 public Object send(RpcRequest request) { 2 Socket socket = null; 3 try { 4 socket = newSocket(); 5 ObjectOutputStream outputStream = new ObjectOutputStream(socket.getOutputStream()); 6 // 通过 ObjectOutputStream 将当前request传给服务端 7 outputStream.writeObject(request); 8 outputStream.flush(); 9 10 ObjectInputStream inputStream = new ObjectInputStream(socket.getInputStream()); 11 Object result = inputStream.readObject(); 12 inputStream.close(); 13 outputStream.close(); 14 return result; 15 } catch (Exception e) { 16 throw new RuntimeException("发起远程调用异常"); 17 } finally { 18 if (socket != null) { 19 try { 20 socket.close(); 21 } catch (IOException ioException) { 22 ioException.printStackTrace(); 23 } 24 } 25 } 26 }
4，请求结果
? * 从下图可以看到，8081和8082端口各有出现；这是因为服务端在注册服务时模拟分布式注册多个服务节点，客户端在处理服务节 点时进行了简易的基于随机数的负载均衡，main方法中循环调用
* 请求main方法分十次进行服务获取，每一次都会随机获取8080或者8081端口服务，并进行调用
1 public class ClientDemo { 2 3 public static void main(String[] args) throws RemoteException, NotBoundException, MalformedURLException, InterruptedException { 4 5 ServiceDiscovery serviceDiscovery = new ServiceDiscoveryImpl(ZkConfig.CONNECTION_STR); 6 RpcClientProxy proxy = new RpcClientProxy(serviceDiscovery); 7 for (int i = 0; i < 10; i++) { 8 IUserService service = proxy.clientProxy(IUserService.class,null); 9 System.out.println(service.sayHello("lisa")); 10 Thread.sleep(1000); 11 } 12 13 } 14 }
输出结果
三、实现多版本服务
1，服务端暴露服务自定义注解上添加version属性
1 @RpcAnnotation(value = IUserService.class,version = "1.0") 2 public class UserServiceImpl implements IUserService { 3 @Override 4 public String sayHello(String msg) { 5 return "[I‘M 8080] node : "+msg; 6 } 7 }
1 @RpcAnnotation(value = IUserService.class,version = "2.0") 2 public class UserServiceImpl2 implements IUserService { 3 @Override 4 public String sayHello(String msg) { 5 return "[I‘M 8081] node : "+msg; 6 } 7 }
2，服务绑定根据版本号分别绑定两个服务并注册到注册中心
* 修改main方法
1 public class ServiceDemo { 2 public static void main(String[] args) { 3 IUserService userService = new UserServiceImpl(); 4 IUserService helloService2 = new UserServiceImpl2(); 5 RegisterCenter registerService = new RegisterCenterImpl(); 6 // 初始化注册中心和服务端口信息 7 RPCServer server = new RPCServer(register, "127.0.0.1:8080"); 8 // 绑定服务 9 server.bind(helloService, helloService2); 10 // 发布服务 11 server.publishServer(); 12 System.out.println("服务发布成功"); 13 System.in.read(); 14 } 15 }
zookeeper节点查询
GitHub 地址：https://github.com/lwx57280/Distributed-topics
标签：信息 循环 ntp inpu exec cep 输出流 ++ 循环调用
原文地址：https://www.cnblogs.com/lwx57280/p/14398754.html
