Redis 应用分析

 1 /**
 2      * 获取分布式锁 DLock
 3      * @param key 锁标志
 4      * @param holder 锁持有者（一般为uuid，唯一标识当前锁的持有者）
 5      * @param timeout 获取锁的超时时间 毫秒
 6      * @return
 7      */
 8     public static boolean getDLock(String key, String holder, long timeout){
 9         Jedis jedis = DRJedis.getJedisPool();
10         //设置获取锁的超时时间
11         long end = System.currentTimeMillis() + timeout;
12         while (System.currentTimeMillis() < end){
13             if (jedis.setnx(LOCK_PREFIX + key, holder) == 1) return true;
14             try {
15                 Thread.sleep(10);
16             } catch (InterruptedException e) {
17                 log.error("获取 DLock 异常", e);
18                 return false;
19             }
20         }
21         log.error("获取 DLock 超时");
22         return false;
23     }

 1 /**
 2      * 释放锁
 3      * @param key 锁标志
 4      * @param holder 锁持有者（一般为uuid，唯一标识当前锁的持有者）
 5      */
 6     public static void releaseDLock(String key, String holder){
 7         Jedis jedis = DRJedis.getJedisPool();
 8         while (true){
 9             try {
10                 jedis.watch(LOCK_PREFIX + key);
11                 String value = jedis.get(LOCK_PREFIX + key);
12                 if(value != null && value.equals(holder)){// 保证释放的是当前持有者的锁
13                     Transaction tran = jedis.multi();
14                     jedis.del(LOCK_PREFIX + key);
15                     tran.exec();
16                 }
17                 jedis.unwatch();
18                 break;
19             }catch (Exception e){
20                 log.error("释放锁异常", e);
21                 break;
22             }
23         }
24     }

 1 /**
 2      * 生产消息
 3      * @param key 消息key
 4      * @param message 消息体，支持批量生产消息，至少生产一条消息
 5      */
 6     public static void product(String key, String... message){
 7         try{
 8             DRJedis.getJedisPool().rpush(MQ_PREFIX + key, message);
 9         }catch(Exception e){
10             log.error("jedis product message", e);
11         }
12     }

 1 /**
 2      * 消费消息
 3      * @param key 消息key
 4      * @param customer 消息消费业务，需要实现Customer接口的customer方法进行业务处理
 5      */
 6     public static void customer(String key, ICustomer customer){
 7         try{
 8             Jedis jedis = DRJedis.getJedisPool();
 9             while (true){
10                 //弹出消息
11                 List<String> messages = jedis.blpop(0, MQ_PREFIX + key);
12                 if (messages.size() >= 2){
13                     try {
14                         //消费消息
15                         customer.customer(messages.get(0), messages.get(1));
16                     }catch (RollbackException e){
17                         //消息消费失败则要重新生产消息，并有左侧压入队列，保证下次第一个被消费
18                         jedis.lpush(messages.get(0), messages.get(1));
19                     }
20                 }
21             }
22         }catch(Exception e){
23             log.error("jedis customer message", e);
24         }
25     }

1 /**
2  * Created by xiao on 2016/5/20.
3 */
4 public interface ICustomer {
5 
6     void customer(String key, String message) throws RollbackException;
7 
8 }

 1 /**
 2  * 回滚异常
 3  */
 4 public class RollbackException extends Exception {
 5 
 6     public RollbackException(String message) {
 7         super(message);
 8     }
 9 
10     public RollbackException() {
11         super();
12     }
13 
14     public RollbackException(String message, Throwable cause) {
15         super(message, cause);
16     }
17 
18     public RollbackException(Throwable cause) {
19         super(cause);
20     }
21 
22     protected RollbackException(String message, Throwable cause, boolean enableSuppression, boolean writableStackTrace) {
23         super(message, cause, enableSuppression, writableStackTrace);
24     }
25 }