Redis 高阶应用

@Componentpublic class IDUtil{	//开始时间戳(单位：秒) 2000-01-01 00:00:00	private static final long START_TIMESTAMP = 946656000L;	//Spring Data Redis 提供的 Redis 操作模板	@Resource	private StringRedisTemplate stringRedisTemplate;	/**	 * 获取 ID   格式：时间戳+序列号	 * @param keyPrefix Redis 序列号前缀	 * @return 生成的 ID	 */	public long getNextId(String keyPrefix){		//获取当前时间戳		LocalDateTime now = LocalDateTime.now();		long nowTimestamp = now.toEpochSecond(ZoneOffset.UTC);		//获取 ID 时间戳		long timestamp = nowSecond - START_TIMESTAMP;		//获取当前日期		String date = now.format(DateTimeFormatter.ofPattern("yyyyMMdd"));		//生成 key		String key = "incr:" + keyPrefix + ":" + date;		//获取序列号		long count = stringRedisTemplate.opsForValue().increment(key);		//生成 ID 并返回		return timestamp << 32 | count;	}}

public class DistributedLockUtil {	//分布式锁前缀	private static final String KEY_PREFIX = "distributed:lock:";	//业务名	private String business;	//分布式锁的值	private String value;	//Spring Data Redis 提供的 Redis 操作模板	private StringRedisTemplate stringRedisTemplate;	//私有化无参构造	private DistributedLockUtil(){}	//有参构造	public DistributedLockUtil(String business,StringRedisTemplate stringRedisTemplate){		this.business = business;		this.stringRedisTemplate = stringRedisTemplate;		this.value = UUID.randomUUID().toString();	}	/**	 * 尝试获取锁	 * @param timeout 超时时间(单位：秒)	 * @return 锁是否获取成功	 */	public boolean tryLock(long timeout){		//生成分布式锁的 key		StringBuffer keyBuffer = new StringBuffer(KEY_PREFIX);		keyBuffer.append(business);		Boolean success = stringRedisTemplate.opsForValue().setIsAbsent(keyBuffer.toString(),value,timeout, TimeUnit.SECONDS);		//返回结果  注意：为了防止自动拆箱时出现空指针，所以这里用了 equals 判断		return Boolean.TRUE.equals(success);	}	/**	 * 释放锁(不安全版)	 */	public void unLock(){		//生成分布式锁的 key		StringBuffer keyBuffer = new StringBuffer(KEY_PREFIX);		keyBuffer.append(business);		//获取分布式锁的值		String redisValue = stringRedisTemplate.opsForValue().get(keyBuffer.toString());		//判断值是否一致，防止误删		if (value.equals(redisValue)) {			//当代码执行到这里时，如果 JVM 恰巧执行了垃圾回收(虽然几率极低)，就会导致所有线程阻塞等待，因此这里仍然会有线程安全的问题			stringRedisTemplate.delete(keyBuffer.toString());		}	}	/**	 * 通过脚本释放锁(彻底解决线程安全问题)	 */	public void unLockWithScript(){		//加载 lua 脚本,实际工作中我们可以将脚本设置为常量，并在静态代码块中初始化(脚本内容在下文)		DefaultRedisScript<Long> script = new DefaultRedisScript<>();		script.setLocation(new ClassPathResource("unlock.lua"));		script.setResultType(Long.class);		//生成分布式锁的 key		StringBuffer keyBuffer = new StringBuffer(KEY_PREFIX);		keyBuffer.append(business);		//调用 lua 脚本释放锁		stringRedisTemplate.execute(script,				Collections.singletonList(keyBuffer.toString()),				value);	}}

-- 判断值是否一致，防止误删if(redis.call('get',KEYS[1]) == VRGV[1]) then	-- 判断通过，释放锁	return redis.call('del',KEYS[1])end-- 判断不通过，返回 0return 0

public class RedisQueueUtil{	//Spring Data Redis 提供的 Redis 操作模板	private StringRedisTemplate stringRedisTemplate;	/**	 * 获取消息队列中的数据，执行该方法前，一定要确保消费者组已经创建	 * @param queueName 队列名	 * @param groupName 消费者组名	 * @param consumerName 消费者名	 * @param type 返回值类型	 * @return 消息队列中的数据	 */	public <T> T getQueueData(String queueName, String groupName, String consumerName, Class<T> type){		while (true){			try {				//获取消息队列中的信息				List<MapRecord<String,Object,Object>> list = stringRedisTemplate.opsForStream().read(						Consumer.from(groupName,consumerName),						StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)),						StreamOffset.create(queueName, ReadOffset.lastConsumed())				);				//判断消息是否获取成功				if (list == null || list.isEmpty()){					//如果获取失败，说明没有消息，继续下一次循环					continue;				}				//如果获取成功，则解析消息中的数据				MapRecord<String,Object,Object> record = list.get(0);				Map<Object,Object> values = record.getValue();				String jsonString = JSON.toJSONString(values);				T result = JSON.parseObject(jsonString, type);				// ACK				stringRedisTemplate.opsForStream().acknowledge(queueName,groupName,record.getId());				//返回结果				return result;			}catch (Exception e){				while (true){					try {						//获取 pending-list 队列中的信息						List<MapRecord<String,Object,Object>> list = stringRedisTemplate.opsForStream().read(								Consumer.from(groupName,consumerName),								StreamReadOptions.empty().count(1)),								StreamOffset.create(queueName,ReadOffset.from("0")						);						//判断消息是否获取成功						if (list == null || list.isEmpty()){							//如果获取失败，说明 pending-list 没有异常消息，结束循环							break;						}						//如果获取成功，则解析消息中的数据						MapRecord<String,Object,Object> record = list.get(0);						Map<Object,Object> values = record.getValue();						String jsonString = JSON.toJSONString(values);						T result = JSON.parseObject(jsonString, type);						// ACK						stringRedisTemplate.opsForStream().acknowledge(queueName,groupName,record.getId());						//返回结果						return result;					}catch (Exception ex){						log.error("处理 pending-list 订单异常",ex);						try {							Thread.sleep(50);						}catch (InterruptedException err){							err.printStackTrace();						}					}				}			}		}	}	/**	 * 向消息队列中发送数据	 * @param queueName 消息队列名	 * @param map 要发送数据的集合	 */	public void sendQueueData(String queueName, Map<String,Object> map){		StringBuilder builder = new StringBuilder("redis.call('xadd','");		builder.append(queueName).append("','*','");		Set<String> keys = map.keySet();		for(String key:keys){			builder.append(key).append("','").append(map.get(key)).append("','");		}		String script = builder.substring(0, builder.length() - 2);		script += ")";		stringRedisTemplate.execute(new DefaultRedisScript<Long>(script,Long.class),Collections.emptyList());	}}