Fork me on GitHub

java对RedisApi使用

字数统计: 6.3k字   |   阅读时长≈ 23分

击穿

redis作为缓存

key：过期时间

自动清除LRU，LFU

【前提一定是发生了高并发】

击穿的意思就是：redis的一个key正好过期，而请求正好访问该k，

那么就击穿了，去到了数据库

—》因为key的过期造成并发访问数据库

解决—>并发有了，阻止并发到达DB，redis又没有key

redis是单进程单实例的–》无论怎么并发，redis处理数据是一条一条的—–>客户端就实现这么一个逻辑—>第一次请求redis，失败并返回，client收到

—》发送创建命令【setnx()这个就当于是一把琐，为啥呢？因为这个操作是key没有才创建，又因为是单进程，一个去数据库处理完之后，后面肯定就会有该key】

—>set key value nx px 成功的进DB,否则sleep+get

1570705222601

中间少了一层代理—其实没有画，因为代理不会去跟数据库打交到，省略了一下

lua脚本

get key1->发现没有

set key1 random() nx px 3000：设个请求service必须设置网络超时时间，并且要小于琐的有效期–避免重复设琐

————-如果set key1 random() nx，会出现死锁，

———————-可能是一个service挂了，琐没有删除

———————-其他的service的请求就会一直执行不了

所有的请求只有三种情况：

————————–set操作成功

——————————————set 成功的请求进入数据库，取回数据，然后set k1

—————————set操作失败

——————————————失败的请求执行get key1，sleep+重试

—————————set网络超时失败

——————————————–set重试

但是琐的过期时间为多少呢：

—————琐过期了，redis触发删除

—————-这时要判断我的service是否存活

—————-死了可以跑其他service的重新set琐

—————–service活着，为了避免进入数据库，

——————如果数据库发生阻塞：

———————————————-去多少都无济于事，反而增加数据库的压力

———————————————所以可以再跑一个线程，去监控琐，

——————————————–线程内容：

———————————————————get key1!=fasle,结束

——————————————————-否则 TTL获取时间《50毫秒

———————————————————-加2秒

穿透

redis作为缓存， redis缓存里面没有该key，数据库也没有该key

就是业务查询的数据库里面根本没有的，这样redis肯定会把请求扔给数据库

使用bloom过滤器，对数据库的所有元素进程hash函数映射

1570706728484

对数据的所有元素通过hash映射函数，bitmap存储：

———–来达到一个过滤,三种形式：

—————————————–形式一：

————————————————数据库所有元素的算法+bitmap在client端

—————————————-形式二：

————————————————数据库所有元素的

————————————————映射算法发生在客户端

————————————————-信息存储bitmap在缓存redis

—————————————–形式三:

———————————————–函数映射和bitmap都在缓存redis发生

几种过滤器

bloom–>缺点–>只能增加，不能删除

布谷鸟过滤器，是可以增删改的

雪崩

redis还是作为缓存

大量的key同时失效是雪崩，间接造成大量的访问进DB

—–分两种—->

指定时间点过期更新的缓存，时点性的—>【就使用击穿的方案，采用排队的方式，慢慢更新到可用】—–》【这与这种，客户端也可以让服务在该时点延时几分钟】 一般是两种相加的方案

—–与时点无关—-》解决—->随机过期时间，让其均匀，就是是多长就是多长，可以用随机过期

1570707604375

雪崩–>发生在过期时间处理不恰当,某一时间点大量的key被删除,间接造成大量请求进入数据库

分两种业务：

—————业务如果是零点必须请缓存，然后重新更新：

———————————————对请求做延时操作

——————————————–为了保证业务的可用性

——————————————–也就是不能延时太久

——————————————–结合方案一，击穿方案

——————————————–对请求里的不同类型的key做活锁

—————-业务如果不是时点清数据的

——————————————–可以使用随机过期时间来对key的过期时间均匀

分布式锁

1570713178694

get key返回没有

使用多个线程同时向

redis无主集群发送set琐的请求，

————使用多线程对锁的时间进行延长

————得到N/2+1的琐

—————————–去数据库取数据

————得到少于N/2+1琐

——————————立马释放琐

——————————-此时sleep+get key

————–数据取到后释放del 删除无主集群的琐

解决就是防止redis单点宕机，造成重复获得琐

缺点：

如果该service获得琐，由于网络原因，一致阻塞，就造成死琐

那么就需要代理层或者client对service做一层健康检查

redis分布式琐比较复杂

zookeeper 速度慢了一点，使用简单

Redis分布式锁

分布式锁在很多场景中是非常有用的原语， 不同的进程必须以独占资源的方式实现资源共享就是一个典型的例子。

有很多分布式锁的库和描述怎么实现分布式锁管理器（DLM)的博客,但是每个库的实现方式都不太一样，很多库的实现方式为了简单降低了可靠性，而有的使用了稍微复杂的设计。

这个页面试图提供一个使用Redis实现分布式锁的规范算法。我们提出一种算法，叫Redlock,我们认为这种实现比普通的单实例实现更安全,我们希望redis社区能帮助分析一下这种实现方法，并给我们提供反馈。

实现细节

在我们开始描述算法之前，我们已经有一些可供参考的实现库.

安全和活性失效保障

按照我们的思路和设计方案，算法只需具备3个特性就可以实现一个最低保障的分布式锁。

  1. 安全属性（Safety property）: 独享（相互排斥）。在任意一个时刻，只有一个客户端持有锁。
  2. 活性A(Liveness property A): 无死锁。即便持有锁的客户端崩溃（crashed)或者网络被分裂（gets partitioned)，锁仍然可以被获取。
  3. 活性B(Liveness property B): 容错。 只要大部分Redis节点都活着，客户端就可以获取和释放锁.

实现Redis分布式锁的最简单的方法就是在Redis中创建一个key，【这个key有一个失效时间（TTL)，以保证锁最终会被自动释放掉（这个对应特性2）】。当客户端释放资源(解锁）的时候，【会删除掉这个key。】

从表面上看，似乎效果还不错，但是这里有一个【问题】：

这个架构中存在一个严重的单点失败问题。如果Redis挂了怎么办？

如果通过增加一个slave节点解决这个问题。这通常是行不通的。

这样做，我们不能实现资源的独享,因为Redis的主从同步通常是异步的。

在这种场景（主从结构）中存在明显的竞态:

  1. 客户端A从master获取到锁
  2. 在master将锁同步到slave之前，master宕掉了。
  3. slave节点被晋级为master节点
  4. 客户端B就可以取得了琐进入数据库。安全失效！

有时候程序就是这么巧，比如说正好一个节点挂掉的时候，两个客户端同时取到了锁。如果你可以接受【这种小概率错误–>一般是可以接受的—>只要琐的数量比较小—–那么获得琐的service不会太多，数据库可以接受】那用这个基于复制的方案就完全没有问题。

单Redis实例实现分布式锁的正确方法

在尝试克服上述单实例设置的限制之前，让我们先讨论一下在这种简单情况下实现分布式锁的正确做法，实际上这是一种可行的方案，尽管存在竞态，结果仍然是可接受的

获取锁使用命令:

1
SET key value NX PX 30000-->key和value随便给

这个命令仅在不存在key的时候才能被执行成功（NX选项），并且这个key有一个30秒的自动失效时间（PX属性）。这个key的值是“my_random_value”(一个随机值），对于一把锁，客户端设置的value必须相同，【不同客户端如果获得同一把琐，clientA阻塞了，然后过期了，clientB在这种情况下获得同一把琐，为了防止A得到资源，删除琐时的误操作，value值就要不同】

value的值必须是随机数主要是为了更安全的释放锁，释放锁的时候使用脚本告诉Redis:只有key存在并且存储的值和我指定的值一样才能告诉我删除成功。可以通过以下Lua脚本实现：

1
2
3
4
5
if redis.call("get",KEYS[1]) == ARGV[1] then
    return redis.call("del",KEYS[1])
else
    return 0
end

【使用这种方式释放锁可以避免删除别的客户端获取成功的锁】。

举个例子：客户端A取得资源锁，但是紧接着被一个其他操作阻塞了，

当客户端A运行完毕其他操作后要释放锁时，

原来的锁早已超时并且被Redis自动释放，并且在【这期间资源锁又被客户端B再次获取到】。

如果仅使用DEL命令将key删除，那么这种情况就会把客户端B的锁给删除掉。使用Lua脚本就不会存在这种情况，因为脚本仅会删除value等于客户端A的value的key（value相当于客户端的一个签名）。

这个随机字符串应该怎么设置？：

————-我认为它应该是从/dev/urandom产生的一个20字节随机数，但是我想你可以找到比这种方法代价更小的方法，只要这个数在你的任务中是唯一的就行。

————–例如一种安全可行的方法是使用/dev/urandom作为RC4的种子和源产生一个伪随机流;一种更简单的方法是把以毫秒为单位的unix时间和客户端ID拼接起来，理论上不是完全安全，但是在多数情况下可以满足需求.

key的失效时间，被称作“锁定有效期”。它不仅是key自动失效时间，而且还是一个客户端持有锁多长时间后可以被另外一个客户端重新获得的时间

截至到目前，我们已经有较好的方法获取锁和释放锁。基于Redis单实例，假设这个单实例总是可用，这种方法已经足够安全。

分布式琐Redlock算法

———–为了防止redis主从复制，切换过程中琐丢失，client重复设琐的问题

———-一般针对于琐的数量过大

在Redis的分布式环境中，我们假设有N个Redis master。shard分片模式。

如何确保在每个redis获取和释放锁。

在这个样例中，我们假设有5个Redis master节点，这是一个比较合理的设置，这样保证他们不会同时都宕掉。

为了取到锁，客户端应该执行以下操作:

  1. 获取当前Unix时间，以毫秒为单位。

  2. 依次尝试从每个redis实例，使用相同的key和随机值获取锁，当向Redis设置锁时,客户端应该设置一个网络连接和响应超时时间，这个超时时间应该小于锁的失效时间。

    ——->例如你的锁自动失效时间为10秒，则超时时间应该在5-50毫秒之间。

    这样服务器端Redis已经挂掉的情况下，客户端不会死死地等待响应结果。

    如果服务器端没有在规定时间内响应，客户端应该尽快尝试另外一个Redis实例。

  3. 客户端使用当前时间减去开始获取锁时间（步骤1记录的时间）就得到获取锁使用的时间。当且仅当从N/2+1个（这里是3个节点）的Redis节点都取到锁，并且使用的时间小于锁失效时间时，锁才算获取成功。

  4. 如果取到了锁，key的真正有效时间等于有效时间减去获取锁时间（步骤3计算的结果）。

  5. 如果因为某些原因，获取锁失败（没有在至少N/2+1个Redis实例取到锁或者取锁时间已经超过了有效时间），客户端应该在所有的Redis实例上进行解锁（即便某些Redis实例根本就没有加锁成功）。

这个算法是异步的么?

算法基于这样一个假设：虽然多个进程之间没有时钟同步，但每个进程都以相同的时钟频率前进，时间差相对于失效时间来说几乎可以忽略不计。这种假设和我们的真实世界非常接近：每个计算机都有一个本地时钟，我们可以容忍多个计算机之间有较小的时钟漂移。

从这点来说，我们必须再次强调我们的互相排斥规则：只有在锁的有效时间（在步骤3计算的结果）范围内客户端能够做完它的工作，锁的安全性才能得到保证（锁的实际有效时间通常要比设置的短，因为计算机之间有时钟漂移的现象）。.

失败时重试

当客户端无法取到锁时，应该在一个随机延迟后重试,防止多个客户端在同时抢夺同一资源的锁（这样会导致脑裂，没有人会取到锁）。同样，客户端取得大部分Redis实例锁所花费的时间越短，脑裂出现的概率就会越低（必要的重试），所以，理想情况一下，客户端应该同时（并发地）向所有Redis发送SET命令。

需要强调，当客户端从大多数Redis实例获取锁失败时，应该尽快地释放（部分）已经成功取到的锁，

这样其他的客户端就不必非得等到锁过完“有效时间”才能取到（然而，如果已经存在网络分裂，客户端已经无法和【所有的Redis实例通信，和一部分还可以通信】，此时就只能等待key的自动释放了，等于被惩罚了）。

释放锁

释放锁比较简单，向所有的Redis实例发送释放锁命令即可，不用关心之前有没有从Redis实例成功获取到锁.

安全争议

这个算法安全么？我们可以从不同的场景讨论一下。

让我们假设客户端从大多数Redis实例取到了锁。所有的实例都包含同样的key，并且key的有效时间也一样。然而，key肯定是在不同的时间被设置上的，所以key的失效时间也不是精确的相同。我们假设第一个设置的key时间是T1(开始向第一个server发送命令前时间），最后一个设置的key时间是T2(得到最后一台server的答复后的时间），我们可以确认，第一个server的key至少会存活 MIN_VALIDITY=TTL-(T2-T1)-CLOCK_DRIFT。所有其他的key的存活时间，都会比这个key时间晚，所以可以肯定，所有key的失效时间至少是MIN_VALIDITY。

当大部分实例的key被设置后，其他的客户端将不能再取到锁，因为至少N/2+1个实例已经存在key。所以，如果一个锁被（客户端）获取后，客户端自己也不能再次申请到锁(违反互相排斥属性）。

然而我们也想确保，当多个客户端同时抢夺一个锁时不能两个都成功。

如果客户端在获取到大多数redis实例锁，使用的时间接近或者已经大于失效时间，客户端将认为锁是失效的锁，并且将释放掉已经获取到的锁，所以我们只需要在有效时间范围内获取到大部分锁这种情况。在上面已经讨论过有争议的地方，在MIN_VALIDITY时间内，将没有客户端再次取得锁。所以只有一种情况，多个客户端会在相同时间取得N/2+1实例的锁，那就是取得锁的时间大于失效时间（TTL time)，这样取到的锁也是无效的.

活性争议

系统的活性安全基于三个主要特性:

  1. 锁的自动释放（因为key失效了）：最终锁可以再次被使用.
  2. 客户端【分布琐通常会将没有获取到的锁删除，一定要删琐】，或者锁被取到后，使用完后，客户端会主动（提前）释放锁
  3. 当客户端重试获取锁时，需要等待一段时间，这个时间必须大于从大多数Redis实例成功获取锁的时间，以最大限度地避免脑裂。.

然而，当网络出现问题时系统在失效时间(TTL)内就无法服务，这种情况下我们的程序就会为此付出代价。如果网络持续的有问题，可能就会出现死循环了。 这种情况发生在当客户端刚取到一个锁还没有来得及释放锁就被网络隔离.

如果网络一直没有恢复，这个算法会导致系统不可用.

性能，崩溃恢复和Redis同步

很多用户把Redis当做分布式锁服务器，使用获取锁和释放锁的响应时间，每秒钟可用执行多少次 acquire / release 操作作为性能指标。为了达到这一要求，增加Redis实例当然可用降低响应延迟

——-没有钱买硬件的”穷人”,也可以在网络方面做优化，使用非阻塞模型，一次发送所有的命令，然后异步的读取响应结果，假设客户端和redis服务器之间的RTT都差不多。

然而，如果我们想使用可以从备份中恢复的redis模式，有另外一种持久化情况你需要考虑，.

我们考虑这样一种场景，假设我们的redis没用使用备份。一个客户端获取到了3个实例的锁。此时，其中一个已经被客户端取到锁的redis实例被重启，在这个时间点，就可能出现3个节点没有设置锁，此时如果有另外一个客户端来设置锁，锁就可能被再次获取到，这样锁的互相排斥的特性就被破坏掉了。

如果我们启用了AOF持久化，情况会好很多。我们可用使用SHUTDOWN命令关闭然后再次重启。因为Redis到期是语义上实现的，所以当服务器关闭时，实际上还是经过了时间，所有（保持锁）需要的条件都没有受到影响. 没有受到影响的前提是redis优雅的关闭。停电了怎么办？如果redis是每秒执行一次fsync，那么很有可能在redis重启之后，key已经丢弃。理论上，如果我们想在Redis重启地任何情况下都保证锁的安全，我们必须开启fsync=always的配置。这反过来将完全破坏与传统上用于以安全的方式实现分布式锁的同一级别的CP系统的性能.

然而情况总比一开始想象的好一些。当一个redis节点重启后，只要它不参与到任意当前活动的锁，没有被当做“当前存活”节点被客户端重新获取到,算法的安全性仍然是有保障的。

为了达到这种效果，我们只需要将新的redis实例，在一个TTL时间内，对客户端不可用即可，在这个时间内，所有客户端锁将被失效或者自动释放.

使用延迟重启可以在不采用持久化策略的情况下达到同样的安全，然而这样做有时会让系统转化为彻底不可用。比如大部分的redis实例都崩溃了，系统在TTL时间内任何锁都将无法加锁成功。

使算法更加可靠：锁的扩展

如果你的工作可以拆分为许多小步骤，可以将有效时间设置的小一些，使用锁的一些扩展机制。在工作进行的过程中，当发现锁剩下的有效时间很短时，可以再次向redis的所有实例发送一个Lua脚本，【让key的有效时间延长一点（前提还是key存在并且value是之前设置的value)】。

客户端扩展TTL时必须像首次取得锁一样在大多数实例上扩展成功才算再次取到锁，并且是在有效时间内再次取到锁（算法和获取锁是非常相似的）。

这样做从技术上将并不会改变算法的正确性，所以扩展锁的过程中仍然需要达到获取到N/2+1个实例这个要求，否则活性特性之一就会失效。

API（jedis，lettuce，springboot，low/high level）

jedis是线程不安全的

1570755736218

多个线程访问同一个jedis会出现数据错乱，例如线程a刚把一个jedis的属性改掉，还没有设回，其他线程就会读到脏数据

那么可以使用一个享元模式，把jedis池化，实例化多个jedis，这样就允许多个线程使用多个jedis实例

lettuce是线程安全的

low/high level

————–low level 可以更灵活的控制使用

————–high level 对low level的包装，易于使用

————-使用

IDEA

新建项目

选择Spring Initializr项目

前面是maven操作

然后就是要选择nosql—->

连接：

1
2
3
4
5
6
7
8


redis-server --port 6379

redis-cli 

CONFIG GET *---->获取所有配置
CONFIG SET protected-mode no---关闭远程保护模式

低高及API使用

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
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180

-------------------
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.ConfigurableApplicationContext;

@SpringBootApplication
public class RedistestApplication {

    public static void main(String[] args) {

        ConfigurableApplicationContext ctx = SpringApplication.run(RedistestApplication.class, args);
        TestRedis bean = ctx.getBean(TestRedis.class);
        bean.testRedis();

    }

}
-----------------------------------------

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.ValueOperations;
import org.springframework.stereotype.Component;

@Component
public class TestRedis {

    @Autowired
    private RedisTemplate redisTemplate;

    @Autowired
    private StringRedisTemplate stringRedisTemplate;


    public void testRedis(){

        /*ValueOperations valueOperations = redisTemplate.opsForValue();
        valueOperations.set("Hello","china");
        System.out.println(valueOperations.get("Hello"));*/
        stringRedisTemplate.opsForValue().set("hello01","world");
        System.out.println(stringRedisTemplate.opsForValue().get("hello01"));
    }


}





127.0.0.1:6379> KEYS *
1) "\xac\xed\x00\x05t\x00\x05Hello"
127.0.0.1:6379> get Hello
(nil)
127.0.0.1:6379> keys *
1) "hello01"
2) "\xac\xed\x00\x05t\x00\x05Hello"


        
    
-----说明，
---------stringRedisTemplate直接的序列化，
---------redisTemplate的序列化参杂了一些东西



-------------low level



    public void testRedis(){

        /**
         * conn 就相当于是redis的直接交互的客户端，linux上redis自带的客户端
         * 有一些什么操作，这个conn都有
         */
        RedisConnection conn = redisTemplate.getConnectionFactory().getConnection();
        
        RedisConnection conne = stringRedisTemplate.getConnectionFactory().getConnection();
        conn.set("hello03".getBytes(), "03".getBytes());
         }
          conn里面参数是byte[]说明是要自己先序列化好


127.0.0.1:6379> keys *
1) "hello01"
2) "\xac\xed\x00\x05t\x00\x05Hello"
3) "hello03"

    

hash

如果需要序列化一个对象 需要明文的方式就得要
Jackson2HashMapper对象序列化json格式
但是需要引入一个依赖
       <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-json</artifactId>
        </dependency>



flatten=true  平面映射
address.city     Castle Black
address.country  The North
flatten=false 正常的映射
address { "city" : "Castle Black", "country" : "The North" }  



    public void testRedis(){
        Person p = new Person();
        p.setName("zhangsan");
        p.setAge(16);

        ObjectMapper objectMapper=new ObjectMapper();
        Jackson2HashMapper jm = new Jackson2HashMapper(objectMapper, false);
        redisTemplate.opsForHash().putAll("zgw", jm.toHash(p));
       ----这里会对key 进行java的序列化
       -------stringRedisTemplate只对string做序列化，不是string会报错
           -------------------------------
           此时可以设置自定义序列化
           这是用json的序列化器序列化的结果
            redisTemplate.setHashValueSerializer(new Jackson2JsonRedisSerializer<Object>(Object.class));
        redisTemplate.setHashKeySerializer(new Jackson2JsonRedisSerializer<Object>(Object.class));
           
 -----------------------------------------          
        
             Map zgw = redisTemplate.opsForHash().entries("zgw");

        Person person = objectMapper.convertValue(zgw, Person.class);

        System.out.println(person==p);

    }

------------发现序列化器是很多时候都要使用的，那么就自定义个序列化器

@Configuration
public class MyTemplate  {

    @Bean
   public StringRedisTemplate jsonTemplate(RedisConnectionFactory fc){

        StringRedisTemplate stringRedisTemplate = new StringRedisTemplate(fc);
        stringRedisTemplate.setHashValueSerializer(new Jackson2JsonRedisSerializer<Object>(Object.class));
        return  stringRedisTemplate;
    }

}


    @Autowired
    @Qualifier("jsonTemplate")
    private StringRedisTemplate stringRedisTemplate;


   stringRedisTemplate.convertAndSend("ooxx","hello");发布PUB
   
   订阅
        RedisConnection connection = stringRedisTemplate.getConnectionFactory().getConnection();
        connection.subscribe(new MessageListener() {
            @Override
            public void onMessage(Message message, byte[] bytes) {
                byte[] body = message.getBody();
                System.out.println(new String(body));
                System.out.println(bytes);
            }
        },"ooxx".getBytes());
       while (true){监听订阅后，是程序不要停止
           stringRedisTemplate.convertAndSend("ooxx","hello");
       }
       


stringRedisTemplate.executePipelined管道

service以什么格式序列化存储：

1
2


赏

谢谢你请我吃糖果

支付宝
微信

扫一扫，分享到微信

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia-plus根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true

主要涉及技术：
Java后端开发
开源爱好者、Linux

联系QQ:2960571342

很惭愧

只做了一点微小的工作
谢谢大家