书成

再这样堕落下去就给我去死啊你这混蛋！！！

0%

Redis使用

发表于 更新于 分类于

Redis

 Redis 是广泛应用的一个 NoSQL 数据库，基于 C 开发的键值对存储数据库，Redis一般用作缓存。但实际上 Redis 除了缓存之外，还有许多更加丰富的使用场景。比如 分布式锁，限流等。

 Redis默认不支持远程连接，要开启远程连接，需要修改配置文件的两个地方：注释掉 bind 127.0.0.1 以及 开启密码校验，去掉 requirepass 的注释，修改自己的密码。

Jedis

 Jedis是 Redis 的 Java 客户端开发包之一，它的优点是 API 与 Redis 的原生 API 基本一致，不必花费时间再记忆 Jedis 的 API 使用。

Jedis 连接 Redis

1
2
3
4
Jedis jedis = new Jedis("192.168.43.196"); // redis地址，默认端口6379
jedis.auth("123456"); // 密码认证
System.out.println(jedis.ping()); // 返回PONG连接成功
jedis.close();

 以上代码就是 Jedis 连接 Redis 的代码，连接上 Redis 之后，对 Redis 的各种操作例如 set，get 的 API 与 Redis 原生 API 一致。

Jedis 连接池

 Jedis 提供了连接池。

1
2
3
4
5
6
JedisPool pool = new JedisPool("192.168.43.196"); // 创建一个连接池
// Jedis 实现了 Closable 接口
try(Jedis jedis = pool.getResource()) { // 获得一个连接并操作
	jedis.auth("123456");
	System.out.println(jedis.ping());
}

 还可以使用 GenericObjectPoolConfig 类来配置连接池。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
public static JedisPool getJedisPool() {
	GenericObjectPoolConfig config = new GenericObjectPoolConfig();
	config.setMaxIdle(300); // 最大空闲数
	config.setMaxTotal(100); // 最大连接数
	config.setMaxWaitMillis(20000); // 连接最大等待时间，-1表示无限制
	config.setTestOnBorrow(true); // 空闲时检查有效性
	return new JedisPool(config, "192.168.43.196", 6379, 20, "123456");
}
public static void main(String[] args) {
	JedisPool pool = getJedisPool();
	try(Jedis jedis = pool.getResource()) {
		System.out.println(jedis.ping());
	}
}

做分布式锁

  Redis 除了用作缓存外，还可以用作分布式锁。当有多个线程竞争 key 时，利用分布式锁可以保证对 key 的操作的同步。

  在 Redis 中，setnx 指令可以用来实现分布式锁，如果 key 不存在， 它会将 key 的 value 设置并返回1，如果 key 存在，则设置失败返回0。因此多线程竞争时，setnx 返回1则表示获得了锁，在操作完成后再将设置的 key 删除表示释放了锁。

1
2
3
4
5
6
7
8
9
10
11
JedisPool pool = getJedisPool();
try(Jedis jedis = pool.getResource()){
	Long setnx = jedis.setnx("lock", "v");
	if(setnx == 1) {
		// 获得了锁，执行操作
				
		jedis.del("lock"); // 操作完成后释放资源
	}else {
		// 未获得锁
	}
}

 但是以上的代码存在问题，如果线程在执行 del 指令之前挂掉了，这个锁就无法释放，后面的所有线程都无法再获得锁。要解决这个问题，最简单的想法是给锁添加过期时间。

 Redis 支持 setnx 和 expire 通过一个命令来执行：SET key value [EX seconds][PX milliseconds] [NX|XX]，其中，EX seconds 设置键过期时间为秒，PX milliseconds 设置键过期时间为毫秒，NX 表示只有在键不存在时才操作，XX 表示只有键存在时才操作，这个命令操作成功后返回 OK ，失败返回 NIL。例如 SET lock 1 ex 10 nx，表示只有 lock 键不存在时才操作，设置 lock 这个键的值为1，过期时间十秒。

1
2
3
4
5
6
7
8
9
10
11
12
JedisPool pool = getJedisPool();
try(Jedis jedis = pool.getResource()){
    // SetParams类可以定义set指令参数
	String result = jedis.set("lock","1", new SetParams().ex(10).nx());
	if(result.equals("OK")) {
		// 获得了锁，执行操作
				
		jedis.del("lock"); // 操作完成后释放资源
	}else {
	// 未获得锁
	}
}

 但是以上代码依旧有问题，如果某个线程A的正常运行时间大于超时时间，那么该线程还没有执行完锁就被自动释放了，其它线程B就可以获得这个锁，线程B还未执行完，线程A就释放了线程B的锁，线程C又会获得这个锁。

 解决这个问题就是如果能判断加锁的线程是哪个线程，解锁的时候必须也是加锁线程才行，例如可以将 key 的 value 设置为随机字符串，每次释放锁都比较随机字符串是否是当前线程生成的，如果是，就可以释放，不是就不能释放。但是这样以来每次释放都需要获得 value，然后比较，然后再删除，三个步骤，不具有原子性。为了让其具有原子性，可以使用 Lua 脚本。

 如果不想编写 Lua 脚本，可以使用 Redission 库，这个库提供了额外的分布式锁的功能。

做消息队列

 假如系统比较简单，可以直接使用 Redis 做消息队列，Redis 并不专业做消息队列，仅仅适合用在简单场景中。通过使用 Redis 的 List 数据结构，就可以实现消息队列。可以使用 List 的 blpop 或者 brpop 来实现阻塞式的读取消息。

 Redis中的 zset 数据结构则可以用来做延迟消息队列。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
public class TestMQ {
	
	private static JedisPool jedisPool = MyJedis.getJedisPool();
	
	private static class Producer implements Runnable{
		public void run() {
			try(Jedis jedis = jedisPool.getResource()){
				int i = 0;
				while(!Thread.currentThread().isInterrupted()) {
					// 向 queue 这个zset中加入消息，延迟3秒
					jedis.zadd("queue", System.currentTimeMillis() + 3000, "msg----" + i);
					System.out.println(new Date() + "放入消息-----" + i++);
					try {
						Thread.sleep(500);
					} catch (InterruptedException e) {
      					break;
					}
				}
			}
		}
	}
	
	private static class Consumer implements Runnable{
		public void run() {
			try(Jedis jedis = jedisPool.getResource()){
				while(!Thread.currentThread().isInterrupted()) {
					Set<String> set = jedis.zrangeByScore("queue", 0, System.currentTimeMillis());
					if(set.isEmpty()) {
						try {
							Thread.sleep(500);
						} catch (InterruptedException e) {
							break;
						}
						continue;
					}
                      // 获得集合中第一个元素
					String message = set.iterator().next();
                      // 如果能删掉该元素，说明抢到了该消息
					if(jedis.zrem("queue", message) > 0) {
						System.out.println(new Date() + "收到了消息--------" + message);
					}
				}
			}
		}
	}
	
	public static void main(String[] args) throws InterruptedException {
		Thread p = new Thread(new Producer());
		Thread c = new Thread(new Consumer());
		p.start();
		c.start();
		Thread.sleep(8000);
		p.interrupt();
		c.interrupt();
	}
}

位操作

 Redis 中对字符串的操作可以使用位操作，例如保存一个用户一年的签到记录，如果利用位操作来做，只需要365个比特就能完全记录下来，在用户量较大的情况下节省了大量空间。

1
2
3
4
5
6
// 常用操作
setbit key offset value  # 设置key的值中的某一位
getbit key offset   # 获取key的值中的某一位
bitcount key [start end]   # 统计key的值的二进制表示的1的数量，start与end是字符的位置，不是比特位的位置
bitpos key bit [start] [end] # 统计起始位置的0或者1的数量，起始位置也是字符的位置  
bitfield  # 批量位操作

HyperLogLog

 HyperLogLog 是用来做基数统计的算法，HyperLogLog 的优点是，在输入元素的数量或者体积非常非常大时，计算基数所需的空间总是固定 的、并且是很小的。在 Redis 里面，每个 HyperLogLog 键只需要花费 12 KB 内存，就可以计算接近 2^64 个不同元素的基 数。这和计算基数时，元素越多耗费内存就越多的集合形成鲜明对比。但是，因为 HyperLogLog 只会根据输入元素来计算基数，而不会储存输入元素本身，所以 HyperLogLog 不能像集合那样，返回输入的各个元素 。 

1
2
3
4
// 常用操作
PFADD key element [element ...] # 添加指定元素到 HyperLogLog 中。
PFCOUNT key [key ...] # 返回给定 HyperLogLog 的基数估算值。
PFMERGE destkey sourcekey [sourcekey ...] # 将多个 HyperLogLog 合并为一个 HyperLogLog

 HyperLogLog 提供了一套不怎么精确但是够用的去重方案，会有误差。例如下列代码输出结果5037，并不完全准确。

1
2
3
4
5
6
7
8
public static void main(String[] args) {
	JedisPool pool = MyJedis.getJedisPool();
	try(Jedis jedis = pool.getResource()){
		for(int i = 0; i < 5000; i++)
			jedis.pfadd("count", "test" + i);
		System.out.println(jedis.pfcount("count"));
	}
}

布隆过滤器

 布隆过滤器相当于是一个不太精确的 set 集合，我们可以利用它里边的 contains 方法去判断某一个对象是否存在，但是这个判断不是特别精确。一般来说，通过 contains 判断某个值不存在， 那就一定不存在，但是判断某个值存在的话，则他可能不存在 。

 布隆过滤器需要一个巨大的位数组和几个 hash 函数，它的 add 操作相当于对元素分别求出几个 hash 值，将位数组上 hash 值对应的位置设置为1即可。判断一个元素是否存在时，求出这个元素的几个 hash 值，如果它们对应的位都是1，表示这个元素存在，否则，这个元素不存在。因此布隆过滤器可能存在误判，一般位数组越大，误判率越小，占用空间越大。

Redis限流

 Redis也可以利用 zset 数据结构作限流，key 与请求有关，分数保存为请求的时间戳，这样，利用 zrangeByScore 可以求出某一段时间内的请求数量，当某个时间段内的请求数量达到一定数量，就拒绝其它访问。能够保证每 N 秒至多有 M 个访问能通过。

GeoHash

 GeoHash是一种地址编码方式，将二维空间的经纬度编码为一个字符串。它表示的是一块区域而不是一个特定的点。例如某地经纬度为(40，116)

  1. 纬度的范围在 (-90,90) 之间，中间值为 0，对于 39.9053908600 值落在 (0,90),因此得到的值为 1

  2. (0,90) 的中间值为 45，39.9053908600 落在 (0,45) 之间，因此得到一个 0

  3. (0,45) 的中间值为 22.5，39.9053908600 落在 (22.5,45)之间，因此得到一个 1

  4. ……

 最后计算出的纬度二进制是 101，经度二进制是110，之后按照经度占偶数位，纬度占奇数位合并，结果为111001，然后按照 Base32 （0-9,b-z,去掉 a i l 0）对合并后的二进制数据进行编码，编码的时候，先将二进制转换为十进制，然后进行编码。

 这种编码格式有特定的规律，例如一个地址编码之后的格式是 123，另一个地址编码之后的格式是 123456， 从字符串上就可以看出来，123456 表示的区域处于 123 表示的区域之中。

事务

 Redis 中的事务并不是严格的事务，它只保证了隔离性，即事务不会被其它事务打断，并不保证原子性。Redis 事务对应的指令有 watch，multi，exec，discard。

1
2
3
4
5
6
7
8
9
10
11
12
13
multi
set k1 v1
set k2 v2
get k2
incr k2
get k2
exec
// 以上命令执行后结果
1) OK
2) OK
3) "v2"
4) (error) ERR value is not an integer or out of range
5) "v2"

 可以看到，虽然事务中有一条指令出现了错误，但是并没有像 MySQL 一样整个事务都被回滚了。