数据库分布式锁

定时任务的数据库分布式锁

问题：当使用乐观锁的version来判断时，对于定时任务来说有可能因为系统中各个实例系统误差较大，这样会出现如下情况
A实例更新完数据了，B实例才获取数据。这个时候B肯定也会执行一边任务，此时就会被执行两次

一个唯一键，时间看成version

-- create table `account`
# DROP TABLE `account` IF EXISTS
CREATE TABLE `account` (
  `id` int(11) NOT NULL AUTO_INCREMENT,
  `name` varchar(20) NOT NULL unique ,
  `money` double DEFAULT NULL,
  `create_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=4 DEFAULT CHARSET=utf8;
INSERT INTO `account` VALUES ('1', 'aaa', '1000', now());
INSERT INTO `account` VALUES ('2', 'bbb', '1000', now());
INSERT INTO `account` VALUES ('3', 'ccc', '1000', now());

package com.forezp.service;

import com.forezp.dao.AccountMapper;
import com.forezp.entity.Account;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.dao.DuplicateKeyException;
import org.springframework.stereotype.Service;

import java.util.Date;
import java.util.List;
import java.util.Objects;
import java.util.Random;

/**
 * Created by fangzhipeng on 2017/4/20.
 */
@Service
public class AccountService {
    private static final int TIME_DIFF = 1000 * 1; //1s
    @Autowired
    private AccountMapper accountMapper;

    public int add(String name, double money) {
        try{
            return accountMapper.add(name, money, new Date());
        }catch (DuplicateKeyException e){
            System.out.println("AccountService.add error" + e.getMessage());
            return 0;
        }
    }
    public int update(String name, double money, int id) {
        return accountMapper.update(name, money, new Date(), id);
    }
    public int delete(int id) {
        return accountMapper.delete(id);
    }
    public Account findAccount(int id) {
        return accountMapper.findAccount(id);
    }
    public List<Account> findAccountList() {
        return accountMapper.findAccountList();
    }
    public Account findAccountOrderTime(String name) { return accountMapper.findAccountOrderTime(name);}
    public boolean updateAccountByTime(Account account, Date nowDate) {
        return 1 == accountMapper.updateAccountByTime(account.getName(), account.getCreateTime(), account.getId(), nowDate);
    }
    
    public void testDbLock(String name){
        System.out.println("start -------- " + name);
        for (int i = 0 ; i < 10; ++i) {
            Thread thread = new Thread(() -> doSome(name));
            thread.setName("testDbLock[" + i + "] ");
            thread.start();
        }
    }


    /**
     * 多个实例之间可能出现时间不一致的情况，取在1s的时间误差
     * @param oldTime
     * @param nowTime
     * @return
     */
    private boolean isSame(long oldTime, long nowTime){
        return Math.abs(oldTime - nowTime) <= TIME_DIFF;
    }

    /**
     * 数据库分布式锁
     * @param name
     */
    private void doSome(String name){
        // 查寻
        Account account = findAccountOrderTime(name);

        if (!Objects.isNull(account)){
            // 数据库中已经有纪律
            account.setMoney(new Random().nextDouble());
            // 老的时间
            long oldTime = account.getCreateTime().getTime();
            // 当前时间
            Date date = new Date();
            long nowTime = date.getTime();
            System.out.println(Thread.currentThread().getName() + " nowTime " + nowTime + " oldTime " + oldTime + " abs= " + Math.abs(oldTime - nowTime));
            //判断当前时间下是否已经更新了
            if (isSame(oldTime, nowTime)){
                System.out.println("判断当前时间下其他的实例已经更新了，请检查各个实例之间的时间误差 isSame " + Thread.currentThread().getName());
                return;
            }
            if (!updateAccountByTime(account, date)){
                // 当前实例没更新成功
                System.out.println("当前实例没更新成功 updateAccountByTime " + Thread.currentThread().getName());
                return;
            }
        }else{
            // 没有记录那就插入了
            if (0 == add(name, new Random().nextDouble())){
                // 当前实例没更新成功
                System.out.println("当前实例没更新成功 add" + Thread.currentThread().getName());
                return;
            }
        }

        System.out.println("更新成功 " + Thread.currentThread().getName());
        System.out.println("start " + Thread.currentThread().getName());
        try{
            Thread.sleep(1000);
        }catch (Exception e){
            e.printStackTrace();
        }
        System.out.println("end " + Thread.currentThread().getName());
    }

}

package com.forezp.dao;

import com.forezp.entity.Account;
import org.apache.ibatis.annotations.*;

import java.util.Date;
import java.util.List;

/**
 * Created by fangzhipeng on 2017/4/20.
 */
@Mapper
public interface AccountMapper {

    @Insert("insert into account(name, money, create_time) values(#{name}, #{money}, #{createTime})")
    int add(@Param("name") String name, @Param("money") double money, @Param("createTime") Date createTime);

    @Update("update account set name = #{name}, money = #{money}, create_time = #{createTime} where id = #{id}")
    int update(@Param("name") String name, @Param("money") double money, @Param("createTime") Date createTime, @Param("id") int  id);

    @Delete("delete from account where id = #{id}")
    int delete(int id);

    @Select("select id, name as name, money as money, create_time as createTime from account where id = #{id}")
    Account findAccount(@Param("id") int id);


    @Select("select id, name as name, money as money, create_time as createTime from account where `name` = #{name} order by `create_time` desc limit 1")
    Account findAccountOrderTime(@Param("name") String name);

    @Update("update account set name = #{name}, create_time = #{createTime} "
            + "where id = #{id} and `create_time` = #{oldTime} and `name` = #{name}")
    int updateAccountByTime(@Param("name") String name,
                            @Param("oldTime") Date oldTime, @Param("id") int id, @Param("createTime") Date createTime);


    @Select("select id, name as name, money as money ,create_time as createTime from account")
    List<Account> findAccountList();
}

package com.forezp;

import com.forezp.entity.Account;
import com.forezp.service.AccountService;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

import javax.annotation.Resource;
import java.util.Date;

@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringbootMybatisApplicationTests {

	@Test
	public void contextLoads() {
	}



	@Resource
	private AccountService accountService;



	@Test
	public void testUniq(){
//		int aaa = accountService.add("aaa", 1234);
//		System.out.println(aaa);


		System.out.println(new Date().getTime());
		System.out.println(new Date().getTime());
		accountService.testDbLock("aaabnb");
		try{
			Thread.sleep(1000 * 10);
		}catch (Exception e){
			e.printStackTrace();
		}
	}
}