第七章分布式锁&服务注册与发现笔记

太过爱你忘了你带给我的痛 2024-04-17 14:44 11阅读 0赞

# **一、分布式锁** #

## **1、为什么需要锁？(一多二写三互斥)** ##

*  多任务环境中才需要
 *  任务都需要对同一共享资源进行写操作；
 *  对资源的访问是互斥的

任务通过竞争获取锁才能对该资源进行操作(竞争锁)；当有一个任务在对资源进行更新时（占有锁），其他任务都不可以对这个资源进行操作（任务阻塞），直到该任务完成更新(释放锁)。

## **2、分布式锁方案比较** ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NjYxNDU4_size_16_color_FFFFFF_t_70][]

### **2、模板方法模式** ###

在父类中编排主流程，将步骤实现延迟到子类去实现。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NjYxNDU4_size_16_color_FFFFFF_t_70 1][]

*  定义锁的接口Lock
 *  在AbstractLock模板锁里面实现getLock方法，实现通用的逻辑。
 *  不能确实的步骤，作为虚拟方法，甩锅给子类实现。
 *  子类只需要聚焦自己的小步骤逻辑，实现tryLock,waitLock,unLock方法

**2.1、模板锁的实现代码如下：**

public abstract  class AbstractTemplate {
        public void shopping() {
            Map<String,Float>  cars = new HashMap();
            cars.put("手机",10f);
            cars.put("电视",20f);
            cars.put("电脑",30f);
            checkGoods(cars);//1.清点商品
            float mony = calculation(cars);//2计算价格
            if(pay(mony)) { //支付
                delivery();//如果支付成功送货
            }
        }
        public void checkGoods(Map<String,Float>  cars) {
            if(cars!=null) {
                System.out.print("你购买的了：");
                for(String key:cars.keySet()) {
                    System.out.print(key+" ");
                }
                System.out.println();
            }
        }
        public float calculation(Map<String,Float>  cars) {
            float result = 0;
            if(cars!=null) {
                for(String key:cars.keySet()) {
                    result += cars.get(key);
                }
            }
            System.out.println("你总共应该支付："+result);
            return  result;//钱
        }
        public abstract boolean pay(Float money) ;
        public  void delivery() {
            System.out.println("请稍等，正在送货中！");
        }
    }
    
    //实现支付方式
    public class WxPayTemplate  extends AbstractTemplate{
        public boolean pay(Float money) {
            System.out.println("hankin使用微信支付了:"+money);
            return true;
        }
        public static void main(String[] args) {
           AbstractTemplate at = new WxPayTemplate();
           at.shopping();
        }
    }

代码运行结果：

你购买的了：电脑 手机 电视

你总共应该支付：60.0

hankin使用微信支付了:60.0

请稍等，正在送货中！

**2.2、使用锁lock的方式实现**

public class OrderNumGenerator {
        public  static int count = 0;//全局订单id
        private java.util.concurrent.locks.Lock lock = new ReentrantLock();
        //以lock的方式解决
        public  String getNumber() {
            try {
                lock.lock();
                SimpleDateFormat simpt = new SimpleDateFormat("yyyy-MM-dd-HH-mm-ss");
                String s = simpt.format(new Date()) + "-" + ++count;
                return s;
            }finally {
                lock.unlock();
            }
        }
    }
    
    //使用多线程模拟生成订单号
    public class OrderService implements Runnable {
        private OrderNumGenerator orderNumGenerator = new OrderNumGenerator();
        private static CountDownLatch countDownLatch = new CountDownLatch(50);
        private static List<String> result = new Vector<String>();
        public void run() {
            try {
                countDownLatch.await();
                result.add(orderNumGenerator.getNumber());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        public static void main(String[] args) throws InterruptedException {
            System.out.println("####生成唯一订单号###");
            OrderService orderService = new OrderService();
            for (int i = 0; i < 50; i++) {
                new Thread( orderService).start();
                countDownLatch.countDown();
            }
            Thread.sleep(1000);
            Collections.sort(result);
            for(String str:result) {
               System.out.println(str);
            }
        }
    }

**2.3、基于MySQL锁的实现代码如下：**

@Service
    public  class MysqlLock extends AbstractLock {
       @Resource
       private LockMapper mapper;
       private static final int LOCK_ID = 1;//所有的线程都往数据库插入主键值相同的数据
       //非阻塞式加锁
       public boolean tryLock() {
          try {
             mapper.insert(LOCK_ID);
          } catch (Exception e) {
             return false;
          }
          return true;
       }
       //让当前线程休眠一段时间deer
       public void waitLock() {
          try {
             Thread.currentThread().sleep(10);
          } catch (InterruptedException e) {
             e.printStackTrace();
          }
       }
       public void unLock() {
          mapper.deleteByPrimaryKey(LOCK_ID);
       }
    }
    
    public interface LockMapper {
        int deleteByPrimaryKey(int id);//删除数据解锁
        int insert(int id);//新增数据加锁，id为同一个值
    }

LockMapper.xml代码如下：

<delete id="deleteByPrimaryKey" parameterType="INTEGER" >
      delete from t_lock
      where id = #{code,jdbcType=INTEGER}
    </delete>
    <insert id="insert" parameterType="INTEGER" >
      insert into t_lock (id)
      values (#{id,jdbcType=INTEGER})
    </insert>

### **3、zookeeper 基于同名节点的分布式锁** ###

**3.1、基础功能代码实现如下：**

**1）****定义锁接口及抽象类的实现方法：**

public interface Lock {
        public void getLock();//获取到锁的资源
        public void unLock();// 释放锁
    }
    
    public abstract class AbstractLock  implements  Lock{
        public void getLock() {
            //任务通过竞争获取锁才能对该资源进行操作(①竞争锁)；当有一个任务在对资源进行更新时（②占有锁），
            // 其他任务都不可以对这个资源进行操作（③任务阻塞），直到该任务完成更新(④释放锁)
            // 尝试获得锁资源
            if (tryLock()) {// ①竞争锁
                System.out.println("##获取lock锁的资源####");
            } else {
                waitLock();//③任务阻塞
                getLock(); // 重新获取锁资源
            }
        }
        public abstract boolean tryLock(); // ②占有锁
        public abstract void waitLock();
    }

2）Zk锁抽象类实现：

//将重复代码写入子类中..
    public abstract class ZookeeperAbstractLock extends AbstractLock {
        // zk连接地址
        private static final String CONNECTSTRING = "127.0.0.1:2181";
        // 创建zk连接
        protected ZkClient zkClient = new ZkClient(CONNECTSTRING);
        protected static final String PATH = "/lock";
        protected static final String PATH2 = "/lock2";
    }

3）订单线程模拟生成50个订单：

public class OrderService implements Runnable {
        private OrderNumGenerator orderNumGenerator = new OrderNumGenerator();
        //private Lock lock = new ZookeeperkLock();
        private Lock lock = new ZookeeperDistrbuteLock2();
        public void run() {
            getNumber();
        }
        public void getNumber() {
            try {
                lock.getLock();
                String number = orderNumGenerator.getNumber();
                System.out.println(Thread.currentThread().getName() + ",生成订单ID:" + number);
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                lock.unLock();
            }
        }
        public static void main(String[] args) {
            System.out.println("####生成唯一订单号###");
    //    OrderService orderService = new OrderService();
            for (int i = 0; i < 50; i++) {
                new Thread( new OrderService()).start();
            }
        }
    }

**3.2、实现zk锁的方式1：**

利用zk临时节点的特性，通过对临时节点的创建、删除实现zk锁

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NjYxNDU4_size_16_color_FFFFFF_t_70 2][]

public class ZookeeperDistrbuteLock extends ZookeeperAbstractLock {
        private CountDownLatch countDownLatch = null;
        //尝试获得锁
    
    @Override
        public  boolean tryLock() {
            try {
                zkClient.createEphemeral(PATH);
                return true;
            } catch (Exception e) {
                //如果创建失败报出异常 e.printStackTrace();
                return false;
            }
        }
        @Override
        public void waitLock() {
            IZkDataListener izkDataListener = new IZkDataListener() {
                public void handleDataDeleted(String path) throws Exception {
                    // 唤醒被等待的线程
                    if (countDownLatch != null) {
                        countDownLatch.countDown();
                    }
                }
                public void handleDataChange(String path, Object data) throws Exception {}
            };
            // 注册事件
            zkClient.subscribeDataChanges(PATH, izkDataListener);
            //如果节点存在
            if (zkClient.exists(PATH)) {
                countDownLatch = new CountDownLatch(1);
                try {
                    countDownLatch.await(); //等待，一直等到接受到事件通知
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
            zkClient.unsubscribeDataChanges(PATH, izkDataListener); // 删除监听
        }
        //释放锁
        public void unLock() {
            if (zkClient != null) {
                zkClient.delete(PATH);
                zkClient.close();
                System.out.println("释放锁资源...");
            }
        }
    }

**3.3、实现zk锁的方式2：**

利用zk临时顺序节点的特性，通过对临时顺序节点的创建、删除实现zk锁

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NjYxNDU4_size_16_color_FFFFFF_t_70 3][]

public class ZookeeperDistrbuteLock2 extends ZookeeperAbstractLock {
        private CountDownLatch countDownLatch= null;
        private String beforePath;//当前请求的节点前一个节点
        private String currentPath;//当前请求的节点
        public ZookeeperDistrbuteLock2() {
            if (!this.zkClient.exists(PATH2)) {
                this.zkClient.createPersistent(PATH2);
            }
        }
        @Override
        public boolean  tryLock() {
            //如果currentPath为空则为第一次尝试加锁，第一次加锁赋值currentPath
            if(currentPath == null || currentPath.length()<= 0){
                //创建一个临时顺序节点
                currentPath = this.zkClient.createEphemeralSequential(PATH2 + '/',"lock");
            }
            //获取所有临时节点并排序，临时节点名称为自增长的字符串如：0000000400
            List<String> childrens = this.zkClient.getChildren(PATH2);
            Collections.sort(childrens);
            if (currentPath.equals(PATH2 + '/'+childrens.get(0))) {//如果当前节点在所有节点中排名第一则获取锁成功
                return true;
            } else {//如果当前节点在所有节点中排名中不是排名第一，则获取前面的节点名称，并赋值给beforePath
                int wz = Collections.binarySearch(childrens,currentPath.substring(7));
                beforePath = PATH2 + '/'+childrens.get(wz-1);
            }
            return false;
        }
        @Override
        public void waitLock() {
            IZkDataListener listener = new IZkDataListener() {
                public void handleDataDeleted(String dataPath) throws Exception {
                    if(countDownLatch!=null){
                        countDownLatch.countDown();
                    }
                }
                public void handleDataChange(String dataPath, Object data) throws Exception {}
            };
            //给排在前面的的节点增加数据删除的watcher,本质是启动另外一个线程去监听前置节点
            this.zkClient.subscribeDataChanges(beforePath, listener);
            if(this.zkClient.exists(beforePath)){
                countDownLatch=new CountDownLatch(1);
                try {
                    countDownLatch.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            this.zkClient.unsubscribeDataChanges(beforePath, listener);
        }
        public void unLock() {
            zkClient.delete(currentPath); //删除当前临时节点
            zkClient.close();
        }
    }

## **二、服务注册与发现** ##

### **1、微服务的本质？** ###

*  负载均衡(Load Balance)
 *  请求分摊到多个操作单元
 *  共同完成工作任务

优点：均摊并发压力，增加系统可用性

**服务粒度细分带来的好处****：**

*  单一职责功能，每个服务都很简单
 *  易于规模化开发，单独团队维护、工作分明，职责清晰
 *  改善故障隔离。一个服务宕机不会影响其他的服务
 *  架构上系统更加清晰

### **2、微服务仅仅是负载均衡？** ###

**负载均衡之外的关注点**

*  复杂性扩散
 *  库的复用与耦合
 *  可替代成本太高

**微服务的治理**

已经实现了服务的负载和服务的细分，但带来了新的问题：混乱

### **3、服务注册与发现** ###

**混乱的解决方式-注册中心****：**

*  产品把信息录入系统（服务注册）
 *  订单查询技师列表（服务发现）
 *  订单从中筛选一个调用服务（负载算法）

**服务的注册：**把服务端的地址信息（IP，端口等）注册到注册中心的过程

**服务的发现：**客户端从注册中心找寻具体的服务的过程（服务的发现+负载）

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NjYxNDU4_size_16_color_FFFFFF_t_70 4][]

### **4、服务注册与发现主流实现方式** ###

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NjYxNDU4_size_16_color_FFFFFF_t_70 5][]

**监听服务（观察者模式）****：**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NjYxNDU4_size_16_color_FFFFFF_t_70 6][]

## **三、服务注册发现代码实现逻辑** ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NjYxNDU4_size_16_color_FFFFFF_t_70 7][]

### **1、产品服务代码如下：** ###

**1）服务注册部分代码（创建一个持久节点）**

public class ServiceRegister {
        private  static final String BASE_SERVICES = "/services";
        private static final String  SERVICE_NAME = "/products";
        public  static void register(String address,int port) {
            try {
                //产品服务最终会注册倒的地址
                String path = BASE_SERVICES + SERVICE_NAME;
                ZooKeeper zooKeeper =  new ZooKeeper("127.0.0.1:2181",5000, (watchedEvent)->{});
                Stat exists = zooKeeper.exists(BASE_SERVICES + SERVICE_NAME, false);
                // 先判断服务根路径是否存在
                if(exists == null) {
                    zooKeeper.create(path,"".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
                }
                String server_path = address+":" +port; // 拼接ip和端口
                // 注册的类型，EPHEMERAL_SEQUENTIAL 零时，并且带序号
                zooKeeper.create(path+"/child",server_path.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.EPHEMERAL_SEQUENTIAL);
                System.out.println("产品服务注册成功");
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

2）服务监听代码如下：

public class InitListener implements ServletContextListener {
        //容器初始化的时候会调用，重启tomcat
        @Override
        public void contextInitialized(ServletContextEvent sce) {
            try {
                Properties properties =  new Properties();
                properties.load(InitListener.class.getClassLoader().getResourceAsStream("application.properties"));
                String hostAddress = InetAddress.getLocalHost().getHostAddress(); //获得IP
                int port = Integer.valueOf(properties.getProperty("server.port")); //获得端口
                ServiceRegister.register(hostAddress,port);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        @Override
        public void contextDestroyed(ServletContextEvent sce) {
        }
    }

3）对外提供接口代码：

@RequestMapping("/product")
    @RestController
    public class ProductController {
        @RequestMapping("/getProduct/{id}")
        public Object getProduct(HttpServletRequest request, @PathVariable("id") String id) {
            int localPort = request.getLocalPort();
            return new Product(id,"productname:"+localPort);
        }
    }

### **2、订单服务部分代码逻辑** ###

1）对外提供接口类代码：

@RestController
    @RequestMapping("/order")
    public class OrderController {
        @Resource
        private RestTemplate restTemplate;
        private LoadBalance loadBalance = new RamdomLoadBalance();
    
        @RequestMapping("/getOrder/{id}")
        public Object getProduct(HttpServletRequest request, @PathVariable("id") String id) {
            //使用随机的方式获得一个IP和端口
            String host = loadBalance.choseServiceHost();
            Product product = restTemplate.getForObject("http://"+host+"/product/getProduct/1", Product.class);
            return new Order(id,"ordername",product);
        }
    }

2）监听器代码逻辑实现如下：

public class InitListener implements ServletContextListener {
        private  static final String BASE_SERVICES = "/services";
        private static final String  SERVICE_NAME="/products";
        private   ZooKeeper zooKeeper;
        private  void init()  {
            try {//连接上zk 获得列表信息
                zooKeeper = new ZooKeeper("127.0.0.1:2181",5000,(watchedEvent)->{
                      //当有节点变更的时候通知倒orderService
                    if(watchedEvent.getType() == Watcher.Event.EventType.NodeChildrenChanged
                            && watchedEvent.getPath().equals(BASE_SERVICES+SERVICE_NAME)) {
                        updateServerList();
                    }
                });
                updateServerList(); //第一次连接的时候要获得列表
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        private  void updateServerList() {
            List<String> newServerList = new ArrayList<String>();
            try {
                List<String> children = zooKeeper.getChildren(BASE_SERVICES  + SERVICE_NAME, true);
                for(String subNode:children) {
                    byte[] data = zooKeeper.getData(BASE_SERVICES  + SERVICE_NAME + "/" + subNode, false, null);
                    String host = new String(data, "utf-8");
                    System.out.println("host:"+host);
                    newServerList.add(host);
                }
                LoadBalance.SERVICE_LIST = newServerList;
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        @Override
        public void contextInitialized(ServletContextEvent sce) {
            init();
        }
        @Override
        public void contextDestroyed(ServletContextEvent sce) { }
    }

3）简单实现一个负载随机算法

public abstract class LoadBalance {
        public volatile static   List<String> SERVICE_LIST;
        public abstract String choseServiceHost();
    }
    
    //实现了一个随机的算法
    public class RamdomLoadBalance extends  LoadBalance {
        @Override
        public String choseServiceHost() {
            String result = "";
            //SERVICE_LIST 产品 服务的列表
            if(!CollectionUtils.isEmpty(SERVICE_LIST)) {
               // 192.168.30.3:8083      192.168.30.3:8084  2
                //index 0,1  传入一个种子
                int index = new Random().nextInt(SERVICE_LIST.size());
                result = SERVICE_LIST.get(index);
            }
            return result ;
        }
    }

4）订单启动类代码如下：

@SpringBootApplication
    public class OrderApp {
        public static void main(String[] args) {
            SpringApplication.run(OrderApp.class,args);
        }
        @Bean
        public RestTemplate restTemplate() {
            return new RestTemplate();
        }
        @Bean
        public ServletListenerRegistrationBean servletListenerRegistrationBean() {
            ServletListenerRegistrationBean servletListenerRegistrationBean = new ServletListenerRegistrationBean();
            servletListenerRegistrationBean.setListener(new InitListener());
            return  servletListenerRegistrationBean;
        }
    }

### **3、测试服务部分代码逻辑** ###

1）入口类代码

@RestController
    public class ProductController {
        @RequestMapping("/getProduct/{id}")
        public  Object getProdcut(HttpServletRequest request, @PathVariable("id") String id) {
            Product product = new Product();
            product.setId(id);
            product.setName("product-"+request.getLocalPort());
            return product;
        }
    }

2）启动类服务代码：

@SpringBootApplication
    public class SbApp {
        public static void main(String[] args) throws  Exception {
            ConfigurableApplicationContext run = SpringApplication.run(SbApp.class, args);
            Thread.sleep(10000);
            System.out.println("执行到这一行");
            run.close();
        }
        //启动监听起，当tomcat启动的时候，会调用InitListener
        @Bean
        public ServletListenerRegistrationBean servletListenerRegistrationBean() {
            ServletListenerRegistrationBean servletListenerRegistrationBean = new ServletListenerRegistrationBean();
            servletListenerRegistrationBean.setListener(new InitListener());
            return  servletListenerRegistrationBean;
        }
        @Bean
        public RestTemplate restTemplate() {
            return new RestTemplate();
        }
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NjYxNDU4_size_16_color_FFFFFF_t_70]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/17/9a611b1249ce4f72865c8aedcf73a0ba.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NjYxNDU4_size_16_color_FFFFFF_t_70 1]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/17/3f6b10c2bc5f43979a305f50db0a5359.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NjYxNDU4_size_16_color_FFFFFF_t_70 2]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/17/71bb7931df684f71a2322f47c1c6e595.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NjYxNDU4_size_16_color_FFFFFF_t_70 3]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/17/5115541e36134bfdbd4875d816b1b948.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NjYxNDU4_size_16_color_FFFFFF_t_70 4]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/17/3474fab7e24848fdb24228efe905c4b4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NjYxNDU4_size_16_color_FFFFFF_t_70 5]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/17/9a73bcae417045788e9e90194a82e95e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NjYxNDU4_size_16_color_FFFFFF_t_70 6]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/17/8c0d84da867444ac9959c42a46fd3a3f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NjYxNDU4_size_16_color_FFFFFF_t_70 7]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/17/8a23d3c422cc4a2c998c63dd7259ed4a.png