redis 高阶应用-kb88凯时官网登录

生成全局唯一 id

• 全局唯一 id 需要满足以下要求：

• 唯一性：在分布式环境中，要全局唯一

• 高可用：在高并发情况下保证可用性

• 高性能：在高并发情况下生成 id 的速度必须要快，不能花费太长时间

• 递增性：要确保整体递增的，以便于数据库创建索引

• 安全性：id 的规律性不能太明显，以免信息泄露

从上面的要求可以看出，全局 id 生成器的条件还是比较苛刻的，而 redis 恰巧可以满足以上要求。

redis 本身就是就是以性能著称，因此完全符合高性能的要求，其次使用 redis 的 incr 命令可以保证递增性，配合相应的分布式 id 生成算法便可以实现唯一性和安全性，redis 可以通过哨兵、主从等集群方案来保证可用性。因此 redis 是一个不错的选择。
下面我们就写一个简单的示例，来让大家感受一下，实际工作中大家可以根据需要进行调整：

@component
public 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;
	}
}

分布式锁

在 jvm 内部会有一个锁监视器来控制线程间的互斥，但在分布式的环境下会有多台机器部署同样的服务，也就是说每台机器都会有自己的锁监视器。而 jvm 的锁监视器只能保证自己内部线程的安全执行，并不能保证不同机器间的线程安全执行，因此也很难避免高并发带来的线程安全问题。因此就需要分布式锁来保证整个集群的线程的安全，而分布式锁需要满足 5 点要求：多进程可见、互斥性、高可用、高性能、安全性
其中核心要求就是多进程之间互斥，而满足这一点的方式有很多，最常见的有三种：mysql、redis、zookeeper。

通过对比我们发现，其中 redis 的效果最理想，所以下面就用 redis 来实现一个简单的分布式锁。

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 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);
	}
}

lua 脚本内容如下：

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

虽然通过 lua 脚本解决了线程不安全的问题，但是仍然存在以下问题：

• 不可重入：同一个线程无法多次获取同一把锁
• 不可重试：获取锁只能尝试一次，失败就返回 false，没有重试机制
• 超时释放：锁超时释放虽然可以避免死锁，但如果业务执行耗时较长，也会导致锁释放，存在安全隐患
• 主从一致性：如果 redis 提供了主从集群，主从同步存在延迟，当宕机时，如果从机还没来得及同步主机的锁数据，则会出现锁失效。

要解决以上问题也非常简单，只需要利用 redis 的 hash 结构记录线程标识和重入次数就可以解决不可重入的问题。利用信号量和 pubsub 功能实现等待、唤醒，获取锁失败的重试机制即可解决不可重试的问题。而超时释放的问题则可以通过获取锁时为锁添加一个定时任务(俗称看门狗)，定期刷新锁的超时时间即可。至于主从一致性问题，我们只需要利用多个独立的 redis 节点(非主从)，必须在所有节点都获取重入锁，才算获取锁成功。

有的人可能说了，虽然说起来简单，但真正实现起来也不是很容易呀。对于这种问题，大家不用担心，俗话说得好想要看的更远，需要站在巨人的肩膀上。对于上述的需求，早就有了成熟的开源方案 ，我们直接拿来用就可以了，无需重复造轮子，具体使用方法可以查看。

轻量化消息队列

虽然市面上有很多优秀的消息中间件如 rocketmq、kafka 等，但对于应用场景较为简单，只需要简单的消息传递，比如任务调度、简单的通知系统等，不需要复杂的消息路由、事务支持的业务来说，用那些专门的消息中间件成本就显得过高。因此我们就可以使用 redis 来做消息队列。
redis 提供了三种不同的方式来实现消息队列：

• list 结构：可以使用 list 来模拟消息队列，可以使用 brpop 或 blpop 命令来实现类似 jvm 阻塞队列的消息队列。
• pubsub：基于发布/订阅的消息模型，但不支持数据持久化，且消息堆积有上限，超出时数据丢失。
• stream：redis 5.0 新增的数据类型，可以实现一个功能非常完善的消息队列，也是我们实现消息队列的首选。

下面我就采用 redis 的 stream 实现一个简单的案例来让大家感受一下，实际工作中大家可以根据需要进行调整：

public class redisqueueutil{
	//spring data redis 提供的 redis 操作模板
	private stringredistemplate stringredistemplate;
	/**
	 * 获取消息队列中的数据，执行该方法前，一定要确保消费者组已经创建
	 * @param queuename 队列名
	 * @param groupname 消费者组名
	 * @param consumername 消费者名
	 * @param type 返回值类型
	 * @return 消息队列中的数据
	 */
	public  t getqueuedata(string queuename, string groupname, string consumername, class type){
		while (true){
			try {
				//获取消息队列中的信息
				list> 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 record = list.get(0);
				map 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> 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 record = list.get(0);
						map 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 map){
		stringbuilder builder = new stringbuilder("redis.call('xadd','");
		builder.append(queuename).append("','*','");
		set 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(script,long.class),collections.emptylist());
	}
}