Fork me on GitHub

乐观锁与悲观锁的应用

发表于 | 分类于 | 阅读次数:
字数统计: 2,372 字 | 阅读时长 ≈ 10 分钟

乐观锁与悲观锁的应用

概念

悲观锁(Pessimistic Lock)

每次去拿数据的时候都认为别人会修改，所以每次在拿数据的时候都会上锁，这样别人想拿这个数据就会block直到它拿到锁。传统的关系型数据库里边就用到了很多这种锁机制，比如行锁，表锁等，读锁，写锁等，都是在做操作之前先上锁。

悲观并发控制主要用于数据争用激烈的环境，以及发生并发冲突时使用锁保护数据的成本要低于回滚事务的成本的环境中。

乐观锁(Optimistic Lock)

每次去拿数据的时候都认为别人不会修改，所以不会上锁，但是在更新的时候会判断一下在此期间别人有没有去更新这个数据，可以使用版本号等机制。乐观锁适用于多读的应用类型，这样可以提高吞吐量，像数据库如果提供类似于write_condition机制的其实都是提供的乐观锁。

实例

假设一个业务场景：数据库中有一条数据，需要获取到当前的值，在当前值的基础上+10，然后再更新回去。
如果此时有两个线程同时并发处理，第一个线程拿到数据是10，+10=20更新回去。第二个线程原本是要在第一个线程的基础上再+20=40,结果由于并发访问取到更新前的数据为10，+20=30。

这就是典型的存在中间状态，导致数据不正确。来看以下的例子：

并发所带来的问题

和上文提到的类似，这里有一张price表，表结构如下：

1
2
3
4
5
6
7
CREATE TABLE `price` (
  `id` int(11) NOT NULL AUTO_INCREMENT COMMENT '主键',
  `total` decimal(12,2) DEFAULT '0.00' COMMENT '总值',
  `front` decimal(12,2) DEFAULT '0.00' COMMENT '消费前',
  `end` decimal(12,2) DEFAULT '0.00' COMMENT '消费后',
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=1268 DEFAULT CHARSET=utf8

我这里写了一个单测：就一个主线程，循环100次，每次把 front 的值减去10，再写入一次流水记录，正常情况是写入的每条记录都会每次减去10。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
  /**
   * 单线程消费
   */
  @Test
  public void singleCounsumerTest1(){
// 使用 Mybaits 
      /*  for (int i=0 ;i<100 ;i++){
          Price price = priceMapper.selectByPrimaryKey(1);
          int ron = 10 ;
          price.setFront(price.getFront().subtract(new BigDecimal(ron)));
          price.setEnd(price.getEnd().add(new BigDecimal(ron)));
          price.setTotal(price.getFront().add(price.getEnd()));
          priceMapper.updateByPrimaryKey(price) ;
          price.setId(null);
          priceMapper.insertSelective(price) ;
      }*/
      
      // 使用 hibernate
      for (int i=0 ;i<100 ;i++){
          Price price = priceDao.selectByPrimaryKey("1");
          int ron = 10 ;
          price.setFront(price.getFront().subtract(new BigDecimal(ron)));
          price.setEnd(price.getEnd().add(new BigDecimal(ron)));
          price.setTotal(price.getFront().add(price.getEnd()));
          priceDao.updateByPrimaryKey(price) ;
         // price.setId(null);
          priceDao.insertSelective(price) ;
      }
      
  }

初始化数据库中的值：

mark

执行结果如下：

mark

但是如果是多线程的情况下会是如何呢：

我这里新建了一个 PriceController

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
/**
  * @Author shenwenfang
  * @Date 2018/3/8 10:50
  * @Description: 线程池 无锁
  */
@RestController
@RequestMapping(value="threadPrice")
public class ThreadPoolConfigController {
    @Autowired
    PriceDao priceDao;
    @Autowired
    ThreadPoolConfig config;
    @RequestMapping(value = "/threadPrice.do")
    @ApiOperation(value = "锁的应用",httpMethod = "POST")
    public void threadPrice(){
        try {
            for(int i= 0;i <10;i++){
                Thread t = new Thread(new Runnable() {
                    public void run() {
                        Price price = priceDao.selectByPrimaryKey("1");
                        int ron = 10;
                        price.setFront(price.getFront().subtract(new BigDecimal(ron)));
                        price.setEnd(price.getEnd().add(new BigDecimal(ron)));
                        priceDao.updateByPrimaryKey(price);
                        priceDao.insertSelective(price);
                    }
                });
                config.submit(t);
            }
        }catch (Exception e){
            e.printStackTrace();
        }
    }
}

其中为了节省资源使用了一个线程池:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
import java.util.concurrent.TimeUnit;
/**
  * @Author shenwenfang
  * @Date 2018/3/8 10:40
  * @Description: 为了节省资源使用一个线程池
  */
@Service
public class ThreadPoolConfig {
    private static final int MAX_SIZE = 10;
    private static final int CORE_SIZE = 5;
    private static final int SECOND = 1000;
    private ThreadPoolExecutor executor;
    public ThreadPoolConfig(){
        executor = new ThreadPoolExecutor(CORE_SIZE,MAX_SIZE,SECOND,TimeUnit.MICROSECONDS,
                new LinkedBlockingQueue<Runnable>());
    }
    public void submit(Thread thread){
        executor.submit(thread);
    }
}

关于线程池的使用今后会仔细探讨。这里就简单理解为有10个线程并发去处理上面单线程的逻辑，来看看结果怎么样：

mark

会看到明显的数据错误，导致错误的原因自然就是有线程读取到了中间状态进行了错误的更新。

进而有了以下两种解决方案：悲观锁和乐观锁。

悲观锁

简单理解下悲观锁：当一个事务锁定了一些数据之后，只有当当前锁提交了事务，释放了锁，其他事务才能获得锁并执行操作。

使用方式如下：
首先要关闭 MySQL 的自动提交：set autocommit = 0;

1
2
3
4
5
6
bigen --开启事务
select id, total, front, end from price where id=1 for update 
insert into price values(?,?,?,?,?)
commit --提交事务

这里使用select for update的方式利用数据库开启了悲观锁，锁定了id=1的这条数据(注意:这里除非是使用了索引会启用行级锁，不然是会使用表锁，将整张表都锁住。)。之后使用commit提交事务并释放锁，这样下一个线程过来拿到的就是正确的数据。

悲观锁一般是用于并发不是很高，并且不允许脏读等情况。但是对数据库资源消耗较大。

优点与不足

悲观并发控制实际上是“先取锁再访问”的保守策略，为数据处理的安全提供了保证。但是在效率方面，处理加锁的机制会让数据库产生额外的开销，还有增加产生死锁的机会；另外，在只读型事务处理中由于不会产生冲突，也没必要使用锁，这样做只能增加系统负载；还有会降低了并行性，一个事务如果锁定了某行数据，其他事务就必须等待该事务处理完才可以处理那行数

乐观锁

那么有没有性能好，支持的并发也更多的方式呢？

那就是乐观锁。

乐观锁是首先假设数据冲突很少，只有在数据提交修改的时候才进行校验，如果冲突了则不会进行更新。

通常的实现方式增加一个version字段，为每一条数据加上版本。每次更新的时候version+1，并且更新时候带上版本号。实现方式如下：

新建了一张price_version表：

1
2
3
4
5
6
7
8
CREATE TABLE `price_version` (
  `id` int(11) NOT NULL AUTO_INCREMENT COMMENT '主键',
  `total` decimal(12,2) DEFAULT '0.00' COMMENT '总值',
  `front` decimal(12,2) DEFAULT '0.00' COMMENT '消费前',
  `end` decimal(12,2) DEFAULT '0.00' COMMENT '消费后',
  `version` int(11) DEFAULT '0' COMMENT '并发版本控制',
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=1268 DEFAULT CHARSET=utf8

更新数据的SQL：

1
2
3
4
5
6
7
8
<!-- Mubatis -->
<update id="updateByVersion" parameterType="com.crossoverJie.pojo.PriceVersion">
    UPDATE price_version
    SET front = #{front,jdbcType=DECIMAL},
        version= version + 1
    WHERE id = #{id,jdbcType=INTEGER}
    AND version = #{version,jdbcType=INTEGER}
  </update>
1
2
3
4
5
6
7
8
9
10
11
12
@Transactional
public int updateByPrimaryKey(PriceVersion priceVersion){
    String sql = "UPDATE price_version\n" +
            "SET front = :front, version= version + 1\n" +
            "WHERE id = :id AND version = :version";
    Map<String,Object> param = new HashMap<String,Object>();
    param.put("front",priceVersion.getFront());
    param.put("id",priceVersion.getId());
    param.put("version",priceVersion.getVersion());
    int count =                                                                getCurrentSession().createSQLQuery(sql).setProperties(param).executeUpdate();
    return count;
}

调用方式：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
/**
 * 线程池，乐观锁
 * @param redisContentReq
 * @return
 */
@RequestMapping(value = "/threadPriceVersion.do")
@ApiOperation(value = "乐观锁的应用",httpMethod = "POST")
public void threadPriceVersion(){
    for(int i = 0;i<3;i++){
        Thread t = new Thread(new Runnable() {
            public void run() {
                PriceVersion priceVersion = goodStoreDao.vselectByPrimaryKey("1");
                int ron = new Random().nextInt(20);
                System.out.println("本次消费="+ron);
                priceVersion.setFront(new BigDecimal(ron));
                int count = goodStoreDao.vupdateByPrimaryKey(priceVersion);
                if(count == 0)
                    System.out.println("更新失败！");
                else
                    System.out.println("更新成功！");
            }
        });
        config.submit(t);
    }
}

处理逻辑：开了三个线程生成了20以内的随机数更新到 front 字段。

当调用该接口时日志如下：

mark

可以看到线程1、4、5分别生成了15，2，11三个随机数。最后线程4、5都更新失败了，只有线程1更新成功了。

查看数据库：

mark

发现也确实是更新的6。

乐观锁在实际应用相对较多，它可以提供更好的并发访问，并且数据库开销较少，但是有可能存在脏读的情况。

优点与不足

乐观并发控制相信事务之间的数据竞争(data race)的概率是比较小的，因此尽可能直接做下去，直到提交的时候才去锁定，所以不会产生任何锁和死锁。但如果直接简单这么做，还是有可能会遇到不可预期的结果，例如两个事务都读取了数据库的某一行，经过修改以后写回数据库，这时就遇到了问题。

补充

1.我们经常会在访问数据库的时候用到锁，怎么实现乐观锁和悲观锁呢？以Hibernate为例，可以通过为记录添加版本或时间戳字段来实现乐观锁。可以用session.Lock()锁定对象来实现悲观锁（本质上就是执行了SELECT * FROM t FOR UPDATE语句）。

2.如果把乐观锁看作是关于冲突检测的，那么悲观锁就是关于冲突避免的。在实际应用的源代码控制系统中，

这两种策略都可以被使用，但是现在大多数源代码开发者更倾向于使用乐观锁策略。（有一种很有道理的说法：乐观锁并不是真正的锁定，但是这种叫法很方便并且广泛流传，以至于不容忽略。）

在乐观锁和悲观锁之间进行选择的标准是：冲突的频率与严重性。如果冲突很少，或者冲突的后果不会很严重，那么通常情况下应该选择乐观锁，因为它能得到更好的并发性，而且更容易实现。但是，如果冲突的结果对于用户来说痛苦的，那么就需要使用悲观策略。

坚持原创技术分享，您的支持将鼓励我继续创作！