package com.test
import redis.clients.jedis.Jedis;

import java.util.Collections;
import java.util.List;

/**
 * @desc redis隊(duì)列實(shí)現(xiàn)方式
 * @anthor 
 * @date 
 **/
public class RedisUcUitl {

  private static final String LOCK_SUCCESS = "OK";
  private static final String SET_IF_NOT_EXIST = "NX";
  private static final String SET_WITH_EXPIRE_TIME = "PX";

  private static final Long RELEASE_SUCCESS = 1L;

  private RedisUcUitl() {

  }
  /**
   * logger
   **/

  /**
   * 存儲(chǔ)redis隊(duì)列順序存儲(chǔ) 在隊(duì)列首部存入
   *
   * @param key  字節(jié)類型
   * @param value 字節(jié)類型
   */
  public static Long lpush(Jedis jedis, final byte[] key, final byte[] value) {

    return jedis.lpush(key, value);
  
  }

  /**
   * 移除列表中最后一個(gè)元素 并將改元素添加入另一個(gè)列表中 ，當(dāng)列表為空時(shí) 將阻塞連接 直到等待超時(shí)
   *
   * @param srckey
   * @param dstkey
   * @param timeout 0 表示永不超時(shí)
   * @return
   */
  public static byte[] brpoplpush(Jedis jedis,final byte[] srckey, final byte[] dstkey, final int timeout) {

    return jedis.brpoplpush(srckey, dstkey, timeout);

  }

  /**
   * 返回制定的key,起始位置的redis數(shù)據(jù)
   * @param redisKey
   * @param start
   * @param end -1 表示到最后
   * @return
   */
  public static List<byte[]> lrange(Jedis jedis,final byte[] redisKey, final long start, final long end) {
    
    return jedis.lrange(redisKey, start, end);
  }

  /**
   * 刪除key
   * @param redisKey
   */
  public static void delete(Jedis jedis, final byte[] redisKey) {
    
     return jedis.del(redisKey);
  }

  /**
   * 嘗試加鎖
   * @param lockKey key名稱
   * @param requestId 身份標(biāo)識(shí)
   * @param expireTime 過(guò)期時(shí)間
   * @return
   */
  public static boolean tryGetDistributedLock(Jedis jedis,final String lockKey, final String requestId, final int expireTime) {
    String result = jedis.set(lockKey, requestId, SET_IF_NOT_EXIST, SET_WITH_EXPIRE_TIME, expireTime);
    return LOCK_SUCCESS.equals(result);

  }

  /**
   * 釋放鎖
   * @param lockKey key名稱
   * @param requestId 身份標(biāo)識(shí)
   * @return
   */
  public static boolean releaseDistributedLock(Jedis jedis,final String lockKey, final String requestId) {
    final String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
    jedis.eval(script, Collections.singletonList(lockKey), Collections.singletonList(requestId));

    return RELEASE_SUCCESS.equals(result);

  }
}

while(true){
  // 消費(fèi)隊(duì)列
  try{
    // 被放入redis隊(duì)列的數(shù)據(jù) 序列化后的
    byte[] bytes = RedisUcUitl.brpoplpush(keyStr.getBytes(UTF_8), dstKeyStr.getBytes(UTF_8), 1);
    if(bytes == null || bytes.isEmpty()){
      // 隊(duì)列中沒數(shù)據(jù)時(shí)退出
      break;
    }
    // 反序列化對(duì)象
    Map<String, Object> singleMap = (Map<String, Object>) ObjectSerialUtil.bytesToObject(bytes);
    // 塞入唯一的值 防止被其他線程誤解鎖
    String requestId = UUID.randomUUID().toString();
    boolean lockGetFlag = RedisUcUitl.tryGetDistributedLock(keyStr,requestId, 100);
    if(lockGetFlag){
      // 成功獲取鎖 進(jìn)行業(yè)務(wù)處理
      //TODO
      // 處理完畢釋放鎖 
      boolean freeLock = RedisUcUitl.releaseDistributedLock(keyStr, requestId);

    }else{
      // 未能獲得鎖放入等待隊(duì)列
     RedisUcUitl.lpush(keyStr.getBytes(UTF_8), ObjectSerialUtil.objectToBytes(param));
  
    }
    
  }catch(Exception e){
    break;
  }
  
}

public class ObjectSerialUtil {

  private ObjectSerialUtil() {
//    工具類
  }

  /**
   * 將Object對(duì)象序列化為byte[]
   *
   * @param obj 對(duì)象
   * @return byte數(shù)組
   * @throws Exception
   */
  public static byte[] objectToBytes(Object obj) throws IOException {
    ByteArrayOutputStream bos = new ByteArrayOutputStream();
    ObjectOutputStream oos = new ObjectOutputStream(bos);
    oos.writeObject(obj);
    byte[] bytes = bos.toByteArray();
    bos.close();
    oos.close();
    return bytes;
  }


  /**
   * 將bytes數(shù)組還原為對(duì)象
   *
   * @param bytes
   * @return
   * @throws Exception
   */
  public static Object bytesToObject(byte[] bytes) {
    try {
      ByteArrayInputStream bin = new ByteArrayInputStream(bytes);
      ObjectInputStream ois = new ObjectInputStream(bin);
      return ois.readObject();
    } catch (Exception e) {
      throw new BaseException("反序列化出錯(cuò)！", e);
    }
  }
}