针对一个账号进入存钱、取钱操作的并发控制，hibernate version乐观锁

这几天在练习针对一个账号进行存钱、取钱操作的并发控制。

 

entiy：Account

import javax.persistence.Column;
import javax.persistence.Entity;
import javax.persistence.Id;
import javax.persistence.Version;

import java.io.Serializable;

@Entity(name = "Account")
public class Account implements Serializable {

    private static final long serialVersionUID = 1L;

    @Id
    @Column(name = "AccountNumber")
    private String accountNumber;
    
   /* @Version
    @Column(name = "Version")
    private long version;*/
    
    @Column(name = "Status")
    private String status;
    
    @Column(name = "Score")
    private Double score;
    
    @Column(name = "Cas")
    private Long cas;

    
    public String getAccountNumber() {
        return accountNumber;
    }
    public void setAccountNumber(String accountNumber) {
        this.accountNumber = accountNumber;
    }
    public String getStatus() {
        return status;
    }

    public void setStatus(String status) {
        this.status = status;
    }

    public Double getScore() {
        return score;
    }

    public void setScore(Double score) {
        this.score = score;
    }

    public Long getCas() {
        return cas;
    }

    public void setCas(Long cas) {
        this.cas = cas;
    }    
}
    

 

 

dao：AccountDao

import org.springframework.stereotype.Repository;
import javax.inject.Inject;
import javax.persistence.FlushModeType;

@Repository
public class AccountDao {

    private ProductionJPAAccess productionJpaAccess;

    public Account get(String accountNumber) {
        return productionJpaAccess.get(Account.class, accountNumber);
    }

    public void save(Account account) {
        productionJpaAccess.save(account);
    }

    public void update(Account account) {
        //System.out.println(productionJpaAccess.getEntityManager().getFlushMode());
        //productionJpaAccess.getEntityManager().setFlushMode(FlushModeType.COMMIT);
        productionJpaAccess.update(account);
        productionJpaAccess.getEntityManager().flush();
    }
    
    @Inject
    public void setProductionJpaAccess(ProductionJPAAccess productionJpaAccess) {
        this.productionJpaAccess = productionJpaAccess;
    }
}

 

 

service: TradeService

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
import org.springframework.test.annotation.Rollback;
import org.springframework.transaction.annotation.Isolation;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;

import javax.inject.Inject;

@Service
public class TradeService {
    private final Logger logger = LoggerFactory.getLogger(TradeService.class);

    private AccountDao accountDao;

    private DataConverter dataConverter;

    @Inject
    private GenIdentityDao genIdentityDao;

    public Account get(String accountNumber) {
        return accountDao.get(accountNumber);
    }    
    
    @Transactional(value = "transactionManager",  propagation=Propagation.REQUIRED)    
    public synchronized void handleSimple(String accountNumber, double amount, boolean deposit) throws ResourceNotFoundException, NotEnoughException {
       // logger.error("{}, enter, {}", Thread.currentThread().getName(), this.hashCode());
        Account account = get(accountNumber);
        if (account == null) {
           throw new ResourceNotFoundException("account(" + accountNumber + ") not found");
        }

        double score = account.getScore();
        if (deposit) {
            account.setScore(convert(account.getScore() + amount));
        } else {
            if (account.getScore() < amount) {
                logger.error("--------------------" + Thread.currentThread().getName() + " " + accountNumber + ":" + score + (deposit ? "+" : "-") + amount + " failed");
               throw new NotEnoughException("account(" + accountNumber + ") not enough");
            }
            account.setScore(convert(account.getScore() - amount));

        }

        // account.setCas(account.getCas() + 1);
        accountDao.update(account);
        logger.error("-----" + Thread.currentThread().getName() + " " + accountNumber + ":" + score + (deposit ? "+" : "-") + amount + "=" + account.getScore());
    }

    
    public double convert(double value) {
        long lg = Math.round(value * 100);
        double d = lg / 100.0;
        return d;
    }

    @Transactional(value = "transactionManager")
    public void saveAccount(Account account) {
        accountDao.save(account);
    }

    @Inject
    public void setAccountDao(AccountDao accountDao) {
        this.accountDao = accountDao;
    }

    @Inject
    public void setDataConverter(DataConverter dataConverter) {
        this.dataConverter = dataConverter;
    }

}

 

 

用了多线程模拟多个请求，

task：MultipleTask

 

import java.util.Random;

public class MultipleTask implements Runnable {

    private TradeService tradeService;

    public MultipleTask(TradeService tradeService) {
        this.tradeService = tradeService;
    }

    @Override
    public void run() {
        String accountNumber = "SYS5";

        Random r = new Random();
        double amount = r.nextDouble() * 10;
        amount = convert(amount);
        boolean deposit = r.nextBoolean();

        try {
            Thread.sleep(r.nextInt(100));
        } catch (InterruptedException e) {
        }

        try {
            tradeService.handleSimple(accountNumber, amount, deposit);
        } catch (ResourceNotFoundException e) {
            e.printStackTrace();
        } catch (NotEnoughException e) {
            e.printStackTrace();
        }
    }

    public double convert(double value) {
        long lg = Math.round(value * 100);
        double d = lg / 100.0;
        return d;
    }

}

 

 

测试类：

 

@Test
    public void testMultiple() throws Exception {
       

        for (int i = 0; i < 30; i++) {
            MultipleTask m = new MultipleTask(tradeService);
            new Thread(m, i + "").start();

        }

        Thread.sleep(1000000000);
    }

 

特别要注意：

1.一定要在accountDao.update()方法的最后加上立即flush的语句，防止因没及时flush，其他线程访问到提交前的数据。

productionJpaAccess.getEntityManager().flush();

 

2.一定要在handleSimple方法里重新从数据库load对象，判断是否存在（已被别的线程删除？）、是否足够（已被别的线程取掉了一些钱？），然后更新。即让重新load，各种判断在synchronized管制之内。

如果这几个判断在synchronized管制之外，有可能A线程在判断通过后，等待B线程执行synchronized代码，然后A进入synchronized代码时，数据已经被改变了，即原来的判断就不正确了。

 

3.在test方法中加上 Thread.sleep(1000000000);是为了防止主线程停止后spring容器被关闭。

 

 

 测试结果：

2014-04-24 13:35:22,399 [2] ERROR TradeService - -----2 SYS5:2.5+1.11=3.61
2014-04-24 13:35:22,413 [29] ERROR TradeService - -----29 SYS5:3.61-1.12=2.49
2014-04-24 13:35:22,414 [15] ERROR TradeService - --------------------15 SYS5:2.49-9.33 failed
2014-04-24 13:35:22,417 [27] ERROR TradeService - -----27 SYS5:2.49+4.06=6.55
2014-04-24 13:35:22,420 [23] ERROR TradeService - -----23 SYS5:6.55+1.61=8.16
2014-04-24 13:35:22,424 [17] ERROR TradeService - -----17 SYS5:8.16+3.82=11.98
2014-04-24 13:35:22,426 [7] ERROR TradeService - -----7 SYS5:11.98+9.48=21.46
2014-04-24 13:35:22,431 [21] ERROR TradeService - -----21 SYS5:21.46+2.91=24.37
2014-04-24 13:35:22,434 [5] ERROR TradeService - -----5 SYS5:24.37-4.36=20.01
2014-04-24 13:35:22,436 [19] ERROR TradeService - -----19 SYS5:20.01+2.95=22.96
2014-04-24 13:35:22,439 [3] ERROR TradeService - -----3 SYS5:22.96-2.72=20.24
2014-04-24 13:35:22,442 [11] ERROR TradeService - -----11 SYS5:20.24-2.25=17.99
2014-04-24 13:35:22,444 [25] ERROR TradeService - -----25 SYS5:17.99+3.97=21.96
2014-04-24 13:35:22,447 [13] ERROR TradeService - -----13 SYS5:21.96+6.31=28.27
2014-04-24 13:35:22,450 [9] ERROR TradeService - -----9 SYS5:28.27+3.7=31.97
2014-04-24 13:35:22,454 [28] ERROR TradeService - -----28 SYS5:31.97-4.84=27.13
2014-04-24 13:35:22,460 [26] ERROR TradeService - -----26 SYS5:27.13+2.29=29.42
2014-04-24 13:35:22,464 [8] ERROR TradeService - -----8 SYS5:29.42-1.73=27.69
2014-04-24 13:35:22,466 [0] ERROR TradeService - -----0 SYS5:27.69-9.4=18.29
2014-04-24 13:35:22,469 [16] ERROR TradeService - -----16 SYS5:18.29+2.03=20.32
2014-04-24 13:35:22,472 [22] ERROR TradeService - -----22 SYS5:20.32+2.58=22.9
2014-04-24 13:35:22,475 [24] ERROR TradeService - -----24 SYS5:22.9-3.39=19.51
2014-04-24 13:35:22,477 [1] ERROR TradeService - -----1 SYS5:19.51+2.27=21.78
2014-04-24 13:35:22,480 [14] ERROR TradeService - -----14 SYS5:21.78-8.68=13.1
2014-04-24 13:35:22,485 [10] ERROR TradeService - -----10 SYS5:13.1-2.81=10.29
2014-04-24 13:35:22,488 [18] ERROR TradeService - -----18 SYS5:10.29-1.59=8.7
2014-04-24 13:35:22,491 [4] ERROR TradeService - -----4 SYS5:8.7-2.44=6.26
2014-04-24 13:35:22,523 [12] ERROR TradeService - --------------------12 SYS5:6.26-6.56 failed
2014-04-24 13:35:22,526 [20] ERROR TradeService - -----20 SYS5:6.26+6.03=12.29
2014-04-24 13:35:22,532 [6] ERROR TradeService - -----6 SYS5:12.29+7.46=19.75

 

 

以上是自己实现的同步控制，适于只部署一台服务器。

如果是分布式环境，多个tomcat，用synchronized就没法控制了。

这时候就需要使用hibernate version，乐观锁。

 

hiberate里使用乐观锁很简单，在entity上加一个long类型的属性version,加上@Version，get,set。。

@Version
@Column(name = "Version")
 private long version;

 

然后就可以正常的save,update了。

在更新时会自动加上 , version = ?， update XX set XXXX where id=?, version = ?

如果记录已经被更新了，version就会加1，就找不到这条记录更新不了了，会抛出OptimisticLockException异常。

 

 

 

 

@Test
    public void testVersion() {
        Account a = new Account();
        a.setAccountNumber("SYS7");
        a.setScore(10d);
        tradeService.saveAccount(a);

       // a = tradeService.get("SYS7");
        System.out.println(a.getVersion());

        for (int i = 0; i < 5; i++) {
            a.setScore(a.getScore() + 1);
            tradeService.update(a);
            
            //a = tradeService.get("SYS7");
            System.out.println(a.getVersion());
        }    
    }

 以上代码会抛出javax.persistence.OptimisticLockException异常，因为update后，数据库的version已经更新了，但a还没更新。所以得重新get出来。

 

@Test
    public void testVersion() {
        Account a = new Account();
        a.setAccountNumber("SYS7");
        a.setScore(10d);
        tradeService.saveAccount(a);

        a = tradeService.get("SYS7");
        System.out.println(a.getVersion());

        for (int i = 0; i < 2; i++) {
            a.setScore(11d);
            tradeService.update(a);
            
            a = tradeService.get("SYS7");
            System.out.println(a.getVersion());
        }    
    }

 以上代码打印结果是

0

1

1

因为2次都是把score set成11d，其实值没变，update就没真正执行，也就不会使version增加。

 

 

@Test
    public void testVersion() {
        Account a = new Account();
        a.setAccountNumber("SYS7");
        a.setScore(10d);
        tradeService.saveAccount(a);

        a = tradeService.get("SYS7");
        System.out.println(a.getVersion());

        for (int i = 0; i < 2; i++) {
            a.setScore(a.getScore() + 1);
            tradeService.update(a);
            
            a = tradeService.get("SYS7");
            System.out.println(a.getVersion());
        }    
    }

以上代码打印

0

1

2

是正常的。