Redis面试总结&史上最全Redis面试题及答案(转)

Posted on 2022-02-12

Redis 在互联网技术存储方面使用如此广泛，几乎所有的后端技术面试官都要在 Redis 的使用和原理方面对小伙伴们进行各种刁难。作为一名在互联网技术行业打击过成百上千名【请允许我夸张一下】的资深技术面试官，看过了无数落寞的身影失望的离开，略感愧疚，故献上此文，希望各位读者以后面试势如破竹，永无失败！

Redis 有哪些数据结构？

字符串 String、字典 Hash、列表 List、集合 Set、有序集合 SortedSet。

如果你是 Redis 中高级用户，还需要加上下面几种数据结构 HyperLogLog、Geo、Pub/Sub。

如果你说还玩过 Redis Module，像 BloomFilter，RedisSearch，Redis-ML，面试官得眼睛就开始发亮了。

使用过 Redis 分布式锁么，它是什么回事？

先拿 setnx 来争抢锁，抢到之后，再用 expire 给锁加一个过期时间防止锁忘记了释放。

这时候对方会告诉你说你回答得不错，然后接着问如果在 setnx 之后执行 expire 之前进程意外 crash 或者要重启维护了，那会怎么样？

这时候你要给予惊讶的反馈：唉，是喔，这个锁就永远得不到释放了。紧接着你需要抓一抓自己得脑袋，故作思考片刻，好像接下来的结果是你主动思考出来的，然后回答：我记得 set 指令有非常复杂的参数，这个应该是可以同时把 setnx 和 expire 合成一条指令来用的！对方这时会显露笑容，心里开始默念：摁，这小子还不错。

假如 Redis 里面有 1 亿个 key，其中有 10w 个 key 是以某个固定的已知的前缀开头的，如果将它们全部找出来？

使用 keys 指令可以扫出指定模式的 key 列表。

对方接着追问：如果这个 redis 正在给线上的业务提供服务，那使用 keys 指令会有什么问题？

这个时候你要回答 redis 关键的一个特性：redis 的单线程的。keys 指令会导致线程阻塞一段时间，线上服务会停顿，直到指令执行完毕，服务才能恢复。这个时候可以使用 scan 指令，scan 指令可以无阻塞的提取出指定模式的 key 列表，但是会有一定的重复概率，在客户端做一次去重就可以了，但是整体所花费的时间会比直接用 keys 指令长。

使用过 Redis 做异步队列么，你是怎么用的？

一般使用 list 结构作为队列，rpush 生产消息，lpop 消费消息。当 lpop 没有消息的时候，要适当 sleep 一会再重试。

如果对方追问可不可以不用 sleep 呢？list 还有个指令叫 blpop，在没有消息的时候，它会阻塞住直到消息到来。

如果对方追问能不能生产一次消费多次呢？使用 pub/sub 主题订阅者模式，可以实现 1:N 的消息队列。

如果对方追问 pub/sub 有什么缺点？在消费者下线的情况下，生产的消息会丢失，得使用专业的消息队列如 rabbitmq 等。

如果对方追问 redis 如何实现延时队列？我估计现在你很想把面试官一棒打死如果你手上有一根棒球棍的话，怎么问的这么详细。但是你很克制，然后神态自若的回答道：使用 sortedset，拿时间戳作为 score，消息内容作为 key 调用 zadd 来生产消息，消费者用 zrangebyscore 指令获取 N 秒之前的数据轮询进行处理。

到这里，面试官暗地里已经对你竖起了大拇指。但是他不知道的是此刻你却竖起了中指，在椅子背后。

如果有大量的 key 需要设置同一时间过期，一般需要注意什么？

如果大量的 key 过期时间设置的过于集中，到过期的那个时间点，redis 可能会出现短暂的卡顿现象。一般需要在时间上加一个随机值，使得过期时间分散一些。

Redis 如何做持久化的？

bgsave 做镜像全量持久化，aof 做增量持久化。因为 bgsave 会耗费较长时间，不够实时，在停机的时候会导致大量丢失数据，所以需要 aof 来配合使用。在 redis 实例重启时，会使用 bgsave 持久化文件重新构建内存，再使用 aof 重放近期的操作指令来实现完整恢复重启之前的状态。

对方追问那如果突然机器掉电会怎样？取决于 aof 日志 sync 属性的配置，如果不要求性能，在每条写指令时都 sync 一下磁盘，就不会丢失数据。但是在高性能的要求下每次都 sync 是不现实的，一般都使用定时 sync，比如 1s1 次，这个时候最多就会丢失 1s 的数据。

对方追问 bgsave 的原理是什么？你给出两个词汇就可以了，fork 和 cow。fork 是指 redis 通过创建子进程来进行 bgsave 操作，cow 指的是 copy on write，子进程创建后，父子进程共享数据段，父进程继续提供读写服务，写脏的页面数据会逐渐和子进程分离开来。

Pipeline 有什么好处，为什么要用 pipeline？

可以将多次 IO 往返的时间缩减为一次，前提是 pipeline 执行的指令之间没有因果相关性。使用 redis-benchmark 进行压测的时候可以发现影响 redis 的 QPS 峰值的一个重要因素是 pipeline 批次指令的数目。

Redis 的同步机制了解么？

Redis 可以使用主从同步，从从同步。第一次同步时，主节点做一次 bgsave，并同时将后续修改操作记录到内存 buffer，待完成后将 rdb 文件全量同步到复制节点，复制节点接受完成后将 rdb 镜像加载到内存。加载完成后，再通知主节点将期间修改的操作记录同步到复制节点进行重放就完成了同步过程。

是否使用过 Redis 集群，集群的原理是什么？

Redis Sentinal 着眼于高可用，在 master 宕机时会自动将 slave 提升为 master，继续提供服务。

Redis Cluster 着眼于扩展性，在单个 redis 内存不足时，使用 Cluster 进行分片存储。

史上最全Redis面试题及答案

1、什么是Redis？

Redis本质上是一个Key-Value类型的内存数据库，很像memcached，整个数据库统统加载在内存当中进行操作，定期通过异步操作把数据库数据flush到硬盘上进行保存。因为是纯内存操作，Redis的性能非常出色，每秒可以处理超过 10万次读写操作，是已知性能最快的Key-Value DB。 Redis的出色之处不仅仅是性能，Redis最大的魅力是支持保存多种数据结构，此外单个value的最大限制是1GB，不像 memcached只能保存1MB的数据，因此Redis可以用来实现很多有用的功能，比方说用他的List来做FIFO双向链表，实现一个轻量级的高性能消息队列服务，用他的Set可以做高性能的tag系统等等。另外Redis也可以对存入的Key-Value设置expire时间，因此也可以被当作一个功能加强版的memcached来用。 Redis的主要缺点是数据库容量受到物理内存的限制，不能用作海量数据的高性能读写，因此Redis适合的场景主要局限在较小数据量的高性能操作和运算上。

2、Redis相比memcached有哪些优势？

(1) memcached所有的值均是简单的字符串，redis作为其替代者，支持更为丰富的数据类型

(2) redis的速度比memcached快很多

(3) redis可以持久化其数据

3、Redis支持哪几种数据类型？

String、List、Set、Sorted Set、hashes

4、Redis主要消耗什么物理资源？

内存。

5、Redis的全称是什么？

Remote Dictionary Server。

6、Redis有哪几种数据淘汰策略？

noeviction:返回错误当内存限制达到并且客户端尝试执行会让更多内存被使用的命令（大部分的写入指令，但DEL和几个例外）

allkeys-lru: 尝试回收最少使用的键（LRU），使得新添加的数据有空间存放。

volatile-lru: 尝试回收最少使用的键（LRU），但仅限于在过期集合的键,使得新添加的数据有空间存放。

allkeys-random: 回收随机的键使得新添加的数据有空间存放。

volatile-random: 回收随机的键使得新添加的数据有空间存放，但仅限于在过期集合的键。

volatile-ttl: 回收在过期集合的键，并且优先回收存活时间（TTL）较短的键,使得新添加的数据有空间存放。

7、Redis官方为什么不提供Windows版本？

因为目前Linux版本已经相当稳定，而且用户量很大，无需开发windows版本，反而会带来兼容性等问题。

8、一个字符串类型的值能存储最大容量是多少？

512M

9、为什么Redis需要把所有数据放到内存中？

Redis为了达到最快的读写速度将数据都读到内存中，并通过异步的方式将数据写入磁盘。所以redis具有快速和数据持久化的特征。如果不将数据放在内存中，磁盘I/O速度为严重影响redis的性能。在内存越来越便宜的今天，redis将会越来越受欢迎。如果设置了最大使用的内存，则数据已有记录数达到内存限值后不能继续插入新值。

10、Redis集群方案应该怎么做？都有哪些方案？

1.twemproxy，大概概念是，它类似于一个代理方式，使用方法和普通redis无任何区别，设置好它下属的多个redis实例后，使用时在本需要连接redis的地方改为连接twemproxy，它会以一个代理的身份接收请求并使用一致性hash算法，将请求转接到具体redis，将结果再返回twemproxy。使用方式简便(相对redis只需修改连接端口)，对旧项目扩展的首选。问题：twemproxy自身单端口实例的压力，使用一致性hash后，对redis节点数量改变时候的计算值的改变，数据无法自动移动到新的节点。

2.codis，目前用的最多的集群方案，基本和twemproxy一致的效果，但它支持在节点数量改变情况下，旧节点数据可恢复到新hash节点。

3.redis cluster3.0自带的集群，特点在于他的分布式算法不是一致性hash，而是hash槽的概念，以及自身支持节点设置从节点。具体看官方文档介绍。

4.在业务代码层实现，起几个毫无关联的redis实例，在代码层，对key 进行hash计算，然后去对应的redis实例操作数据。这种方式对hash层代码要求比较高，考虑部分包括，节点失效后的替代算法方案，数据震荡后的自动脚本恢复，实例的监控，等等。

11、Redis集群方案什么情况下会导致整个集群不可用？

有A，B，C三个节点的集群,在没有复制模型的情况下,如果节点B失败了，那么整个集群就会以为缺少5501-11000这个范围的槽而不可用。

12、MySQL里有2000w数据，redis中只存20w的数据，如何保证redis中的数据都是热点数据？

redis内存数据集大小上升到一定大小的时候，就会施行数据淘汰策略。

13、Redis有哪些适合的场景？

（1）、会话缓存（Session Cache）

最常用的一种使用Redis的情景是会话缓存（session cache）。用Redis缓存会话比其他存储（如Memcached）的优势在于：Redis提供持久化。当维护一个不是严格要求一致性的缓存时，如果用户的购物车信息全部丢失，大部分人都会不高兴的，现在，他们还会这样吗？

幸运的是，随着 Redis 这些年的改进，很容易找到怎么恰当的使用Redis来缓存会话的文档。甚至广为人知的商业平台Magento也提供Redis的插件。

（2）、全页缓存（FPC）

除基本的会话token之外，Redis还提供很简便的FPC平台。回到一致性问题，即使重启了Redis实例，因为有磁盘的持久化，用户也不会看到页面加载速度的下降，这是一个极大改进，类似PHP本地FPC。

再次以Magento为例，Magento提供一个插件来使用Redis作为全页缓存后端。

此外，对WordPress的用户来说，Pantheon有一个非常好的插件 wp-redis，这个插件能帮助你以最快速度加载你曾浏览过的页面。

（3）、队列

Reids在内存存储引擎领域的一大优点是提供 list 和 set 操作，这使得Redis能作为一个很好的消息队列平台来使用。Redis作为队列使用的操作，就类似于本地程序语言（如Python）对 list 的 push/pop 操作。

如果你快速的在Google中搜索“Redis queues”，你马上就能找到大量的开源项目，这些项目的目的就是利用Redis创建非常好的后端工具，以满足各种队列需求。例如，Celery有一个后台就是使用Redis作为broker，你可以从这里去查看。

（4），排行榜/计数器

Redis在内存中对数字进行递增或递减的操作实现的非常好。集合（Set）和有序集合（Sorted Set）也使得我们在执行这些操作的时候变的非常简单，Redis只是正好提供了这两种数据结构。所以，我们要从排序集合中获取到排名最靠前的10个用户–我们称之为“user_scores”，我们只需要像下面一样执行即可：

当然，这是假定你是根据你用户的分数做递增的排序。如果你想返回用户及用户的分数，你需要这样执行：

ZRANGE user_scores 0 10 WITHSCORES

Agora Games就是一个很好的例子，用Ruby实现的，它的排行榜就是使用Redis来存储数据的，你可以在这里看到。

（5）、发布/订阅

最后（但肯定不是最不重要的）是Redis的发布/订阅功能。发布/订阅的使用场景确实非常多。我已看见人们在社交网络连接中使用，还可作为基于发布/订阅的脚本触发器，甚至用Redis的发布/订阅功能来建立聊天系统！（不，这是真的，你可以去核实）。

14、Redis支持的Java客户端都有哪些？官方推荐用哪个？

Redisson、Jedis、lettuce等等，官方推荐使用Redisson。

15、Redis和Redisson有什么关系？

Redisson是一个高级的分布式协调Redis客服端，能帮助用户在分布式环境中轻松实现一些Java的对象 (Bloom filter, BitSet, Set, SetMultimap, ScoredSortedSet, SortedSet, Map, ConcurrentMap, List, ListMultimap, Queue, BlockingQueue, Deque, BlockingDeque, Semaphore, Lock, ReadWriteLock, AtomicLong, CountDownLatch, Publish / Subscribe, HyperLogLog)。

16、Jedis与Redisson对比有什么优缺点？

Jedis是Redis的Java实现的客户端，其API提供了比较全面的Redis命令的支持；Redisson实现了分布式和可扩展的Java数据结构，和Jedis相比，功能较为简单，不支持字符串操作，不支持排序、事务、管道、分区等Redis特性。Redisson的宗旨是促进使用者对Redis的关注分离，从而让使用者能够将精力更集中地放在处理业务逻辑上。

17、Redis如何设置密码及验证密码？

设置密码：config set requirepass 123456

授权密码：auth 123456

18、说说Redis哈希槽的概念？

Redis集群没有使用一致性hash,而是引入了哈希槽的概念，Redis集群有16384个哈希槽，每个key通过CRC16校验后对16384取模来决定放置哪个槽，集群的每个节点负责一部分hash槽。

19、Redis集群的主从复制模型是怎样的？

为了使在部分节点失败或者大部分节点无法通信的情况下集群仍然可用，所以集群使用了主从复制模型,每个节点都会有N-1个复制品.

20、Redis集群会有写操作丢失吗？为什么？

Redis并不能保证数据的强一致性，这意味这在实际中集群在特定的条件下可能会丢失写操作。

21、Redis集群之间是如何复制的？

异步复制

22、Redis集群最大节点个数是多少？

16384个。

23、Redis集群如何选择数据库？

Redis集群目前无法做数据库选择，默认在0数据库。

24、怎么测试Redis的连通性？

ping

25、Redis中的管道有什么用？

一次请求/响应服务器能实现处理新的请求即使旧的请求还未被响应。这样就可以将多个命令发送到服务器，而不用等待回复，最后在一个步骤中读取该答复。

这就是管道（pipelining），是一种几十年来广泛使用的技术。例如许多POP3协议已经实现支持这个功能，大大加快了从服务器下载新邮件的过程。

26、怎么理解Redis事务？

事务是一个单独的隔离操作：事务中的所有命令都会序列化、按顺序地执行。事务在执行的过程中，不会被其他客户端发送来的命令请求所打断。

事务是一个原子操作：事务中的命令要么全部被执行，要么全部都不执行。

27、Redis事务相关的命令有哪几个？

MULTI、EXEC、DISCARD、WATCH

28、Redis key的过期时间和永久有效分别怎么设置？

EXPIRE和PERSIST命令。

29、Redis如何做内存优化？

尽可能使用散列表（hashes），散列表（是说散列表里面存储的数少）使用的内存非常小，所以你应该尽可能的将你的数据模型抽象到一个散列表里面。比如你的web系统中有一个用户对象，不要为这个用户的名称，姓氏，邮箱，密码设置单独的key,而是应该把这个用户的所有信息存储到一张散列表里面.

30、Redis回收进程如何工作的？

一个客户端运行了新的命令，添加了新的数据。
Redi检查内存使用情况，如果大于maxmemory的限制, 则根据设定好的策略进行回收。
一个新的命令被执行，等等。
所以我们不断地穿越内存限制的边界，通过不断达到边界然后不断地回收回到边界以下。
如果一个命令的结果导致大量内存被使用（例如很大的集合的交集保存到一个新的键），不用多久内存限制就会被这个内存使用量超越。

31、Redis回收使用的是什么算法？

LRU算法

32、Redis如何做大量数据插入？

Redis2.6开始redis-cli支持一种新的被称之为pipe mode的新模式用于执行大量数据插入工作。

33、为什么要做Redis分区？

分区可以让Redis管理更大的内存，Redis将可以使用所有机器的内存。如果没有分区，你最多只能使用一台机器的内存。分区使Redis的计算能力通过简单地增加计算机得到成倍提升,Redis的网络带宽也会随着计算机和网卡的增加而成倍增长。

34、你知道有哪些Redis分区实现方案？

客户端分区就是在客户端就已经决定数据会被存储到哪个redis节点或者从哪个redis节点读取。大多数客户端已经实现了客户端分区。

代理分区意味着客户端将请求发送给代理，然后代理决定去哪个节点写数据或者读数据。代理根据分区规则决定请求哪些Redis实例，然后根据Redis的响应结果返回给客户端。redis和memcached的一种代理实现就是Twemproxy

查询路由(Query routing) 的意思是客户端随机地请求任意一个redis实例，然后由Redis将请求转发给正确的Redis节点。Redis Cluster实现了一种混合形式的查询路由，但并不是直接将请求从一个redis节点转发到另一个redis节点，而是在客户端的帮助下直接redirected到正确的redis节点。

35、Redis分区有什么缺点？

涉及多个key的操作通常不会被支持。例如你不能对两个集合求交集，因为他们可能被存储到不同的Redis实例（实际上这种情况也有办法，但是不能直接使用交集指令）。
同时操作多个key,则不能使用Redis事务.
分区使用的粒度是key，不能使用一个非常长的排序key存储一个数据集（The partitioning granularity is the key, so it is not possible to shard a dataset with a single huge key like a very big sorted set）.
当使用分区的时候，数据处理会非常复杂，例如为了备份你必须从不同的Redis实例和主机同时收集RDB / AOF文件。
分区时动态扩容或缩容可能非常复杂。Redis集群在运行时增加或者删除Redis节点，能做到最大程度对用户透明地数据再平衡，但其他一些客户端分区或者代理分区方法则不支持这种特性。然而，有一种预分片的技术也可以较好的解决这个问题。
36、Redis持久化数据和缓存怎么做扩容？

如果Redis被当做缓存使用，使用一致性哈希实现动态扩容缩容。
如果Redis被当做一个持久化存储使用，必须使用固定的keys-to-nodes映射关系，节点的数量一旦确定不能变化。否则的话(即Redis节点需要动态变化的情况），必须使用可以在运行时进行数据再平衡的一套系统，而当前只有Redis集群可以做到这样。
37、分布式Redis是前期做还是后期规模上来了再做好？为什么？

既然Redis是如此的轻量（单实例只使用1M内存）,为防止以后的扩容，最好的办法就是一开始就启动较多实例。即便你只有一台服务器，你也可以一开始就让Redis以分布式的方式运行，使用分区，在同一台服务器上启动多个实例。

一开始就多设置几个Redis实例，例如32或者64个实例，对大多数用户来说这操作起来可能比较麻烦，但是从长久来看做这点牺牲是值得的。

这样的话，当你的数据不断增长，需要更多的Redis服务器时，你需要做的就是仅仅将Redis实例从一台服务迁移到另外一台服务器而已（而不用考虑重新分区的问题）。一旦你添加了另一台服务器，你需要将你一半的Redis实例从第一台机器迁移到第二台机器。

38、Twemproxy是什么？

Twemproxy是Twitter维护的（缓存）代理系统，代理Memcached的ASCII协议和Redis协议。它是单线程程序，使用c语言编写，运行起来非常快。它是采用Apache 2.0 license的开源软件。 Twemproxy支持自动分区，如果其代理的其中一个Redis节点不可用时，会自动将该节点排除（这将改变原来的keys-instances的映射关系，所以你应该仅在把Redis当缓存时使用Twemproxy)。 Twemproxy本身不存在单点问题，因为你可以启动多个Twemproxy实例，然后让你的客户端去连接任意一个Twemproxy实例。 Twemproxy是Redis客户端和服务器端的一个中间层，由它来处理分区功能应该不算复杂，并且应该算比较可靠的。

39、支持一致性哈希的客户端有哪些？

Redis-rb、Predis等。

40、Redis与其他key-value存储有什么不同？

Redis有着更为复杂的数据结构并且提供对他们的原子性操作，这是一个不同于其他数据库的进化路径。Redis的数据类型都是基于基本数据结构的同时对程序员透明，无需进行额外的抽象。
Redis运行在内存中但是可以持久化到磁盘，所以在对不同数据集进行高速读写时需要权衡内存，应为数据量不能大于硬件内存。在内存数据库方面的另一个优点是，相比在磁盘上相同的复杂的数据结构，在内存中操作起来非常简单，这样Redis可以做很多内部复杂性很强的事情。同时，在磁盘格式方面他们是紧凑的以追加的方式产生的，因为他们并不需要进行随机访问。
41、Redis的内存占用情况怎么样？

给你举个例子： 100万个键值对（键是0到999999值是字符串“hello world”）在我的32位的Mac笔记本上用了100MB。同样的数据放到一个key里只需要16MB，这是因为键值有一个很大的开销。在Memcached上执行也是类似的结果，但是相对Redis的开销要小一点点，因为Redis会记录类型信息引用计数等等。

当然，大键值对时两者的比例要好很多。

64位的系统比32位的需要更多的内存开销，尤其是键值对都较小时，这是因为64位的系统里指针占用了8个字节。但是，当然，64位系统支持更大的内存，所以为了运行大型的Redis服务器或多或少的需要使用64位的系统。

42、都有哪些办法可以降低Redis的内存使用情况呢？

如果你使用的是32位的Redis实例，可以好好利用Hash,list,sorted set,set等集合类型数据，因为通常情况下很多小的Key-Value可以用更紧凑的方式存放到一起。

43、查看Redis使用情况及状态信息用什么命令？

info

44、Redis的内存用完了会发生什么？

如果达到设置的上限，Redis的写命令会返回错误信息（但是读命令还可以正常返回。）或者你可以将Redis当缓存来使用配置淘汰机制，当Redis达到内存上限时会冲刷掉旧的内容。

45、Redis是单线程的，如何提高多核CPU的利用率？

可以在同一个服务器部署多个Redis的实例，并把他们当作不同的服务器来使用，在某些时候，无论如何一个服务器是不够的，所以，如果你想使用多个CPU，你可以考虑一下分片（shard）。

46、一个Redis实例最多能存放多少的keys？List、Set、Sorted Set他们最多能存放多少元素？

理论上Redis可以处理多达232的keys，并且在实际中进行了测试，每个实例至少存放了2亿5千万的keys。我们正在测试一些较大的值。

任何list、set、和sorted set都可以放232个元素。

换句话说，Redis的存储极限是系统中的可用内存值。

47、Redis常见性能问题和解决方案？

(1) Master最好不要做任何持久化工作，如RDB内存快照和AOF日志文件

(2) 如果数据比较重要，某个Slave开启AOF备份数据，策略设置为每秒同步一次

(3) 为了主从复制的速度和连接的稳定性，Master和Slave最好在同一个局域网内

(4) 尽量避免在压力很大的主库上增加从库

(5) 主从复制不要用图状结构，用单向链表结构更为稳定，即：Master <- Slave1 <- Slave2 <- Slave3…

这样的结构方便解决单点故障问题，实现Slave对Master的替换。如果Master挂了，可以立刻启用Slave1做Master，其他不变。

48、Redis提供了哪几种持久化方式？

RDB持久化方式能够在指定的时间间隔能对你的数据进行快照存储.
AOF持久化方式记录每次对服务器写的操作,当服务器重启的时候会重新执行这些命令来恢复原始的数据,AOF命令以redis协议追加保存每次写的操作到文件末尾.Redis还能对AOF文件进行后台重写,使得AOF文件的体积不至于过大.
如果你只希望你的数据在服务器运行的时候存在,你也可以不使用任何持久化方式.
你也可以同时开启两种持久化方式, 在这种情况下, 当redis重启的时候会优先载入AOF文件来恢复原始的数据,因为在通常情况下AOF文件保存的数据集要比RDB文件保存的数据集要完整.
最重要的事情是了解RDB和AOF持久化方式的不同,让我们以RDB持久化方式开始。
49、如何选择合适的持久化方式？

一般来说，如果想达到足以媲美PostgreSQL的数据安全性，你应该同时使用两种持久化功能。如果你非常关心你的数据，但仍然可以承受数分钟以内的数据丢失，那么你可以只使用RDB持久化。

有很多用户都只使用AOF持久化，但并不推荐这种方式：因为定时生成RDB快照（snapshot）非常便于进行数据库备份，并且 RDB 恢复数据集的速度也要比AOF恢复的速度要快，除此之外，使用RDB还可以避免之前提到的AOF程序的bug。

50、修改配置不重启Redis会实时生效吗？

针对运行实例，有许多配置选项可以通过 CONFIG SET 命令进行修改，而无需执行任何形式的重启。从 Redis 2.2 开始，可以从 AOF 切换到 RDB 的快照持久性或其他方式而不需要重启 Redis。检索 ‘CONFIG GET *’ 命令获取更多信息。

但偶尔重新启动是必须的，如为升级 Redis 程序到新的版本，或者当你需要修改某些目前 CONFIG 命令还不支持的配置参数的时候。

转载于：https://cloudpai.gitee.io/2018/04/18/2018-04-18-3/

SpringBoot实现图片上传demo以及Nginx进行代理显示

Posted on 2022-02-12

公司项目需要一个图片上传的功能，就图片能上传到服务器（公司用的windows服务器），然后nginx能进行代理访问到就行了，先简单介绍一下nginx，然后再来实现功能。

一、nginx简介

Nginx (engine x) 是一个高性能的HTTP和反向代理web服务器，特点是占有内存少，并发能力强，事实上nginx的并发能力在同类型的网页服务器中表现较好。
Nginx专门为性能优化而开发，性能是其最重要的考量，实际上非常注重效率，能经起高负载的考验，有报告表明能支持高达50000个并发连接数。

二、反向代理

1.正向代理
在客户端（浏览器）配置代理服务器，通过代理服务器进行互联网访问。
2.反向代理
我们只需要将请求发送到反向代理服务器，由反向代理服务器去选择目标服务器获取数据后，再返回给客户端，此时反向代理服务器和目标服务器对外就是一个服务器，暴露的是代理服务器地址，隐藏了真实服务器IP地址。

三、负载均衡

增加服务器的数量，然后将请求分发到各个服务器上，将原先请求集中到单个服务器上的情况改为分发到多个服务器上，将负载分发到不同的服务器，也就是我们所说的负载均衡。

四、动静分离

为了加快网站的解析速度，可以把动态页面和静态页面由不同的服务器来解析，加快解析速度。减低原来单个服务器的压力。

五、nginx常用命令

1.使用nginx操作命令前提条件：必须进行nginx的目录

2.查看nginx的版本号
nginx -v

3.启动nginx
nginx

4.关闭nginx
nginx -s stop

5.重新加载nginx
nginx -s reload

六、nginx的配置文件（nginx.conf）

nginx配置文件有三部分组成
1.全局块
从配置文件开始到events块之间的内容，主要会设置一些影响nginx服务器整体运行的配置指令。
比如：worker_processes 1; worker_processes值越大，可以支持的并发处理量也越多。

2.events块

events块涉及的指令主要影响nginx服务器与用户的网络连接。
比如：worker_connection 1024; 支持的最大连接数。

3.http块
nginx服务器配置中最频繁的部分，http块也可以包括http全局块、server块。

七、nginx配置图片的访问路径

图片文件上传至服务器D:/images中，然后通过IP地址/upload/加图片名称进行访问。

#user  nobody;
worker_processes  1;

#error_log  logs/error.log;
#error_log  logs/error.log  notice;
#error_log  logs/error.log  info;

#pid        logs/nginx.pid;


events {
    worker_connections  1024;
}


http {
    include       mime.types;
    default_type  application/octet-stream;

    #log_format  main  '$remote_addr - $remote_user [$time_local] "$request" '
    #                  '$status $body_bytes_sent "$http_referer" '
    #                  '"$http_user_agent" "$http_x_forwarded_for"';

    #access_log  logs/access.log  main;

    sendfile        on;
    #tcp_nopush     on;

    #keepalive_timeout  0;
    keepalive_timeout  65;

    #gzip  on;

    server {
        listen       80;
        server_name  localhost;

        #charset koi8-r;

        #access_log  logs/host.access.log  main;

        #访问路径拼接 upload 访问本地绝对路径下的某图片 
        location /upload/ {
        alias D:/images/;
        autoindex on;
        }

        location / {
            root   html;
            index  index.html index.htm;
        }

        #error_page  404              /404.html;

        # redirect server error pages to the static page /50x.html
        #
        error_page   500 502 503 504  /50x.html;
        location = /50x.html {
            root   html;
        }

        # proxy the PHP scripts to Apache listening on 127.0.0.1:80
        #
        #location ~ \.php$ {
        #    proxy_pass   http://127.0.0.1;
        #}

        # pass the PHP scripts to FastCGI server listening on 127.0.0.1:9000
        #
        #location ~ \.php$ {
        #    root           html;
        #    fastcgi_pass   127.0.0.1:9000;
        #    fastcgi_index  index.php;
        #    fastcgi_param  SCRIPT_FILENAME  /scripts$fastcgi_script_name;
        #    include        fastcgi_params;
        #}

        # deny access to .htaccess files, if Apache's document root
        # concurs with nginx's one
        #
        #location ~ /\.ht {
        #    deny  all;
        #}
    }


    # another virtual host using mix of IP-, name-, and port-based configuration
    #
    #server {
    #    listen       8000;
    #    listen       somename:8080;
    #    server_name  somename  alias  another.alias;

    #    location / {
    #        root   html;
    #        index  index.html index.htm;
    #    }
    #}


    # HTTPS server
    #
    #server {
    #    listen       443 ssl;
    #    server_name  localhost;

    #    ssl_certificate      cert.pem;
    #    ssl_certificate_key  cert.key;

    #    ssl_session_cache    shared:SSL:1m;
    #    ssl_session_timeout  5m;

    #    ssl_ciphers  HIGH:!aNULL:!MD5;
    #    ssl_prefer_server_ciphers  on;

    #    location / {
    #        root   html;
    #        index  index.html index.htm;
    #    }
    #}

}

配置好nginx.conf记得重启一下服务器。效果如图：

八、java后台代码

/**
 * 文件上传
 */
@RestController
public class FileController {

    @PostMapping(value = "/fileUpload")
    public String fileUpload(@RequestParam(value = "file") MultipartFile file) {
        if (file.isEmpty()) {
            System.out.println("请选择图片");
        }
        String fileName = file.getOriginalFilename();  // 文件名
        String suffixName = fileName.substring(fileName.lastIndexOf("."));  // 后缀名
        String filePath = "D:/images/"; // 上传后的路径
        fileName = UUID.randomUUID() + suffixName; // 新文件名
        File dest = new File(filePath + fileName);
        if (!dest.getParentFile().exists()) {
            dest.getParentFile().mkdirs();
        }
        try {
            file.transferTo(dest);
        } catch (IOException e) {
            e.printStackTrace();
        }
        //返回图片名称
        return fileName;
    }
}

这里将图片上传至D:/images文件夹中，因为前面配置了nginx的缘故，我这里是在本地测试的，直接使用localhost/upload/+返回的图片名称就可以访问到了。

如果想要通过项目的地址外加端口号进行访问的话，可以配置一个资源映射路径。

/**
 * 资源映射路径
 */
@Configuration
public class MyWebMvcConfig implements WebMvcConfigurer {
    @Override
    public void addResourceHandlers(ResourceHandlerRegistry registry) {
        registry.addResourceHandler("/upload/**").addResourceLocations("file:D:/images/");
    }
}

like this!

项目代码可在github进行查看

SpringBoot整合Shiro实现登录拦截以及认证授权功能

Posted on 2022-02-11

一、Shiro是什么？

Apache Shiro是一个Java安全权限框框架。

Shiro可以非常容易的开发出足够好的应用，其不仅可以在javaEE环境。

Shiro可以完成，认证，授权，加密，会话管理，Web集成，缓存等。

二、Shiro工作原理

应用代码的交互对象是 “Subject”，该对象代表了当前 “用户”，而所有用户的安全操作都会交给 SecurityManager 来管理，而管理过程中会从 Realm 中获取用户对应的角色和权限，可以把 Realm 堪称是安全数据源。

也就是说，我们要使用最简单的 Shiro 应用：

通过 Subject 来进行认证和授权，而 Subject 又委托给了 SecurityManager 进行管理
我们需要给 SecurityManager 注入 Realm 以便其获取用户和权限进行判断
也即，Shiro 不提供用户和权限的维护，需要由开发者自行通过 Realm 注入

三、SpringBoot整合Shiro

1.xml导入jar包

<dependency>
    <groupId>org.apache.shiro</groupId>
    <artifactId>shiro-spring</artifactId>
    <version>1.4.1</version>
</dependency>

2.config目录下创建ShiroConfig.java，拦截以及授权等功能都在这里配置

@Configuration
public class ShiroConfig {
    //1.创建realm对象 需要自定义
    @Bean
    public UserRealm userRealm(){
        return new UserRealm();
    }

    //2.DefaultWebSecurityManager
    @Bean(name = "securityManager")
    public DefaultWebSecurityManager getDefaultWebSecurityManager(@Qualifier("userRealm") UserRealm userRealm){
        DefaultWebSecurityManager securityManager = new DefaultWebSecurityManager();
        //关联UserRealm
        securityManager.setRealm(userRealm);
        return securityManager;
    }

    //3.ShiroFilterFactoryBean
    @Bean
    public ShiroFilterFactoryBean getShiroFilterFactoryBean(@Qualifier("securityManager")DefaultWebSecurityManager defaultWebSecurityManager){
        ShiroFilterFactoryBean bean = new ShiroFilterFactoryBean();
        //设置安全管理器
        bean.setSecurityManager(defaultWebSecurityManager);

        //添加shiro的内置过滤器
        /**
         * anon：无需认证就可以访问
         * authc：必须认证了才能访问
         * user：必须拥有 记住我 功能才能用
         * perms：拥有对某个资源的权限才能访问
         * role：拥有某个角色权限才能访问
         */
        Map<String, String> filterMap = new LinkedHashMap<>();
        //授权
        filterMap.put("/user/add","perms[user:add]"); //进入需要授权（授权规则用户后面接：add）才可以进入add
        filterMap.put("/user/update","perms[user:update]");

        //拦截功能
        filterMap.put("/user/*","authc"); //表示访问user接口的资源都要认证
        bean.setFilterChainDefinitionMap(filterMap);

        //******处理权限不够或者需要授权的业务********
        //如果没有权限authc 设置登录的请求
        bean.setLoginUrl("/toLogin");
        //未授权页面
        bean.setUnauthorizedUrl("/noauth");
        return bean;
    }
}

3.自定义UserRealm类，继承AuthorizingRealm重写用户授权和用户认证的方法

public class UserRealm extends AuthorizingRealm {//用户授权
    @Override
    protected AuthorizationInfo doGetAuthorizationInfo(PrincipalCollection principalCollection) {

        SimpleAuthorizationInfo info = new SimpleAuthorizationInfo();
//        info.addStringPermission("user:add"); 手动添加了权限

        //拿到当前登录的这个对象
        Subject subject= SecurityUtils.getSubject();
        User currentUser = (User) subject.getPrincipal(); //拿到user对象 可以设置用户权限
        //设置当前用户的权限 从数据库上面拿
        info.addStringPermission(currentUser.getPerms());
        System.out.println(currentUser.getPerms());
        return info;
    }

    //用户认证
    @Override
    protected AuthenticationInfo doGetAuthenticationInfo(AuthenticationToken token) throws AuthenticationException {

        UsernamePasswordToken userToken = (UsernamePasswordToken) token;
        //认证用户（连接数据库）
        User user = userService.queryUserByName(userToken.getUsername());
        if(user==null){
            return null; //UnknownAccountException
        }

        //判断session是否有值，显示登录按钮
        Subject currentSubject = SecurityUtils.getSubject();
        Session session = currentSubject.getSession();
        session.setAttribute("loginUser",user);

        //可以加密：MD5 MD5盐值加密（更高级）
        //密码认证（shiro完成）
        return new SimpleAuthenticationInfo(user,user.getPassword(),"");
    }
}

Shiro大致的配置就在这里了，具体功能如拦截功能、授权认证功能，可以在我的
github https://github.com/ruiyeclub/SpringBoot-Hello，进行查看。

Shiro核心概述可参考文章：https://www.cnblogs.com/deng-cc/p/9401900.html

Compose搭建Redis一主二从三哨兵高可用集群

Posted on 2022-02-10

一、Docker Compose介绍

https://docs.docker.com/compose/

Docker官方的网站是这样介绍Docker Compose的：

Compose是用于定义和运行多容器Docker应用程序的工具。通过Compose，您可以使用YAML文件来配置应用程序的服务。然后，使用一个命令，就可以从配置中创建并启动所有服务。

这里Docker Compose给我的感受就是便捷、快速。只需编写一个docker-compose.yml文件，然后通过命令docker-compose up -d，这里就可以搭建多个服务起来，非常适合搭建集群环境。

二、安装Docker Compose工具

通过命令安装Compose工具，安装Compose的前提是安装了Docker环境，如何安装和快速使用Docker，可以翻看我之前的那篇博客–，这里不做赘述了。

1	sudo curl -L https://github.com/docker/compose/releases/download/1.21.2/docker-compose-$(uname -s)-$(uname -m) -o /usr/local/bin/docker-compose

给安装脚本添加执行权限

1	sudo chmod +x /usr/local/bin/docker-compose

这里可以使用命令查看是否安装成功

1	docker-compose -v

到这里Compose工具就安装好了

三、使用Compose搭建Redis主从服务器

主从复制：主节点负责写数据，从节点负责读数据，主节点定期把数据同步到从节点保证数据的一致性。

从性能方面，redis复制功能增加了读的性能，理论上说，每增加一倍的服务器，整个系统的读能力就增加一倍。

选择好路径，通过命令创建文件夹redis，然后进入文件夹创建docker-compose.yml，或者在本地创建，然后通过Xftp工具上传至服务器。docker-compose.yml写入如下内容。

version: '2'
services:
  master:
    image: redis　　　　## 镜像
    container_name: redis-master　　　　##容器别名
    command: redis-server --requirepass 123456　　　　##redis密码
    ports:
    - "6379:6379"　　　　##暴露端口
    networks:
    - sentinel-master
  slave1:
    image: redis
    container_name: redis-slave-1
    ports:
    - "6380:6379"
    command: redis-server --slaveof redis-master 6379 --requirepass 123456 --masterauth 123456 
    depends_on:
    - master
    networks:
    - sentinel-master
  slave2:
    image: redis
    container_name: redis-slave-2
    ports:
    - "6381:6379"
    command: redis-server --slaveof redis-master 6379 --requirepass 123456 --masterauth 123456
    depends_on:
    - master
    networks:
    - sentinel-master

启动redis集群

1	docker-compose up -d

docker ps查看运行的实例，这里我们需要用到主节点的ip地址，注意不是电脑的ip哦，是节点的ip地址。

1	docker inspect 主节点容器id

这里先记着这个ip地址，后面使用到哨兵需要绑定这个主节点的ip。

四、哨兵sentinel模式搭建

上面我们已经搭建好了主从模式，当主服务器宕机后，需要手动把一台服务器切换成主服务器，这里需要人工干预，费事费力，还会造成一段时间内服务不可用。这不是一种推荐的方式，更多时候，我们优先考虑哨兵模式。
哨兵是redis高可用的解决方案，由一个或多个Sentinel实例组成的Sentinel系统可以监视任意多个主服务器以及这些主服务器属下的所有从服务器，它能够在被监视的主服务器下线时，自动将该主服务器属下的某个优先的从服务器升级为新的主服务器，由这个主服务器代替已下线的主服务器继续处理命令请求。

哨兵的功能：
1.监控：哨兵会不断地监控检测主节点和从节点是否运作正常。
2.自动故障转移：当主节点不能正常工作时，哨兵会开始自动故障转移操作，它会将失效主节点的其中一个从节点升级为新的主节点，并让其他从节点改为复制新的主节点。
3.通知：哨兵可以将故障转移的结果发送给客户端。
4.配置提供者：客户端在初始化时，通过连接哨兵来获得当前Redis服务的主节点地址。

这里创建sentinel文件夹，然后同样编写docker-compose.yml文件

version: '2'
services:
  sentinel1:
    image: redis       ## 镜像
    container_name: redis-sentinel-1
    ports:
    - "26379:26379"
    command: redis-sentinel /usr/local/etc/redis/sentinel.conf
    volumes:
    - "./sentinel.conf:/usr/local/etc/redis/sentinel.conf"
  sentinel2:
    image: redis                ## 镜像
    container_name: redis-sentinel-2
    ports:
    - "26380:26379"           
    command: redis-sentinel /usr/local/etc/redis/sentinel.conf
    volumes:
    - "./sentinel2.conf:/usr/local/etc/redis/sentinel.conf"
  sentinel3:
    image: redis                ## 镜像
    container_name: redis-sentinel-3
    ports:
    - "26381:26379"           
    command: redis-sentinel /usr/local/etc/redis/sentinel.conf
    volumes:
    - ./sentinel3.conf:/usr/local/etc/redis/sentinel.conf
networks:
  default:
    external:
      name: redis_sentinel-master

继续在此目录编写文件，编写sentinel.conf文件

port 26379
dir /tmp
#172.18.0.3填写自己的主节点ip
sentinel monitor mymaster 172.18.0.3 6379 2
sentinel auth-pass mymaster 123456 
sentinel down-after-milliseconds mymaster 30000
sentinel parallel-syncs mymaster 1
sentinel failover-timeout mymaster 10000  
sentinel deny-scripts-reconfig yes

对命令的解释

第三行表示Redis监控一个叫做mymaster的运行在172.18.0.3:6379的master，投票达到2则表示master以及挂掉了。
第四行设置主节点的密码。
第五行表示在一段时间范围内sentinel向master发送的心跳PING没有回复则认为master不可用了。
第六行的parallel-syncs表示设置在故障转移之后，同时可以重新配置使用新master的slave的数量。数字越低，更多的时间将会用故障转移完成，但是如果slaves配置为服务旧数据，你可能不希望所有的slave同时重新同步master。因为主从复制对于slave是非阻塞的，当停止从master加载批量数据时有一个片刻延迟。通过设置选项为1，确信每次只有一个slave是不可到达的。
第七行表示10秒内mymaster还没活过来，则认为master宕机了。

创建好文件后，复制好三份

1
2
3

cp sentinel.conf sentinel1.conf
cp sentinel.conf sentinel2.conf
cp sentinel.conf sentinel3.conf

启动redis哨兵模式

1	docker-compose up -d

五、故障转移测试

上面既然说到了哨兵可以自动转移故障，也就是当主节点宕机的时候，它们能通过选举制度，在从节点中选出一个节点来代替主节点。

我们先来看看主节点此时的信息，这里a288c55db497是我主节点容器的ip，这里先是进入容器，再进入redis客户端，再输入密码，然后使用info查看节点信息。

这里可以看到这个节点的身份是master也就是主节点，然后有两个连接着的从节点

现在我们停掉这个节点，看看哨兵是否可以实现故障转移

1	docker stop a288c55db497

然后选择其中一个从节点进入客户端，输入info查看信息

这里发现其中一台从节点已经变成主节点了，而且连接的从节点，也变成了一个。

了解Docker Compose可参考：https://www.jianshu.com/p/658911a8cff3

了解Redis主从哨兵可参考：https://www.cnblogs.com/leeSmall/p/8398401.html

Idea使用插件实现逆向工程搭建SpringBoot项目

Posted on 2022-02-07

之前写SpringBoot项目，每次都要手动去写实体类、dao层啥的，尤其是数据库表字段特别多的时候，特别麻烦。然后很多小伙伴都会用逆向工程来自动生成这些类，省去许多没必要的代码量，但是Mybatis的逆向工程依然需要配置，导逆向工程的jar啊，还有编写generatorConfig.xml文件啊（有兴趣的朋友可以看看这篇）。今天逛gitee的时候，看到了一款可以免去许多配置的idea逆向工程插件，几个步骤简单使用一下

这个插件。

一、下载并安装EasyCode插件

Setting->Plugins。搜索EasyCode插件，并重启Idea编辑器

二、配置MYSQL数据源

点击Idea右侧的Database，然后再点一下加号，选择你的数据库类型，我这里选择的是MySQL

三、配置连接参数

配置好自己的数据库连接参数后，点击Apply，然后再点击Schemas选择使用哪个数据库

四、生成代码

如图，点击Generate Code

这里选择项目模块，还有需要生成的代码

生成后的目录就这个样子，里面还写好了单表的CRUD操作

Java并发编程(四)：并发容器(转)

Posted on 2022-02-06

解决并发情况下的容器线程安全问题的。给多线程环境准备一个线程安全的容器对象。
线程安全的容器对象: Vector, Hashtable。线程安全容器对象，都是使用synchronized 方法实现的。concurrent包中的同步容器，大多数是使用系统底层技术实现的线程安全。类似native。Java8 中使用 CAS。

1、Map/Set

1.1 ConcurrentHashMap/ConcurrentHashSet

底层哈希实现的同步 Map(Set)。效率高，线程安全。使用系统底层技术实现线程安全。量级较 synchronized 低。key 和 value 不能为 null。

1.2 ConcurrentSkipListMap/ConcurrentSkipListSet

底层跳表(SkipList)实现的同步 Map(Set)。有序，效率比 ConcurrentHashMap 稍低。

import java.util.Map;
import java.util.Random;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;

public class Test_01_ConcurrentMap {

    public static void main(String[] args) {
//        final Map<String, String> map = new Hashtable<>();
         final Map<String, String> map = new ConcurrentHashMap<>();
         //ConcurrentSkipListMap跳表实现的，是排序的，最慢
//         final Map<String, String> map = new ConcurrentSkipListMap<>();
        final Random r = new Random();
        Thread[] array = new Thread[100];
        final CountDownLatch latch = new CountDownLatch(array.length);

        long begin = System.currentTimeMillis();
        for (int i = 0; i < array.length; i++) {
            array[i] = new Thread(new Runnable() {
                @Override
                public void run() {
                    for (int j = 0; j < 100000; j++) {
                        map.put("key" + r.nextInt(100000), "value" + r.nextInt(100000));
                    }
                    latch.countDown();
                }
            });
        }
        for (Thread t : array) {
            t.start();
        }
        try {
            latch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        long end = System.currentTimeMillis();
        System.out.println("执行时间为 ： " + (end - begin) + "毫秒！");
    }

}

2、List

2.1 CopyOnWriteArrayList

写时复制集合。写入效率低，读取效率高。每次写入数据，都会创建一个新的底层数组。

import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.CountDownLatch;

public class Test_02_CopyOnWriteList {

    public static void main(String[] args) {
         final List<String> list = new ArrayList<>();
//         final List<String> list = new Vector<>();
//        final List<String> list = new CopyOnWriteArrayList<>();
        final Random r = new Random();
        Thread[] array = new Thread[100];
        final CountDownLatch latch = new CountDownLatch(array.length);

        long begin = System.currentTimeMillis();
        for (int i = 0; i < array.length; i++) {
            array[i] = new Thread(new Runnable() {
                @Override
                public void run() {
                    for (int j = 0; j < 1000; j++) {
                        list.add("value" + r.nextInt(100000));
                    }
                    latch.countDown();
                }
            });
        }
        for (Thread t : array) {
            t.start();
        }
        try {
            latch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        long end = System.currentTimeMillis();
        System.out.println("执行时间为 ： " + (end - begin) + "毫秒！");
        System.out.println("List.size() : " + list.size());
    }

}

3、Queue

3.1 ConcurrentLinkedQueue

基础链表同步队列。

/**
 * 并发容器 - ConcurrentLinkedQueue
 * 队列 - 链表实现的。
 */
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;

public class Test_03_ConcurrentLinkedQueue {

    public static void main(String[] args) {
        Queue<String> queue = new ConcurrentLinkedQueue<>();
        for (int i = 0; i < 10; i++) {
            queue.offer("value" + i);
        }

        System.out.println(queue);
        System.out.println(queue.size());

        // peek() -> 查看queue中的首数据
        System.out.println(queue.peek());
        System.out.println(queue.size());

        // poll() -> 获取queue中的首数据
        System.out.println(queue.poll());
        System.out.println(queue.size());
    }

}

3.2 LinkedBlockingQueue

阻塞队列，队列容量不足自动阻塞，队列容量为 0 自动阻塞。

/**
 * 并发容器 - LinkedBlockingQueue
 * 阻塞容器。
 * put & take - 自动阻塞。
 * put自动阻塞， 队列容量满后，自动阻塞
 * take自动阻塞方法， 队列容量为0后，自动阻塞。
 */

import java.util.Random;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

public class Test_04_LinkedBlockingQueue {

    final BlockingQueue<String> queue = new LinkedBlockingQueue<>();
    final Random r = new Random();

    public static void main(String[] args) {
        final Test_04_LinkedBlockingQueue t = new Test_04_LinkedBlockingQueue();

        new Thread(new Runnable() {
            @Override
            public void run() {
                while (true) {
                    try {
                        t.queue.put("value" + t.r.nextInt(1000));
                        TimeUnit.SECONDS.sleep(1);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        }, "producer").start();

        for (int i = 0; i < 3; i++) {
            new Thread(new Runnable() {
                @Override
                public void run() {
                    while (true) {
                        try {
                            System.out.println(Thread.currentThread().getName() +
                                    " - " + t.queue.take());
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }
            }, "consumer" + i).start();
        }
    }

}

3.3 ArrayBlockingQueue

底层数组实现的有界队列。自动阻塞。根据调用 API(add/put/offer)不同，有不同特性。
当容量不足的时候，有阻塞能力。
add 方法：在容量不足的时候，抛出异常。put 方法在容量不足的时候，阻塞等待。
offer 方法：单参数 offer 方法，不阻塞。容量不足的时候，返回 false。当前新增数据操作放弃。三参数 offer 方法(offer(value,times,timeunit))，容量不足的时候，阻塞 times 时长(单位为 timeunit)，如果在阻塞时长内，有容量空闲，新增数据返回 true。如果阻塞时长范围内，无容量空闲，放弃新增数据，返回 false。

/**
 * 并发容器 - ArrayBlockingQueue
 * 有界容器。
 */

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;

public class Test_05_ArrayBlockingQueue {

    final BlockingQueue<String> queue = new ArrayBlockingQueue<>(3);

    public static void main(String[] args) {
        final Test_05_ArrayBlockingQueue t = new Test_05_ArrayBlockingQueue();

        for (int i = 0; i < 5; i++) {
            // System.out.println("add method : " + t.queue.add("value"+i));
            /*try {
                t.queue.put("put"+i);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("put method : " + i);*/
            // System.out.println("offer method : " + t.queue.offer("value"+i));
            try {
                System.out.println("offer method : " +
                        t.queue.offer("value" + i, 1, TimeUnit.SECONDS));
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

        System.out.println(t.queue);
    }

}

3.4 DelayQueue

延时队列。根据比较机制，实现自定义处理顺序的队列。常用于定时任务。
如:定时关机。

/**
 * 并发容器 - DelayQueue
 * 无界容器。
 */

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;

public class Test_06_DelayQueue {

    static BlockingQueue<MyTask_06> queue = new DelayQueue<>();

    public static void main(String[] args) throws InterruptedException {
        long value = System.currentTimeMillis();
        MyTask_06 task1 = new MyTask_06(value + 2000);
        MyTask_06 task2 = new MyTask_06(value + 1000);
        MyTask_06 task3 = new MyTask_06(value + 3000);
        MyTask_06 task4 = new MyTask_06(value + 2500);
        MyTask_06 task5 = new MyTask_06(value + 1500);

        queue.put(task1);
        queue.put(task2);
        queue.put(task3);
        queue.put(task4);
        queue.put(task5);

        System.out.println(queue);
        System.out.println(value);
        for (int i = 0; i < 5; i++) {
            System.out.println(queue.take());
        }
    }

}

class MyTask_06 implements Delayed {

    private long compareValue;

    public MyTask_06(long compareValue) {
        this.compareValue = compareValue;
    }

    /**
     * 比较大小。自动实现升序
     * 建议和getDelay方法配合完成。
     * 如果在DelayQueue是需要按时间完成的计划任务，必须配合getDelay方法完成。
     */
    @Override
    public int compareTo(Delayed o) {
        return (int) (this.getDelay(TimeUnit.MILLISECONDS) - o.getDelay(TimeUnit.MILLISECONDS));
    }

    /**
     * 获取计划时长的方法。
     * 根据参数TimeUnit来决定，如何返回结果值。
     */
    @Override
    public long getDelay(TimeUnit unit) {
        return unit.convert(compareValue - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
    }

    @Override
    public String toString() {
        return "Task compare value is : " + this.compareValue;
    }

}

3.5 LinkedTransferQueue

转移队列，是一个容量为 0 的队列。使用 transfer 方法，实现数据的即时处理。没有消费者，就阻塞。

/**
 * 并发容器 - LinkedTransferQueue
 * 转移队列
 * add - 队列会保存数据，不做阻塞等待。
 * transfer - 是TransferQueue的特有方法。必须有消费者（take()方法的调用者）。
 * 如果没有任意线程消费数据，transfer方法阻塞。一般用于处理即时消息。
 */

import java.util.concurrent.LinkedTransferQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TransferQueue;

public class Test_07_TransferQueue {

    TransferQueue<String> queue = new LinkedTransferQueue<>();

    public static void main(String[] args) {
        final Test_07_TransferQueue t = new Test_07_TransferQueue();
        
        /*new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    System.out.println(Thread.currentThread().getName() + " thread begin " );
                    System.out.println(Thread.currentThread().getName() + " - " + t.queue.take());
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }, "output thread").start();
        
        try {
            TimeUnit.SECONDS.sleep(2);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        
        try {
            t.queue.transfer("test string");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }*/

        new Thread(new Runnable() {

            @Override
            public void run() {
                try {
                    t.queue.transfer("test string");
                    // t.queue.add("test string");
                    System.out.println("add ok");
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }).start();

        try {
            TimeUnit.SECONDS.sleep(2);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    System.out.println(Thread.currentThread().getName() + " thread begin ");
                    System.out.println(Thread.currentThread().getName() + " - " + t.queue.take());
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }, "output thread").start();

    }

}

3.6 SynchronusQueue

同步队列，是一个容量为0的队列。是一个特殊的TransferQueue。必须现有消费线程等待，才能使用的队列。
add 方法，无阻塞。若没有消费线程阻塞等待数据，则抛出异常。
put 方法，有阻塞。若没有消费线程阻塞等待数据，则阻塞。

/**
 * 并发容器 - SynchronousQueue
 */
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.TimeUnit;

public class Test_08_SynchronusQueue {

    BlockingQueue<String> queue = new SynchronousQueue<>();

    public static void main(String[] args) {
        final Test_08_SynchronusQueue t = new Test_08_SynchronusQueue();

        new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    System.out.println(Thread.currentThread().getName() + " thread begin ");
                    try {
                        TimeUnit.SECONDS.sleep(2);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    System.out.println(Thread.currentThread().getName() + " - " + t.queue.take());
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }, "output thread").start();
        
        /*try {
            TimeUnit.SECONDS.sleep(3);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }*/
        // t.queue.add("test add");
        try {
            t.queue.put("test put");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        System.out.println(Thread.currentThread().getName() + " queue size : " + t.queue.size());
    }

}

Java并发编程(三)：ReentrantLock

Posted on 2022-02-05

ReentrantLock是可以用来代替synchronized的。ReentrantLock比synchronized更加灵活，功能上面更加丰富，性能方面自synchronized优化后两者性能没有什么太大差别。

说一下两者的区别首先ReetrantLock是基于JDK实现层面的，而synchronized是基于JVM层面实现的。ReentrantLock可以进行tryLock尝试锁定，支持公平锁的实现。

Lock lock = new ReentrantLock(); 
lock.lock();
try {
    //业务逻辑  
} finally {
  lock.unlock();
}

一个ReentrantLock的简单实现，要注意的是，必须要手动释放锁，不然很容易产生死锁。使用sync锁定的话遇到异常，jvm会自动释放锁，但是reentrantLock必须手动释放锁，因此经常在finally中进行锁的释放。

Lock lock=new ReentrantLock();
    boolean locked=false;
    try{
        try {
            locked=lock.tryLock(5,TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }finally {
        if(locked){
            lock.unlock();
        }
    }

ReentrantLock可以进行tryLock尝试锁定，tryLock的方法就是进行尝试锁定，如果能在指定时间内得到锁，就返回true，反之返回false。指定时间内无法锁定时，线程可以决定是否继续等待。

1	private static ReentrantLock lock=new ReentrantLock(true); //参数为true表示公平锁

ReentrantLock可以指定为公平锁，ReentrantLock和sync默认是非公平锁，非公平锁：线程加锁时直接尝试获取锁，获取不到就自动到队尾等待。而公平锁：加锁前先查看是否有排队等待的线程，有的话优先处理排在前面的线程，先来先得。

Java并发编程(二)：volatile关键字

Posted on 2022-02-04

volatile是Java虚拟机提供的轻量级的同步机制。volatile关键字有如下两个作用，一句话概括就是内存可见性和禁止重排序。
1）保证被volatile修饰的共享变量对所有线程总是可见的，也就是当一个线程修改了一个被volatile修饰共享变量的值，新值总是可以被其他线程立即得知。
2）禁止指令重排序优化。在执行程序时为了提高性能，编译器和处理器通常会重新安排指令的执行顺序。

public class T {

    /*volatile*/ boolean running=true;

    void m(){
        System.out.println("m start");
        while (running){}
        System.out.println("m end");
    }

    public static void main(String[] args) {
        T t=new T();

        new Thread(()->t.m(),"t1").start();

        try {
            TimeUnit.SECONDS.sleep(1);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        t.running=false;
    }
}

这段代码中没加volatile的话，循环的后面的输出语句”m end”不会输出，但是main方法里面确实是已经将running的值改成了false。这里有关java的内存模型，简单来说就是程序运行时，running的值已经被加载到了缓存里面，而改的是t对象的running值在堆内存中。加了volatile就不一样了，它改完数值之后，会通知其他线程，让其他线程重新读一下堆内存中的值，也就是内存可见性。

public class T1 {

    volatile int count=0;

    /*synchronized*/ void m(){
        for (int i=0;i<10000;i++){
            count++;
        }
    }

    public static void main(String[] args) {

        T1 t=new T1();
        List<Thread> threads=new ArrayList<Thread>();

        for(int i=0;i<10;i++){
            threads.add(new Thread(()->t.m(),"thread-"+i));
        }
        threads.forEach((o)->o.start());
        threads.forEach((o)->{
            try {
                o.join();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
        System.out.println(t.count);
    }
}

这段代码中，十个线程执行count++加到1w循环，理想状态当然是加到10w，但是volatile只能保持线程的可见性，不能保持原子性。给count变量加上了volatile，最后输出的count值也达不到10w。这就是volatile与synchronized的区别，volatile只能保持可见性，而synchronized可以保持可见性和原子性。但是volatile更加轻量级，能用volatile的情况下，尽量别用synchronized。

Java并发编程(一)：线程基础知识以及synchronized关键字

Posted on 2022-02-03

1.线程与多线程的概念：在一个程序中，能够独立运行的程序片段叫作“线程”（Thread）。多线程（multithreading）是指从软件或者硬件上实现多个线程并发执行的技术。

2.多线程的意义：多线程可以在时间片里被cpu快速切换，资源能更好被调用、程序设计在某些情况下更简单、程序响应更快、运行更加流畅。

2.如何启动一个线程：继承Thread类、实现Runnable接口、实现Callable接口

3.为什么要保证线程的同步？：java允许多线程并发控制，当多个线程同时操作一个可共享的资源变量时（如数据的增删改查），将会导致数据不准确，相互之间产生冲突，因此加入同步锁以避免在该线程没有完成操作之前，被其他线程的调用，从而保证了该变量的唯一性和准确性。

4.基本的线程同步：使用synchronized关键字、特殊域变量volatile、wait和notify方法等。

public class T {
    private int count=10;
    private Object o=new Object();

    public void m(){
        synchronized (o){
            //任何线程要执行下面的代码，必须先拿到o的锁
            count--;
            System.out.println(Thread.currentThread().getName()+"count="+count);
        }
    }
}

假设这段代码中有多个线程，当第一个线程执行到m方法来的时候，sync锁住了堆内存中的o对象，这个时候第二个线程是进不来的，它必须等第一个线程执行完，锁释放掉才可以接着执行。这里有个锁的概念叫做互斥锁，sync是一种互斥锁。

public class T1 {
    private static int count =10;

    public synchronized static void m(){
        //这里等同于synchronized(T3.class)
        count--;
        System.out.println(Thread.currentThread().getName()+"count="+count);
    }

    public static void mm(){
        synchronized (T1.class){
            //思考：这里写成synchronized(this)是否可以？
            count--;
        }
    }
}

思考这里，注意sync修饰的是一个静态方法和静态的属性，静态修饰的方法和属性是不需要new出对象来就可以访问的，所以这里没有new出对象，sync锁定的是T3.class对象。

public class T2 implements Runnable{
    private int count =10;

    @Override
    public /*synchronized*/ void run(){
        count--;
        System.out.println(Thread.currentThread().getName()+"count="+count);
    }

    public static void main(String[] args) {
        T2 t=new T2();
        for (int i=0;i<5;i++){
            new Thread(t,"THREAD"+i).start();
        }
    }
}

上面代码中开启了五个线程，执行run方法对count进行减一的操作，这里会出现线程抢占资源的问题。当第一个线程在执行run方法时，减减的过程中，第二个线程也进入了方法，同样也在执行减减，可能会出现第二个线程减完的时候，第一个线程才输出count，这时候就出现了线程重入。处理方法可以加上synchronized关键字，只有等第一个线程执行完毕，第二个线程才可以进入，即同一时刻只能有一个线程来对它进行操作，也就是原子性。

public class Account {

    /**
     * 账号持有人和账号余额
     */
    String name;
    double balance;

    /**
     * 设置账号余额，线程睡眠目的是先让它读数据
     * @param name
     * @param balance
     */
    public synchronized void set(String name,double balance){
        this.name=name;

        try {
            Thread.sleep(2000);
        }catch (InterruptedException e){
            e.printStackTrace();
        }
        this.balance=balance;
    }

    public double getBalance(String name){
        return this.balance;
    }

    public static void main(String[] args) {
        Account a=new Account();
        new Thread(()->a.set("zhangsan",100.0)).start();

        try {
            TimeUnit.SECONDS.sleep(1);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println(a.getBalance("zhangsan"));

        try {
            TimeUnit.SECONDS.sleep(2);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println(a.getBalance("zhangsan"));
    }
}

这是一个小demo，代码中只对set方法加了锁，没有对get方法加锁，这个时候会出现脏读现象。解决方法是读和写的方法都加锁。

public class T3 {

    synchronized void m1(){
        System.out.println("m1 start");
        try {
            TimeUnit.SECONDS.sleep(1);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        m2();
    }

    synchronized void m2() {
        try {
            TimeUnit.SECONDS.sleep(2);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("m2");
    }
}

这个案例中，m1和m2方法都已经加上了锁，当执行m1的时候再去执行m2，这样是可以的。一个同步方法可以调用另外一个同步方法，也就是说sync获得的锁是可重入锁。还有个概念是死锁，死锁是指多个线程抢占资源而造成的一种互相等待。举个例子，一个箱子需要两把钥匙才可以打开，钥匙分别在两个人手中，这两个人互相抢占另外一个人的钥匙，导致箱子打不开。

Docker快速上手之部署SpringBoot项目

Posted on 2022-02-01

Docker是基于Go语言实现的云开源项目。

Docker的主要目标是“Build，Ship and Run Any App,Anywhere”，也就是通过对应用组件的封装、分发、部署、运行等生命周期的管理，使用户的APP（可以是一个WEB应用或数据库应用等等）及其运行环境能够做到“一次封装，到处运行”。

Linux 容器技术的出现就解决了这样一个问题，而 Docker 就是在它的基础上发展过来的。将应用运行在 Docker 容器上面，而 Docker 容器在任何操作系统上都是一致的，这就实现了跨平台、跨服务器。只需要一次配置好环境，换到别的机子上就可以一键部署好，大大简化了操作。

一、我所理解的Docker

我喜欢将Docker比喻成”方便面”，为什么说是方便面，之前部署项目环境的配置十分麻烦，换一台机器，就要重来一次，费力费时。举个栗子，我们部署一个SpringBoot项目，我们需要在服务器上面配置项目的运行环境，要安装各种各样的软件JDK/MySQL/Redis/nginx，安装和配置这些东西十分麻烦，下次需要换个服务器重新部署又得重新安装一遍，简直要命。而Docker能将项目连带着运行环境一同部署过去，就好像泡面，泡面所需的调料包还有工具都附加在了里面。一句话总结Docker的好处，Docker解决了运行环境和配置问题，方便做持续集成并有助于整体发布的容器虚拟化技术。

二、Docker与虚拟机的区别

	Docker容器	虚拟机（VM）
操作系统	与宿主机共享OS	宿主机OS上运行虚拟机OS
存储大小	镜像小，便于存储与传输	镜像庞大（vmdk、vdi等）
运行性能	几乎无额外性能损失	操作系统额外的CPU、内存消耗
移植性	轻便、灵活，适用于Linux	笨重，与虚拟化技术耦合度高
硬件亲和性	面向软件开发者	面向硬件运维者

三、Docker里面三个重要的概念Dockerfile、镜像(image)、容器(Container)

1.Dockerfile是用来构建Docker镜像的构建文件，是由一系列命令和参数构成的脚本。

Dockerfile体系结构：

FROM　　基础镜像，当前新镜像是基于哪个镜像的
MAINTAINER　　镜像维护者的姓名和邮箱地址
RUN　　容器构建时需要运行的命令
EXPOSE　　当前容器对外暴露出的端口
WORKDIR　　指定在创建容器后，终端默认登录进来的工作目录，一个落脚点
ENV　　用来在构建镜像过程中设置环境变量
AD　　将宿主机目录下的文件拷贝进镜像且ADD命令会自动处理URL和解压tar压缩包
COPY　　类似ADD，拷贝文件和目录到镜像中。将从构建上下文目录中<源路径>的文件/目录复制到新的一层的镜像内的<目标路径>位置
VOLUME　　容器数据卷，用于数据保存和持久化工作
CMD　　指定一个容器启动时要运行的命令。Dockerfile中可以有多个CMD指令，但只要最后一个生效，CMD会被docker run之后的参数替换
ENTRYPOINT　　指定一个容器启动时要运行的命令。ENTRYPOINT的目的和CMD一样，都是在指定容器启动程序及参数
ONBUILD　　当构建一个被继承的Dockerfile时运行命令，父镜像在被子继承后父镜像的onbuild被触发

2.镜像是一种轻量级、可执行的独立软件包，用来打包软件运行环境和基于运行环境开发的软件，它包含运行某个软件所需的所有内容，包括代码、运行时、库、环境变量和配置文件。

3.容器是用镜像创建的运行实例。它可以被启动、开始、停止、删除。每个容器都是相互隔离的、保证安全的平台。可以把容器看做是一个简易版的Linux环境（包括root用户权限、用户空间和网络空间等）和运行在其中的应用程序。

四、安装Docker

1.安装Docker依赖包

1	yum install -y yum-utils device-mapper-persistent-data lvm2

2.设置阿里云镜像源

1	yum-config-manager --add-repo https://mirrors.aliyun.com/docker-ce/linux/centos/docker-ce.repo

1	3.安装 Docker-CE

1	yum install docker-ce

4.查看是否安装成功

1	docker version

五、使用Docker安装MySQL

1.使用Docker拉取Mysql镜像(这里安装的是MySQL5.6版本)

1	docker pull mysql:5.6

2.安装mysql命令

1	docker run -d -p 3306:3306 --name mysql -v /ray/mysql/conf:/etc/mysql/conf.d -v /ray/mysql/logs:/logs -v /ray/mysql/data:/var/lib/mysql -e MYSQL_ROOT_PASSWORD=admin -d mysql:5.7

命令说明：

-p 3306:3306　　将主机的3306端口映射到docker容器的3306端口
--name mysql　　运行服务器名字
-v /ray/mysql/conf:/etc/mysql/conf.d　　将主机/ray/mysql目录下的conf/my.cnf挂载到容器的/etc/mysql/conf.d
-v /ray/mysql/logs:/logs　　将主机/ray/mysql目录下的logs目录挂载到容器的/logs
-v /ray/mysql/data:/var/lib/mysql　　将主机/ray/mysql目录下的data目录挂载到容器的/var/lib/mysql
-e MYSQL_ROOT_PASSWORD=admin　　初始化root用户的密码
-d mysql:5.7　　后台程序运行mysql5.7

3.进入mysql

1	docker exec -it MySQL运行成功后的容器ID /bin/bash

然后即可登录MySQL，创建数据库，或者使用Navicat等工具创建数据库。

六、使用Docker安装Redis

1.使用Docker拉取Redis镜像(这里安装的是Redis3.2版本)

1	docker pull redis:3.2

2.安装redis命令(这里不对命令做解释)

1	docker run -p 6379:6379 -v /ray/redis/data:/data -v /ray/redis/conf/redis.conf:/usr/local/etc/redis/redis.conf -d redis:3.2 redis-server /usr/local/etc/redis/redis.conf --appendonly yes

加密码--requirepass "admin"

七、Docker部署SpringBoot项目

1.将项目打包成jar包(假设名字为myblog.jar)，并编写一个Dockerfile文件，文件内容如下：

# Docker image for springboot file run
# VERSION 0.0.1
# Author: Ray
# 基础镜像使用java
FROM java:8
# 作者
MAINTAINER Ray <185048761@qq.com>
# VOLUME 指定了临时文件目录为/tmp。
# 其效果是在主机 /var/lib/docker 目录下创建了一个临时文件，并链接到容器的/tmp
VOLUME /tmp
# 将jar包添加到容器中并更名为app.jar
ADD myblog.jar app.jar
# 运行jar包
RUN bash -c 'touch /app.jar'
ENTRYPOINT ["java","-Djava.security.egd=file:/dev/./urandom","-jar","/app.jar"]
#暴露8080端口
EXPOSE 80

2.将这两个文件一并上传至服务器中的同一个目录下面，进入该文件夹后执行此命令构建镜像：

1	docker build -t myblog.jar .

3.生成docker容器，并运行

1	docker run -d -p 80:80 myblog.jar

等一会儿，SpringBoot项目跑起来了后，就可以使用浏览器通过80端口进行访问了