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

142536a

　　余量扣除，即在高并发，大用户下，每个用户的余量数据频繁发生变化。例如：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(&quot;anyMap&quot;);
String prevObject = map.put(&quot;123&quot;, new String());
String currentObject = map.putIfAbsent(&quot;323&quot;, new String());
String obj = map.remove(&quot;123&quot;);
map.fastPut(&quot;321&quot;, new String());
map.fastRemove(&quot;321&quot;);
Future<String> putAsyncFuture = map.putAsync(&quot;321&quot;);
Future<Void> fastPutAsyncFuture = map.fastPutAsync(&quot;321&quot;);
map.fastPutAsync(&quot;321&quot;, new String());
map.fastRemoveAsync(&quot;321&quot;);
redisson.shutdown();
*/
/**
* Distributed Object storage example
* Redisson redisson = Redisson.create();
RBucket<AnyObject> bucket = redisson.getBucket(&quot;anyObject&quot;);
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(&quot;anySet&quot;);
set.add(new SomeObject());
set.remove(new SomeObject());
set.addAsync(new SomeObject());
redisson.shutdown();
**/
final RAtomicLong atomicLong = redisson.getAtomicLong(&quot;anyAtomicLong&quot;);
//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(&quot; error less than 0&quot;);
return ;
}
long r=atomicLong.decrementAndGet();
System.out.println(&quot;get &quot;+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(&quot;redis.masterName&quot;);
String masteAddress=props.getProperty(&quot;redis.masteAddress&quot;);
int deployment_model=Integer.valueOf(props.getProperty(&quot;redis.deploymentModel&quot;)).intValue();
String hosts=props.getProperty(&quot;redis.hosts&quot;);
//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(&quot;,&quot;);
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(&quot;,&quot;);
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(&quot;127.0.0.1:7000&quot;, &quot;127.0.0.1:7001&quot;)
.setMasterConnectionPoolSize(10000)
.setSlaveConnectionPoolSize(10000);
Redisson redisson = Redisson.create(config);
return redisson;
}
}
　　

余量扣除代码片段2，扣除任意数&#20540;。
　　
　　

/**
* 扣取现金账本
* @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(&quot;Redisson is NULL&quot;);
return false;
}
String key=&quot;CASH_&quot;+actid;
String lock_point=&quot;LOCK_CASH_&quot;+actid;
RLock lock=redisson.getLock(lock_point);//获取账户锁对象
logger.info(&quot;get lock &quot;+lock_point);
boolean locked=lock.tryLock(10,60, TimeUnit.SECONDS);//尝试锁住账户对象,waitTime第一个参数获取锁超时时间30毫秒,leaseTime第二参数,锁自动释放时间
if(!locked) {
logger.info(&quot;cann't get lock ,id=&quot;+actid);
return false;
}
//lock.lock();
logger.info(&quot;get lock &quot;+lock_point+&quot; ok&quot;);
RBucket<Integer> atomicbalance = redisson.getBucket(key);//获取原子余量
boolean result_flag=true;
if(atomicbalance.get()==0) {
logger.error(&quot; error ,balance less than  or equal to 0&quot;);
result_flag=false;
}
else{
atomicbalance.set(atomicbalance.get().intValue()-amount);//扣除余量
logger.info(&quot;balance is  &quot;+atomicbalance.get());
result_flag=true;
}
lock.unlock();//解锁
logger.info(&quot;debut cash , cost time:&quot;+(System.currentTimeMillis()-start));
return result_flag;
}


　　

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




版权声明：本文为博主原创文章，未经博主允许不得转载。