redis分布式内存锁：余量扣除示例

余量扣除，即在高并发，大用户下，每个用户的余量数据频繁发生变化。例如：12306的某车次票的余量，商品库存，短信余量账本等。

针对，此类频繁发生修改的原子类余量对象，采用mysql,oracle等数据，一定会存在操作瓶颈。本文拟采用内存的办法实现，使用redis+Redisson客户端完成。当然，或许可以采用mangodb这类no-sql数据库。

Redisson客户端

https://github.com/mrniko/redisson/wiki

实现redis分布锁的客户端开源项目，redission支持4中连接redis方式，分别为单机，主从， Sentinel ， Cluster 集群，并提供以下类库

1.AtomicLong原子操作

2.分布式List

3.分布式Set

4.分布式Map

5.分布式Queue,

6.分布式SortedSet,

7.分布式ConcureentMap

8.分布式Lock

9.分布式CountDownLatch

10. 分布式Publish / Subscribe, HyperLogLog等

 

余量扣除代码片段

resources/redis.properties

 

#redis部署模式 SingleHost 1 MasterSlave 2 Sentinel 3  Cluster 4
redis.deploymentModel=2
redis.hosts=192.168.161.73:6379,192.168.161.129:6379
redis.masterName=mymaster
redis.masteAddress=192.168.161.73:6379

 

RedissonClient

 

import java.io.FileNotFoundException;
import java.io.IOException;
import java.util.Properties;
import java.util.concurrent.Future;

import org.redisson.Config;
import org.redisson.Redisson;
import org.redisson.connection.RandomLoadBalancer;
import org.redisson.core.RAtomicLong;
import org.redisson.core.RMap;


/**
 *分布式锁客户端
 * @author 
 *
 */

public class RedissonClient {
	private static RedissonClient instance;
	private RedissonClient(String filename) throws FileNotFoundException, IOException{
		init(filename);
 	}
	public static synchronized RedissonClient getInstance(String filename){
		if(instance==null){
			try {
				 instance=new  RedissonClient(filename);
			 } catch (FileNotFoundException e) {
				// TODO Auto-generated catch block
  				e.printStackTrace();
			} catch (IOException e) {
				 // TODO Auto-generated catch block
				 e.printStackTrace();
			}
 		}
 		return instance;	
	 }
	 public static void main(String[] args){
		RedissonClient client= RedissonClient.getInstance("resources/redis.properties");
 		Redisson redisson=client.getSingleClient("ip:6379");
		/*
		RMap<String, String> map = redisson.getMap("anyMap");
		String prevObject = map.put("123", new String());
		String currentObject = map.putIfAbsent("323", new String());
		String obj = map.remove("123");

		map.fastPut("321", new String());
		map.fastRemove("321");

		Future<String> putAsyncFuture = map.putAsync("321");
		Future<Void> fastPutAsyncFuture = map.fastPutAsync("321");

		map.fastPutAsync("321", new String());
		map.fastRemoveAsync("321");
		redisson.shutdown();
		*/
		
		/**
		 * Distributed Object storage example
		 * Redisson redisson = Redisson.create();

			RBucket<AnyObject> bucket = redisson.getBucket("anyObject");
			bucket.set(new AnyObject());
			bucket.setAsync(new AnyObject());
			AnyObject obj = bucket.get();
			
			redisson.shutdown();
		 */
		
		/**Distributed Set example
		 Redisson redisson = Redisson.create();

 		RSet<SomeObject> set = redisson.getSet("anySet");
		set.add(new SomeObject());
		set.remove(new SomeObject());

		set.addAsync(new SomeObject());

		redisson.shutdown();
		**/
                final RAtomicLong atomicLong = redisson.getAtomicLong("anyAtomicLong");
		//atomicLong.set(1000);//初始化余量为1000，可以通过redis linux客户端设置1000
		for(int i=0;i<2000;i++){//开始扣减余量，每次扣1.共计扣2000次。
			new Runnable(){
	
				@Override
				public void run() {
					
					try {
						Thread.currentThread().sleep(5);
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
					if(atomicLong.get()==0) {//当余量为0时，系统提示余量不够并退出
						System.out.println(" error less than 0");
						return ;
					}
					long r=atomicLong.decrementAndGet();
					System.out.println("get "+r);
					
				}
				
			}.run();;
	    }
		
	
	}
	private static Redisson redisson=null;
	public Redisson  getClient(){
		return  redisson;
	}
	public  void  init(String filename) throws FileNotFoundException, IOException
	{
		//RedissonClient client=new RedissonClient();
		PropertyReader propReader=PropertyReader.getInstance(filename);
		Properties props=propReader.getProperties(filename);
		String masterName=props.getProperty("redis.masterName");
		String masteAddress=props.getProperty("redis.masteAddress");
		int deployment_model=Integer.valueOf(props.getProperty("redis.deploymentModel")).intValue();
		String hosts=props.getProperty("redis.hosts");
		//redis部署模式 SingleHost 1 MasterSlave 2 Sentinel 3  Cluster 4 
		switch(deployment_model){
			case 1:
				redisson=getSingleClient(hosts);//单机
				break;
			case 2:
				redisson=getMasterSlaveClient(masteAddress,hosts);
				break;
			case 3:
				redisson=getSentinelClient(masterName,hosts);
				break;
			case 4:
				redisson=getClusterClient(hosts);
				break;
		}
		
	}
	public Redisson getSingleClient(String host){
		//Single server connection:
		// connects to default Redis server 127.0.0.1:6379
		//edisson redisson = Redisson.create();
		
		// connects to single Redis server via Config
		Config config = new Config();
		    config.useSingleServer()
		          .setAddress(host)
		          .setConnectionPoolSize(1000)
		          ;

		Redisson redisson = Redisson.create(config);
		return redisson;
	}
	//Master/Slave servers connection:
	public Redisson getMasterSlaveClient(String add,String hosts){
		Config config = new Config();
		String[] hostarr=hosts.split(",");
		config.useMasterSlaveConnection()
		    .setMasterAddress(add)
		    .setLoadBalancer(new RandomLoadBalancer()) // RoundRobinLoadBalancer used by default
		   
		    .addSlaveAddress(hostarr)
		    .setMasterConnectionPoolSize(10000)
		    .setSlaveConnectionPoolSize(10000);

		Redisson redisson = Redisson.create(config);
		return redisson;
	}
	//Sentinel servers connection:
	public Redisson getSentinelClient(String masterName,String hosts){
		String[] hostarr=hosts.split(",");
		Config config = new Config();
		config.useSentinelConnection()
		    .setMasterName(masterName)
		    .addSentinelAddress(hostarr)
		    .setMasterConnectionPoolSize(10000)
		    .setSlaveConnectionPoolSize(10000);

		Redisson redisson = Redisson.create(config);
		return redisson;
	}
	//Cluster nodes connections:
	public Redisson getClusterClient(String hosts){
		Config config = new Config();
		config.useClusterServers()
		    .setScanInterval(2000) // sets cluster state scan interval
		    .addNodeAddress("127.0.0.1:7000", "127.0.0.1:7001")
		    .setMasterConnectionPoolSize(10000)
		    .setSlaveConnectionPoolSize(10000);

		Redisson redisson = Redisson.create(config);
		return redisson;
	}
}

 

 

余量扣除代码片段2，扣除任意数值。

 

 

/**
	 * 扣取现金账本
	 * @param actid  账户id
	  * @return
	  * @throws FileNotFoundException
	 * @throws IOException
	  * @throws InterruptedException 
	 */
	private boolean deductCashAccount(String actid,int amount) throws FileNotFoundException, IOException, InterruptedException{
		/**扣余额直接操作redis缓存数据库，key由账户ID，-字符，字符串balance组成*/
		 long start=System.currentTimeMillis();
		 if(redisson==null) {
			 logger.error("Redisson is NULL");
			 return false;
		}
		String key="CASH_"+actid;
		String lock_point="LOCK_CASH_"+actid;
		RLock lock=redisson.getLock(lock_point);//获取账户锁对象
		logger.info("get lock "+lock_point);
		boolean locked=lock.tryLock(10,60, TimeUnit.SECONDS);//尝试锁住账户对象,waitTime第一个参数获取锁超时时间30毫秒,leaseTime第二参数,锁自动释放时间
		if(!locked) {
			logger.info("cann't get lock ,id="+actid);
			return false;
		}
		//lock.lock();
		logger.info("get lock "+lock_point+" ok");
		RBucket<Integer> atomicbalance = redisson.getBucket(key);//获取原子余量
		boolean result_flag=true;
		if(atomicbalance.get()==0) {
			logger.error(" error ,balance less than  or equal to 0");
			
			result_flag=false;
		}
		else{
			atomicbalance.set(atomicbalance.get().intValue()-amount);//扣除余量
			logger.info("balance is  "+atomicbalance.get());
			result_flag=true;
		}
		lock.unlock();//解锁
		 logger.info("debut cash , cost time:"+(System.currentTimeMillis()-start));
		return result_flag;
	}

 

实验结论：

本人在一项目中，使用以上两个分布式锁对象，进行了ab压力测试。设置初始余量，将以前代码片，发布为http api服务,ab多线程压力测试扣除余量操作，最终扣除的余量正确。

评论

1 楼 newboy2004 2015-07-30  

这样搞的话，你redis里的数据和数据库里的数据 什么时候做同步呢？而且还有你这个扣除余量的这个数据，是都从数据库加载到这个redis缓存里吗？还是直接就不存在数据库，一开始所有的这些东西都存在于redis中，请指点，谢谢