【利用zookeeper分布式锁模拟12306抢票场景】利用锁完成高并发抢票的模拟

一、版本一: 单机锁: 采用 ReentrantLock实现.*

对比synchronized与 reentrantLock的区别？…

1.1RedisController类的编写

@RestController
public class RedisController {

 @Value("${server.port}")
    private String port;

    @Autowired
    private StringRedisTemplate redisTemplate;
     private ReentrantLock rt = new ReentrantLock();
     
    @GetMapping("/setKill2")
    public String testRedis()throws Exception {

      try{
          rt.lock();
          int num = Integer.parseInt(redisTemplate.opsForValue().get("goods"));
          if(num > 0){
              int realNum = num -1;
              redisTemplate.opsForValue().set("goods",realNum+"");

              System.out.println("由服务器 "+port+"抢购第"+num+"张票");
              return "由服务器"+port+"抢购第"+num+"张票成功";
          }else {
              System.out.println("库存不足");
              return "由服务器"+port+"抢购时库存不足";
          }
      }catch (Exception e){
          e.printStackTrace();
          return "系统异常";
      }finally {
          rt.unlock();
      }

    }
}

1.2 redis yml配置

#Redis服务器地址
spring.redis.host=124.220.200.205
#Redis服务器连接端口
spring.redis.port=6379
#Redis数据库索引（默认为0）
spring.redis.database= 0
#连接超时时间（毫秒）
spring.redis.timeout=1800000
#连接池最大连接数（使用负值表示没有限制）
spring.redis.lettuce.pool.max-active=20
#最大阻塞等待时间(负数表示没限制)
spring.redis.lettuce.pool.max-wait=-1
#连接池中的最大空闲连接
spring.redis.lettuce.pool.max-idle=5
#连接池中的最小空闲连接
spring.redis.lettuce.pool.min-idle=0

1.3在Redis里面设置goods的值

1.4进行ab压力测试 1.4.1 不加锁之前

  private ReentrantLock rt = new ReentrantLock();
      try{
//          rt.lock(); //注释
          int num = Integer.parseInt(redisTemplate.opsForValue().get("goods"));
                return "系统异常";
      }finally {
//          rt.unlock(); 已经注释

1.4.2压力测试

1.4.3结果分析

不加锁 private ReentrantLock rt = new ReentrantLock();
出现一票多卖的情况

1.4.5加锁之后

结果没有出现一票多买的情况

二、版本二: 利用nginx完成服务集群的搭建，并实现了nginx的负载均衡.

**
解决的问题: 高并发场景下单服务器的处理能力不够的问题. 利用ab测试模拟多个用户( 1000个用户 ) 密集并发的场景
问题: 单机锁无法保证在服务集群下数据操作的安全性. 存在一票多卖的情况，但单机上不会出现重复的票.

2.1 在集群条件下单机锁失效 2.1.1 在springboot建立多个服务进程

2.1.2多集群yml配置

2.1.3利用nginx对集群实现负载均衡配置

2.1.4集群下单机锁实现问题测试

结果展示：
我们发现在多个集群条件下出现了一票多买的情况即单机锁失效

学到的技术: 1. ab模拟高并发测试系统性能. 2.nginx服务器: 包括基本配置, 负载均衡配置,动静分离等. 还包括nginx的特点???.

三、版本三: 自定义基于zk的分布式锁. **基于zk的分布锁有两种实现方案: 1. 独占，就是所有试图来获取这个锁（znode)的客户端，最终只有一个可以成功获得这把锁。通常的做法是把zk上的一个znode看作是一把锁，通过create znode的方式来实现。所有客户端都去创建 /distribute_lock 节点，最终成功创建的那个客户端也即拥有了这把锁。 2. 控制时序:

1。客户端在/locks根节点下面创建临时有序节点（这个可以通过节点的属性控制：CreateMode.EPHEMERAL_SEQUENTIAL来指定）, 2。 client调用getChildren(“/root/lock_”,watch)来获取所有已经创建的子节点，并同时在这个节点上注册子节点变更通知的Watcher 3.客户端获取到所有子节点Path后，如果发现自己在步骤1中创建的节点是所有节点中最小的(排序），那么就认为这个客户端获得了锁 4.如果在步骤3中，发现不是最小的，那么等待，直到下次子节点变更通知的时候，在进行子节点的获取，判断是否获取到锁 5.释放锁也比较容易，就是删除自己创建的那个节点即可羊群效应.
另外了解了其它分布式锁的方案对比: 1. 利用数据库的表级锁和行级锁来完成分布式锁 2。利用 redis中的 setnx 完成分布式锁.**

3.1 自定义zookeeper分布式锁

```java

public class MyDistributedLock {
    

    private final int sessionTimeOut = 2000;

   private final ZooKeeper zk;
    //根节点
    private final String LOCK = "/locks";
    //同步工具类，构造传入计数值，当计数达到0，释放所有等待的线程
    private  CountDownLatch connectLatch = new CountDownLatch(1); //zk连接等待
    private CountDownLatch waitLatch = new CountDownLatch(1);//等待节点删除
    //监听上一个节点的路径
    private String currentNode;
    //当前节点
    private String waitPath;

    //构造方法：创建zk的联接
   public MyDistributedLock(String connectString) throws IOException, InterruptedException, KeeperException {
       //获取连接
        zk = new ZooKeeper(connectString, sessionTimeOut, new Watcher() {

            @Override
            //接收事件通知后的回调函数（用户的业务逻辑)
            public void process(WatchedEvent watchedEvent) {
                //连接上zk，释放连接等待锁

                if (watchedEvent.getState() == Event.KeeperState.SyncConnected) {//只有回调的状态是连接上才激活这个判断
                    System.out.println("zk客户端建立与服务器的连接");
                    //计数-1
                    connectLatch.countDown();//只有连接建立了再释放这把锁这样主线程才可以继续执行

                }
                if(watchedEvent.getType() == Event.EventType.NodeDeleted && watchedEvent.getPath().equals(waitPath)){
                  //释放锁
                    waitLatch.countDown();
                }

            }
        });
        //等待zk正常连接后，程序往下执行
       //在计数器为0之前一直等待
           waitLatch.await();//在主线程中await ，阻塞不为0就阻塞
       //判断根节点  /locks是否存在，不存在就创建根节点
       Stat stat =zk.exists(LOCK,false);
       if(stat == null){
           //创建下个根节点(持久节点） /locks
           zk.create(LOCK,"locks".getBytes(),ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT);
       }
    }

    void lock() throws InterruptedException, KeeperException {

        //1.创建临时有序的节点到  /locks下.
        //创建对应的
     currentNode =   zk.create(LOCK + "/seq-",null, ZooDefs.Ids.CREATOR_ALL_ACL, CreateMode.EPHEMERAL_SEQUENTIAL);

//        MyMatch nm = new MyMatch();
        List children = zk.getChildren(LOCK,false);
        if(children.size() == 1){
            return;
        }else {
            Collections.sort(children);
            String thisNode = currentNode.substring((LOCK+"/").length());
            int index = children.indexOf(thisNode);
            if(index == 0){
                return;
            }else {
                waitPath = LOCK + "/" +children.get(index - 1);
                zk.getData(waitPath,true,new Stat());
                waitLatch.await();
                return;
            }
            // 2. 获取  /locks下所有的子节点列表 /
            //  3. 判断  此列表长度为 1的话，则当前 currentNode就可以获取到锁,  return .
            //          长度不为1. , 有多个节点
            //                           4. 子节点列表排序
            //                            //取最小值 .
            //                                监听上一节点....
        }

    }

    void unlock() throws InterruptedException, KeeperException {
        //删除节点,   哪一个.  加了lock那个.     currentNode;
        zk.delete(currentNode,-1);
    }






}

3.2 基于分布式锁的redisController

@RestController
public class RedisController2 {

    @Value("${server.port}")
    private String port;
    @Autowired
    private StringRedisTemplate redisTemplate;
    //1.单机锁  为什么写在secKill（）控制方法的外面  可重入

    private ReentrantLock rt = new ReentrantLock();

    @GetMapping("/setKill3")
    public String testRedis()throws Exception {
 MyDistributedLock rt = new MyDistributedLock("124.220.200.205:2181");
        try{
            rt.lock();
            int num = Integer.parseInt(redisTemplate.opsForValue().get("goods"));
            if(num > 0){
                int realNum = num -1;
                redisTemplate.opsForValue().set("goods",realNum+"");

                System.out.println("由服务器 "+port+"抢购第"+num+"张票");
                return "由服务器"+port+"抢购第"+num+"张票成功";
            }else {
                System.out.println("库存不足");
                return "由服务器"+port+"抢购时库存不足";
            }
        }catch (Exception e){
            e.printStackTrace();
            return "系统异常";
        }finally {
            rt.unlock();
        }

    }
}

3.3进行ab测试

电脑太卡了就没测试了
因为每一次都要创建一次zookeeper连接太消耗性能了。

四、版本四: 引用zookeeper的 curator 的框架锁. InterProcessMutex

zookeeper分布式框架，每次只需建立一次连接就好了，比前面大大提高了性能

@RestController
public class RedisControllerEnd {

    @Value("${server.port}")
    private String port;

    @Autowired
    private StringRedisTemplate redisTemplate;

    InterProcessMutex lock = new InterProcessMutex(getzkClient(),"/locks");


    //1.单机锁  为什么写在secKill（）控制方法的外面  可重入

    private ReentrantLock rt = new ReentrantLock();

    @GetMapping("/setKill")
    public String testRedis()throws Exception {
//        MyDistributedLock rt = new MyDistributedLock("124.220.200.205:2181");
        try{
//            rt.lock();
            lock.acquire();
            int num = Integer.parseInt(redisTemplate.opsForValue().get("goods"));
            if(num > 0){
                int realNum = num -1;
                redisTemplate.opsForValue().set("goods",realNum+"");

                System.out.println("由服务器 "+port+"抢购第"+num+"张票");
                return "由服务器"+port+"抢购第"+num+"张票成功";
            }else {
                System.out.println("库存不足");
                return "由服务器"+port+"抢购时库存不足";
            }
        }catch (Exception e){
            e.printStackTrace();
            return "系统异常";
        }finally {
//            rt.unlock();
            lock.release();
        }

    }

    private CuratorFramework getzkClient() {

        ExponentialBackoffRetry policy = new ExponentialBackoffRetry(3000,3);
        CuratorFramework client = CuratorFrameworkFactory.builder()
                .connectString("124.220.200.205:2181")
                .connectionTimeoutMs(2000)
                .sessionTimeoutMs(2000)
                .retryPolicy(policy)
                .build();
        client.start();
        return client;
    }

}

4.2进行ab测试解决分布式条件下一票多买情况

【利用zookeeper分布式锁模拟12306抢票场景】利用锁完成高并发抢票的模拟

Linux相关栏目本月热门文章