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多线程压力测试扣除余量操作,最终扣除的余量正确。