select * from table where id=1 。此SQL无论执行多少次，虽然结果有可能出现不同，都不会对数据产生
改变，具备幂等性。
insert into table(id,name) values(1,'heima') 。此SQL如果id或name有唯一性约束，多次操作只允许插
入一条记录，则具备幂等性。如果不是，则不具备幂等性，多次操作会产生多条数据。
update table set score=100 where id = 1 。此SQL无论执行多少次，对数据产生的影响都是相同的。具备
幂等性。
update table set score=50+score where id = 1 。此SQL涉及到了计算，每次操作对数据都会产生影响。
不具备幂等性。
delete from table where id = 1 。此SQL多次操作，产生的结果相同，具备幂等性。

幂等性设计主要从两个维度进行考虑：空间、时间。
空间：定义了幂等的范围，如生成订单的话，不允许出现重复下单。
时间：定义幂等的有效期。有些业务需要永久性保证幂等，如下单、支付等。而部分业务只要保证一段时间
幂等即可。
同时对于幂等的使用一般都会伴随着出现锁的概念，用于解决并发安全问题。

接口幂等

? 对于幂等的考虑，主要解决两点前后端交互与服务间交互。这两点有时都要考虑幂等性的实现。从前端的思路解决
的话，主要有三种：前端防重、PRG模式、Token机制。
2.1）前端防重
? 通过前端防重保证幂等是最简单的实现方式，前端相关属性和JS代码即可完成设置。可靠性并不好，有经验的人员可以通过工具跳过页面仍能重复提交。主要适用于表单重复提交或按钮重复点击。
2.2）PRG模式
? PRG模式即POST-REDIRECT-GET。当用户进行表单提交时，会重定向到另外一个提交成功页面，而不是停留在原先的表单页面。这样就避免了用户刷新导致重复提交。同时防止了通过浏览器按钮前进/后退导致表单重复提交。
是一种比较常见的前端防重策略。
2.3）token机制
? 2.3.1）方案介绍
? 通过token机制来保证幂等是一种非常常见的解决方案，同时也适合绝大部分场景。该方案需要前后端进行一定程度的交互来完成。

1）服务端提供获取token接口，供客户端进行使用。服务端生成token后，如果当前为分布式架构，将token存放于redis中，如果是单体架构，可以保存在jvm缓存中。

2）当客户端获取到token后，会携带着token发起请求。

3）服务端接收到客户端请求后，首先会判断该token在redis中是否存在。如果存在，则完成进行业务处理，业务处理完成后，再删除token。如果不存在，代表当前请求是重复请求，直接向客户端返回对应标识。

? 但是现在有一个问题，当前是先执行业务再删除token。在高并发下，很有可能出现第一次访问时token存在，完成具体业务操作。但在还没有删除token时，客户端又携带token发起请求，此时，因为token还存在，第二次请求也会验证通过，执行具体业务操作。

对于这个问题的解决方案的思想就是并行变串行。会造成一定性能损耗与吞吐量降低。
第一种方案：对于业务代码执行和删除token整体加线程锁。当后续线程再来访问时，则阻塞排队。
第二种方案：借助redis单线程和incr是原子性的特点。当第一次获取token时，以token作为key，对其进行自增。
然后将token进行返回，当客户端携带token访问执行业务代码时，对于判断token是否存在不用删除，而是对其继续incr。如果incr后的返回值为2。则是一个合法请求允许执行，如果是其他值，则代表是非法请求，直接返回。

? 那如果先删除token再执行业务呢？其实也会存在问题，假设具体业务代码执行超时或失败，没有向客户端返回明确结果，那客户端就很有可能会进行重试，但此时之前的token已经被删除了，则会被认为是重复请求，不再进行业务处理。

? 这种方案无需进行额外处理，一个token只能代表一次请求。一旦业务执行出现异常，则让客户端重新获取令牌，重新发起一次访问即可。推荐使用先删除token方案
? 但是无论先删token还是后删token，都会有一个相同的问题。每次业务请求都回产生一个额外的请求去获取token。但是，业务失败或超时，在生产环境下，一万个里最多也就十个左右会失败，那为了这十来个请求，让其他九千九百多个请求都产生额外请求，就有一些得不偿失了。虽然redis性能好，但是这也是一种资源的浪费。

服务幂等

防重表

对于防止数据重复提交，还有一种解决方案就是通过防重表实现。防重表的实现思路也非常简单。首先创建一张表
作为防重表，同时在该表中建立一个或多个字段的唯一索引作为防重字段，用于保证并发情况下，数据只有一条。
在向业务表中插入数据之前先向防重表插入，如果插入失败则表示是重复数据。

对于防重表的解决方案，可能有人会说为什么不使用悲观锁。悲观锁在使用的过程中也是会发生死锁的。悲观锁是
通过锁表的方式实现的。 假设现在一个用户A访问表A（锁住了表A），然后试图访问表B； 另一个用户B访问表
B（锁住了表B），然后试图访问表A。 这时对于用户A来说，由于表B已经被用户B锁住了，所以用户A必须等到用
户B释放表B才能访问。 同时对于用户B来说，由于表A已经被用户A锁住了，所以用户B必须等到用户A释放表A才
能访问。此时死锁就已经产生了。

Mysql乐观锁保证幂等

MySQL乐观锁是基于数据库完成分布式锁的一种实现，实现的方式有两种：基于版本号、基于条件。但是实现思
想都是基于MySQL的行锁思想来实现的。

通过版本号控制是一种非常常见的方式，适合于大多数场景。但现在库存扣减的场景来说，通过版本号控制就是多
人并发访问购买时，查询时显示可以购买，但最终只有一个人能成功，这也是不可以的。其实最终只要商品库存不
发生超卖就可以。那此时就可以通过条件来进行控制。

mysql乐观锁更适用于一些需要计数的表上，而且在竞争不激烈，出现并发冲突几率较小时，推荐使用乐观锁。虽
然通过MySQL乐观锁可以完成并发控制，但锁的操作是直接作用于数据库上，这样就会在一定程度上对数据库性能产生影响。并且mysql的连接数量是有限的，如果出现大量锁操作占用连接时，也会造成MySQL的性能瓶颈。

zookeeper分布式锁

实现思想

对于分布式锁的实现，zookeeper天然携带的一些特性能够很完美的实现分布式锁。其内部主要是利用znode节点
特性和watch机制完成。

在zookeeper中节点会分为四类，分别是：
持久节点：一旦创建，则永久存在于zookeeper中，除非手动删除。
持久有序节点：一旦创建，则永久存在于zookeeper中，除非手动删除。同时每个节点都会默认存在节点序号，每个节点的序号都是有序递增的。如demo000001、demo000002.....demo00000N。
临时节点：当节点创建后，一旦服务器重启或宕机，则被自动删除。
临时有序节点：当节点创建后，一旦服务器重启或宕机，则被自动删除。同时每个节点都会默认存在节点序号，每个节点的序号都是有序递增的。如demo000001、demo000002.....demo00000N。

watch监听机制

watch监听机制主要用于监听节点状态变更，用于后续事件触发，假设当B节点监听A节点时，一旦A节点发生修
改、删除、子节点列表发生变更等事件，B节点则会收到A节点改变的通知，接着完成其他额外事情。

实现原理

其实现思想是当某个线程要对方法加锁时，首先会在zookeeper中创建一个与当前方法对应的父节点，接着每个要
获取当前方法的锁的线程，都会在父节点下创建一个临时有序节点，因为节点序号是递增的，所以后续要获取锁的
线程在zookeeper中的序号也是逐次递增的。根据这个特性，当前序号最小的节点一定是首先要获取锁的线程，因
此可以规定序号最小的节点获得锁。所以，每个线程再要获取锁时，可以判断自己的节点序号是否是最小的，如果
是则获取到锁。当释放锁时，只需将自己的临时有序节点删除即可。

在并发下，每个线程都会在对应方法节点下创建属于自己的临时节点，且每个节点都是临时且有序的。
那么zookeeper又是如何有序的将锁分配给不同线程呢？ 这里就应用到了watch监听机制。每当添加一个新的临时
节点时，其都会基于watcher机制监听着它本身的前一个节点等待前一个节点的通知，当前一个节点删除时，就轮
到它来持有锁了。然后依次类推。

优缺点

1）zookeeper是基于cp模式，能够保证数据强一致性。

2）基于watch机制实现锁释放的自动监听，锁操作性能较好。

3）频繁创建节点，对于zk服务器压力较大，吞吐量没有redis强。

原理剖析

低效锁思想

在通过zookeeper实现分布式锁时，有另外一种实现的写法，这种也是非常常见的，但是它的效率并不高，此处可
以先对这种实现方式进行探讨。

此种实现方式，只会存在一个锁节点。当创建锁节点时，如果锁节点不存在，则创建成功，代表当前线程获取到
锁，如果创建锁节点失败，代表已经有其他线程获取到锁，则该线程会监听锁节点的释放。当锁节点释放后，则继
续尝试创建锁节点加锁。

这种方案的低效点就在于，只有一个锁节点，其他线程都会监听同一个锁节点，一旦锁节点释放后，其他线程都会
收到通知，然后竞争获取锁节点。这种大量的通知操作会严重降低zookeeper性能，对于这种由于一个被watch的
znode节点的变化，而造成大量的通知操作，叫做羊群效应。

高效锁思想

为了避免羊群效应的出现，业界内普遍的解决方案就是，让获取锁的线程产生排队，后一个监听前一个，依次排
序。推荐使用这种方式实现分布式锁

按照上述流程会在根节点下为每一个等待获取锁的线程创建一个对应的临时有序节点，序号最小的节点会持有锁，
并且后一个节点只监听其前面的一个节点，从而可以让获取锁的过程有序且高效。

代码实现

public abstract class AbstractLock {
 
    //zookeeper服务器地址
    public static final String ZK_SERVER_ADDR="192.168.200.131:2181";
 
    //zookeeper超时时间
    public static final int CONNECTION_TIME_OUT=30000;
    public static final int SESSION_TIME_OUT=30000;
 
    //创建zk客户端
    protected ZkClient zkClient = new 
ZkClient(ZK_SERVER_ADDR,SESSION_TIME_OUT,CONNECTION_TIME_OUT);
 
    /**
     * 获取锁
     * @return
     */
    public abstract boolean tryLock();
 
    /**
     * 等待加锁
     */
    public abstract void waitLock();
 
    /**
     * 释放锁
     */
    public abstract void releaseLock();
 
    public void getLock() {
 
        String threadName = Thread.currentThread().getName();
 
        if (tryLock()) {
            System.out.println(threadName+":   获取锁成功");
        }else {
            System.out.println(threadName+":   获取锁失败，等待中");
            //等待锁
            waitLock();
            getLock();
        }
    }
}

  public class HighLock extends AbstractLock{
 
    private static final String PARENT_NODE_PATH="/high_lock";
 
    //当前节点路径
    private String currentNodePath;
 
    //前一个节点的路径
    private String preNodePath;
 
    private CountDownLatch countDownLatch;
 
    @Override
    public boolean tryLock() {
 
        //判断父节点是否存在
        if (!zkClient.exists(PARENT_NODE_PATH)){
            //不存在
            zkClient.createPersistent(PARENT_NODE_PATH);
        }
 
        //创建第一个临时有序子节点
        if (currentNodePath == null || "".equals(currentNodePath)){
 
            //根节点下没有节点信息，将当前节点作为第一个子节点,类型：临时有序
            currentNodePath = zkClient.createEphemeralSequential(PARENT_NODE_PATH+"/","lock");
        }
 
        //不是第一个子节点，获取父节点下所有子节点
        List<String> childrenNodeList = zkClient.getChildren(PARENT_NODE_PATH);
 
        //子节点升序排序
        Collections.sort(childrenNodeList);
 
        //判断是否加锁成功
        if (currentNodePath.equals(PARENT_NODE_PATH+"/"+childrenNodeList.get(0))){
            //当前节点是序号最小的节点
            return true;
        }else {
            //当前节点不是序号最小的节点，获取其前面的节点名称，并赋值
            int length = PARENT_NODE_PATH.length();
            int currentNodeNumber = Collections.binarySearch(childrenNodeList, 
currentNodePath.substring(length + 1));
            preNodePath = PARENT_NODE_PATH+"/"+childrenNodeList.get(currentNodeNumber‐1);
        }
        return false;
    }
 
    @Override
    public void waitLock() {
 
        IZkDataListener zkDataListener = new IZkDataListener() {
            @Override
            public void handleDataChange(String dataPath, Object data) throws Exception {
 
            }
 
            @Override
            public void handleDataDeleted(String dataPath) throws Exception {
 
                if (countDownLatch != null){
                    countDownLatch.countDown();
                }
            }
        };
 
        //监听前一个节点的改变
        zkClient.subscribeDataChanges(preNodePath,zkDataListener);
 
        if (zkClient.exists(preNodePath)){
            countDownLatch = new CountDownLatch(1);
 
            try {
                countDownLatch.await();
            } catch (InterruptedException e) {
 
            }
        }
 
        zkClient.unsubscribeDataChanges(preNodePath,zkDataListener);
    }
 
    @Override
    public void releaseLock() {
        zkClient.delete(currentNodePath);
        zkClient.close();
    }
}

public class SingleRedisLock {
 
    JedisPool jedisPool = new JedisPool("192.168.200.128",6379);
 
    //锁过期时间
    protected long internalLockLeaseTime = 30000;
 
    //获取锁的超时时间
    private long timeout = 999999;
 
    /**
     * 加锁
     * @param lockKey 锁键
     * @param requestId 请求唯一标识
     * @return
     */
    SetParams setParams = SetParams.setParams().nx().px(internalLockLeaseTime);
 
    public boolean tryLock(String lockKey, String requestId){
 
        String threadName = Thread.currentThread().getName();
 
        Jedis jedis = this.jedisPool.getResource();
 
        Long start = System.currentTimeMillis();
 
        try{
            for (;;){
                String lockResult = jedis.set(lockKey, requestId, setParams);
                if ("OK".equals(lockResult)){
                    System.out.println(threadName+":   获取锁成功");
                    return true;
                }
                //否则循环等待，在timeout时间内仍未获取到锁，则获取失败
                System.out.println(threadName+":   获取锁失败，等待中");
                long l = System.currentTimeMillis() ‐ start;
                if (l>=timeout) {
                    return false;
                }
                try {
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }finally {
            jedis.close();
        }
 
    }
}

/**
     * 解锁
     * @param lockKey 锁键
     * @param requestId 请求唯一标识
     * @return
     */
public boolean releaseLock(String lockKey,String requestId){
 
    String threadName = Thread.currentThread().getName();
    System.out.println(threadName+"：释放锁");
    Jedis jedis = this.jedisPool.getResource();
 
    String lua =
        "if redis.call('get',KEYS[1]) == ARGV[1] then" +
        "   return redis.call('del',KEYS[1]) " +
        "else" +
        "   return 0 " +
        "end";
 
    try {
        Object result = jedis.eval(lua, Collections.singletonList(lockKey),
                                   Collections.singletonList(requestId));
        if("1".equals(result.toString())){
            return true;
        }
        return false;
    }finally {
        jedis.close();
    }
 
}

public class LoclTest {
 
    public static void main(String[] args) {
 
        //模拟多个5个客户端
        for (int i=0;i<5;i++) {
            Thread thread = new Thread(new LockRunnable());
            thread.start();
        }
    }
 
    private static class LockRunnable implements Runnable {
        @Override
        public void run() {
 
            SingleRedisLock singleRedisLock = new SingleRedisLock();
 
            String requestId = UUID.randomUUID().toString();
            boolean lockResult = singleRedisLock.tryLock("lock", requestId);
            if (lockResult){
 
                try {
                    TimeUnit.SECONDS.sleep(5);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
 
            singleRedisLock.releaseLock("lock",requestId);
        }
    }
}

<dependency>
    <groupId>org.apache.commons</groupId>
    <artifactId>commons‐pool2</artifactId>
</dependency>
<!‐‐Redis分布式锁‐‐>
<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson‐spring‐boot‐starter</artifactId>
    <version>3.13.1</version>
</dependency>

server:
  redis:
    host: 192.168.200.150
    port: 6379
    database: 0
    jedis:
      pool:
        max‐active: 500
        max‐idle: 1000
        min‐idle: 4

@Value("${spring.redis.host}")
private String host;
 
@Value("${spring.redis.port}")
private String port;
 
@Bean
public RedissonClient redissonClient(){
    RedissonClient redissonClient;
 
    Config config = new Config();
    String url = "redis://" + host + ":" + port;
    config.useSingleServer().setAddress(url);
 
    try {
        redissonClient = Redisson.create(config);
        return redissonClient;
    } catch (Exception e) {
        e.printStackTrace();
        return null;
    }
}

 @Component
public class RedissonLock {
 
    @Autowired
    private RedissonClient redissonClient;
 
    /**
     * 加锁
     * @param lockKey
     * @return
     */
    public boolean addLock(String lockKey){
 
        try {
            if (redissonClient == null){
                System.out.println("redisson client is null");
                return false;
            }
 
            RLock lock = redissonClient.getLock(lockKey);
 
            //设置锁超时时间为5秒，到期自动释放
            lock.lock(5, TimeUnit.SECONDS);
 
            System.out.println(Thread.currentThread().getName()+":  获取到锁");
 
            //加锁成功
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
    }
 
    public boolean releaseLock(String lockKey){
 
        try{
            if (redissonClient == null){
                System.out.println("redisson client is null");
                return false;
            }
 
            RLock lock = redissonClient.getLock(lockKey);
            lock.unlock();
            System.out.println(Thread.currentThread().getName()+":  释放锁");
            return true;
        }catch (Exception e){
            e.printStackTrace();
            return false;
        }
    }
}

@SpringBootTest
@RunWith(SpringRunner.class)
public class RedissonLockTest {
 
    @Autowired
    private RedissonLock redissonLock;
 
    @Test
    public void easyLock(){
        //模拟多个10个客户端
        for (int i=0;i<10;i++) {
            Thread thread = new Thread(new LockRunnable());
            thread.start();
        }
 
        try {
            System.in.read();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
 
    private class LockRunnable implements Runnable {
        @Override
        public void run() {
            redissonLock.addLock("demo");
            try {
                TimeUnit.SECONDS.sleep(3);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            redissonLock.releaseLock("demo");
        }
    }
}

@Configuration
public class RedissonRedLockConfig {
 
    public RedissonRedLock initRedissonClient(String lockKey){
 
        Config config1 = new Config();
        config1.useSingleServer().setAddress("redis://192.168.200.150:7000").setDatabase(0);
        RedissonClient redissonClient1 = Redisson.create(config1);
 
        Config config2 = new Config();
        config2.useSingleServer().setAddress("redis://192.168.200.150:7001").setDatabase(0);
        RedissonClient redissonClient2 = Redisson.create(config2);
 
        Config config3 = new Config();
        config3.useSingleServer().setAddress("redis://192.168.200.150:7002").setDatabase(0);
        RedissonClient redissonClient3 = Redisson.create(config3);
 
        Config config4 = new Config();
        config4.useSingleServer().setAddress("redis://192.168.200.150:7003").setDatabase(0);
        RedissonClient redissonClient4 = Redisson.create(config4);
 
        Config config5 = new Config();
        config5.useSingleServer().setAddress("redis://192.168.200.150:7004").setDatabase(0);
        RedissonClient redissonClient5 = Redisson.create(config5);
 
        RLock rLock1 = redissonClient1.getLock(lockKey);
        RLock rLock2 = redissonClient2.getLock(lockKey);
        RLock rLock3 = redissonClient3.getLock(lockKey);
        RLock rLock4 = redissonClient4.getLock(lockKey);
        RLock rLock5 = redissonClient5.getLock(lockKey);
 
        RedissonRedLock redissonRedLock = new 
RedissonRedLock(rLock1,rLock2,rLock3,rLock4,rLock5);
 
        return redissonRedLock;
    }
}

@SpringBootTest
@RunWith(SpringRunner.class)
public class RedLockTest {
 
    @Autowired
    private RedissonRedLockConfig redissonRedLockConfig;
 
    @Test
    public void easyLock(){
        //模拟多个10个客户端
        for (int i=0;i<10;i++) {
            Thread thread = new Thread(new RedLockTest.RedLockRunnable());
            thread.start();
        }
 
        try {
            System.in.read();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
 
    private class RedLockRunnable implements Runnable {
        @Override
        public void run() {
            RedissonRedLock redissonRedLock = redissonRedLockConfig.initRedissonClient("demo");
 
            try {
                boolean lockResult = redissonRedLock.tryLock(100, 10, TimeUnit.SECONDS);
 
                if (lockResult){
                    System.out.println("获取锁成功");
                    TimeUnit.SECONDS.sleep(3);
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }finally {
                redissonRedLock.unlock();
                System.out.println("释放锁");
            }
        }
    }
}