8锁问题 8锁问题什么是锁？锁的对象是谁？package com.sw.lock8; import java.util.concurrent.TimeUnit; /** * @Author suaxi * @Date 2021/1/4 15:44 * 1、执行顺序：发短信 打电话 * 2、增加睡眠时间后，执行顺序：发短信 打电话 */ public class Test1 { public static void main(String[] args) { Phone1 phone1 = new Phone1(); new Thread(() ->{ phone1.sms(); },"A").start(); try { TimeUnit.SECONDS.sleep(1); } catch (InterruptedException e) { e.printStackTrace(); } new Thread(() ->{ phone1.call(); },"B").start(); } } class Phone1{ //synchronized锁的对象是方法的调用者 //两个方法用的是同一个锁，谁先拿到锁，谁先调用 public synchronized void sms(){ try { TimeUnit.SECONDS.sleep(3); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println("发短信"); } public synchronized void call(){ System.out.println("打电话"); } }package com.sw.lock8; import java.util.concurrent.TimeUnit; /** * @Author suaxi * @Date 2021/1/4 15:51 * 3、增加一个hello的普通方法后，执行顺序为 hello 发短信 * 4、两个对象、两个同步方式，执行顺序为 打电话 发短信 */ public class Test2 { public static void main(String[] args) { //两个对象、两个调用者、两把锁 Phone2 phone1 = new Phone2(); Phone2 phone2 = new Phone2(); new Thread(() ->{ phone1.sms(); },"A").start(); try { TimeUnit.SECONDS.sleep(1); } catch (InterruptedException e) { e.printStackTrace(); } new Thread(() ->{ phone2.call(); },"B").start(); } } class Phone2{ //synchronized锁的对象是方法的调用者 public synchronized void sms(){ try { TimeUnit.SECONDS.sleep(3); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println("发短信"); } public synchronized void call(){ System.out.println("打电话"); } //普通方法不受锁的限制 public void hello(){ System.out.println("hello"); } }package com.sw.lock8; import java.util.concurrent.TimeUnit; /** * @Author suaxi * @Date 2021/1/4 15:58 * 5、增加static后，执行顺序为 发短信 打电话 * 6、两个Phone对象，执行顺序依然为为 发短信 打电话 */ public class Test3 { public static void main(String[] args) { //两个对象的Class类模板只有一个，与问题5一样锁的也是Class Phone3 phone = new Phone3(); Phone3 phone1 = new Phone3(); new Thread(() ->{ phone.sms(); },"A").start(); try { TimeUnit.SECONDS.sleep(1); } catch (InterruptedException e) { e.printStackTrace(); } new Thread(() ->{ phone1.call(); },"B").start(); } } class Phone3{ //synchronized锁的对象是方法的调用者 //static静态方法，类一加载就有了，此处锁的是Class对象 public static synchronized void sms(){ try { TimeUnit.SECONDS.sleep(3); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println("发短信"); } public static synchronized void call(){ System.out.println("打电话"); } }package com.sw.lock8; import java.util.concurrent.TimeUnit; /** * @Author suaxi * @Date 2021/1/4 16:05 * 7、1个静态同步方法，1个普通同步方法，一个对象，执行顺序 打电话 发短信 * 8、1个静态同步方法，1个普通同步方法，两个对象，执行顺序 打电话 发短信 */ public class Test4 { public static void main(String[] args) { //两个对象的Class类模板只有一个，与问题5一样锁的也是Class Phone4 phone1 = new Phone4(); Phone4 phone2 = new Phone4(); new Thread(() ->{ phone1.sms(); },"A").start(); try { TimeUnit.SECONDS.sleep(1); } catch (InterruptedException e) { e.printStackTrace(); } new Thread(() ->{ phone2.call(); },"B").start(); } } class Phone4{ //静态同步方法 锁的是Class类模板 public static synchronized void sms(){ try { TimeUnit.SECONDS.sleep(3); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println("发短信"); } //普通同步方法 锁的是调用者，不需要再等待发短信的sleep睡眠时间，两个方法用的不是同一个锁 public synchronized void call(){ System.out.println("打电话"); } }小结一个对象，拿的是同一把锁两个对象拿的是不同的锁加了static静态方法之后，不论有几个对象，锁的只有一个Class模板普通方法不受锁的限制

生产者和消费者问题 生产者和消费者问题package com.sw.pc; /** * @Author suaxi * @Date 2021/1/4 14:25 */ public class A { public static void main(String[] args) { Data data = new Data(); new Thread(() ->{ for (int i = 0; i < 10; i++) { try { data.incr(); } catch (InterruptedException e) { e.printStackTrace(); } } },"A").start(); new Thread(() ->{ for (int i = 0; i < 10; i++) { try { data.decr(); } catch (InterruptedException e) { e.printStackTrace(); } } },"B").start(); } } class Data{ private int num = 0; //+1 public synchronized void incr() throws InterruptedException { if (num!=0){ //等待 this.wait(); } num++; System.out.println(Thread.currentThread().getName()+"===>"+num); //通知 this.notifyAll(); } //-1 public synchronized void decr() throws InterruptedException { if (num==0){ //等待 this.wait(); } num--; System.out.println(Thread.currentThread().getName()+"===>"+num); //通知 this.notifyAll(); } } 当增加多个线程时（A，B，C，D），会出现虚假唤醒的问题虚假唤醒：线程可以被唤醒，但不会被通知、中断或超时解决方法：将if判断换为whilepackage com.sw.pc; /** * @Author suaxi * @Date 2021/1/4 14:25 */ public class A { public static void main(String[] args) { Data data = new Data(); new Thread(() ->{ for (int i = 0; i < 10; i++) { try { data.incr(); } catch (InterruptedException e) { e.printStackTrace(); } } },"A").start(); new Thread(() ->{ for (int i = 0; i < 10; i++) { try { data.decr(); } catch (InterruptedException e) { e.printStackTrace(); } } },"B").start(); new Thread(() ->{ for (int i = 0; i < 10; i++) { try { data.incr(); } catch (InterruptedException e) { e.printStackTrace(); } } },"C").start(); new Thread(() ->{ for (int i = 0; i < 10; i++) { try { data.decr(); } catch (InterruptedException e) { e.printStackTrace(); } } },"D").start(); } } class Data{ private int num = 0; //+1 public synchronized void incr() throws InterruptedException { while (num!=0){ //等待 this.wait(); } num++; System.out.println(Thread.currentThread().getName()+"===>"+num); //通知 this.notifyAll(); } //-1 public synchronized void decr() throws InterruptedException { while (num==0){ //等待 this.wait(); } num--; System.out.println(Thread.currentThread().getName()+"===>"+num); //通知 this.notifyAll(); } } JUC版生产者消费者问题使用了Condition接口下的await()等待和signalAll()通知package com.sw.pc; import java.util.concurrent.locks.Condition; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; /** * @Author suaxi * @Date 2021/1/4 14:55 */ public class B { public static void main(String[] args) { Data1 data1 = new Data1(); new Thread(() ->{ for (int i = 0; i < 10; i++) { try { data1.incr(); } catch (InterruptedException e) { e.printStackTrace(); } } },"A").start(); new Thread(() ->{ for (int i = 0; i < 10; i++) { try { data1.decr(); } catch (InterruptedException e) { e.printStackTrace(); } } },"B").start(); new Thread(() ->{ for (int i = 0; i < 10; i++) { try { data1.incr(); } catch (InterruptedException e) { e.printStackTrace(); } } },"C").start(); new Thread(() ->{ for (int i = 0; i < 10; i++) { try { data1.decr(); } catch (InterruptedException e) { e.printStackTrace(); } } },"D").start(); } } class Data1{ private int num = 0; Lock lock = new ReentrantLock(); Condition condition = lock.newCondition(); //+1 public void incr() throws InterruptedException { try { lock.lock(); //加锁 while (num!=0){ //等待 condition.await(); } num++; System.out.println(Thread.currentThread().getName()+"===>"+num); //通知 condition.signalAll(); } catch (Exception e) { e.printStackTrace(); } finally { lock.unlock(); //解锁 } } //-1 public void decr() throws InterruptedException { try { lock.lock(); //加锁 while (num==0){ //等待 condition.await(); } num--; System.out.println(Thread.currentThread().getName()+"===>"+num); //通知 condition.signalAll(); } catch (Exception e) { e.printStackTrace(); } finally { lock.unlock(); //解锁 } } }Condition精准通知与线程唤醒可以看到之前的线程执行顺序是无序的。package com.sw.pc; import java.util.concurrent.locks.Condition; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; /** * @Author suaxi * @Date 2021/1/4 15:08 * Condition精准通知与线程唤醒 * ABC三个人互相打电话，A->B,B->C,C->A顺序执行 */ public class C { public static void main(String[] args) { Data2 data2 = new Data2(); new Thread(() ->{ for (int i = 0; i < 10; i++) { data2.CallA(); } },"A").start(); new Thread(() ->{ for (int i = 0; i < 10; i++) { data2.CallB(); } },"B").start(); new Thread(() ->{ for (int i = 0; i < 10; i++) { data2.CallC(); } },"C").start(); } } class Data2{ private int num = 1; // 1A 2B 3C private Lock lock = new ReentrantLock(); Condition condition1 = lock.newCondition(); Condition condition2 = lock.newCondition(); Condition condition3 = lock.newCondition(); public void CallA(){ try { lock.lock(); //加锁 while (num!=1){ //等待 condition1.await(); } num = 2; System.out.println(Thread.currentThread().getName()+"===>A打完了"); //通知 condition2.signal(); } catch (Exception e) { e.printStackTrace(); } finally { lock.unlock(); //解锁 } } public void CallB(){ try { lock.lock(); //加锁 while (num!=2){ //等待 condition2.await(); } num = 3; System.out.println(Thread.currentThread().getName()+"===>B打完了"); //通知 condition3.signal(); } catch (Exception e) { e.printStackTrace(); } finally { lock.unlock(); //解锁 } } public void CallC(){ try { lock.lock(); //加锁 while (num!=3){ //等待 condition3.await(); } num = 1; System.out.println(Thread.currentThread().getName()+"===>C打完了"); //通知 condition1.signal(); } catch (Exception e) { e.printStackTrace(); } finally { lock.unlock(); //解锁 } } } 通过不同的监视器监视线程来达到精准通知与线程唤醒的目的

Lock锁 Lock锁传统synchronized锁package com.sw; /** * @Author suaxi * @Date 2021/1/4 14:00 */ public class SaleTicket01 { public static void main(String[] args) { Ticket1 ticket1 = new Ticket1(); new Thread(() ->{for (int i = 0; i < 30; i++) ticket1.sale();},"A").start(); new Thread(() ->{for (int i = 0; i < 30; i++) ticket1.sale();},"B").start(); new Thread(() ->{for (int i = 0; i < 30; i++) ticket1.sale();},"C").start(); } } //synchronized锁 class Ticket1{ private int num = 30; public synchronized void sale(){ if (num>0){ System.out.println(Thread.currentThread().getName()+"卖了"+(num--)+"张票，剩余："+num); } } } Lock接口公平锁：先来后到非公平锁：可以插队（默认使用）package com.sw; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; /** * @Author suaxi * @Date 2021/1/4 11:28 */ public class SaleTicket02 { public static void main(String[] args) { Ticket ticket = new Ticket(); new Thread(() ->{for (int i = 0; i < 30; i++) ticket.sale();},"A").start(); new Thread(() ->{for (int i = 0; i < 30; i++) ticket.sale();},"B").start(); new Thread(() ->{for (int i = 0; i < 30; i++) ticket.sale();},"C").start(); } } //Lock锁 class Ticket{ private int num = 30; //1.新建重入锁 Lock lock = new ReentrantLock(); public void sale(){ lock.lock(); //2.加锁 try{ //业务代码 if (num>0){ System.out.println(Thread.currentThread().getName()+"卖了"+(num--)+"张票，剩余："+num); } }catch (Exception e){ e.printStackTrace(); }finally { lock.unlock(); //3.解锁 } } } synchronized与Lock的区别1、synchronized 是内置的Java关键字；Lock是一个Java类2、synchronized 无法判断 获取锁 的状态；Lock可以判断3、synchronized 会自动释放锁；Lock必须手动释放，如果不释放，会产生死锁问题4、synchronized 线程1（获得锁，阻塞），线程2（一直傻傻的等待）；Lock锁就不一定会一直等待下去lock.trylock()5、synchronized 可重入锁，不可中断，非公平；Lock 可重入锁，可以判断锁的状态，非公平（可自定义公平性）6、synchronized 适用于锁少量的同步代码问题；Lock 适合大量同步代码问题

Redis缓存穿透和雪崩 Redis缓存穿透和雪崩缓存穿透1、概念当用户查询一个数据时，发现redis内存数据库中没有（缓存未命中），于是向持久层数据库（如MySQL）查询，发现也没有要查询数据，造成本次查询请求失败。当用户很多时，出现了同样的情况，就会给持久层数据库造成很大的压力，这个过程就称为缓存穿透。解决方案布隆过滤器：它是一种数据结构，对所有可能查询的参数以hash形式存储，在控制层先进行校验，不符合则丢弃，从而避免对底层数据库的查询压力。缓存空对象：当存储层不命中后，将返回的空对象也存储起来，同时设置一个过期时间，之后有请求访问这个数据时从缓存中获取，以此保护后端数据源存在的问题：1、存储空对象需要更多的存储空间2、即使设置了过期时间，缓存层和存储层的数据存在一定时间的窗口期，对数据一致性要求较高的业务可能会出现问题缓存击穿1、概念：当某个key在过期的瞬间，有大量的并发请求访问（这类数据一般是热点数据），由于缓存过期，请求会去访问数据库来进行查询，并且回写缓存，会导致数据库瞬间压力过大，如：微博热搜宕机解决方案1、设置热点数据永不过期2、加互斥锁使用分布式锁，保证对于每个key只有一个线程去查询后端服务，其他没有获得分布式锁的线程需要等待缓存雪崩1、概念指在某一个时间段，缓存集中过期，查询请求全部落到后端数据库头上。例：双11的抢购，将热点商品集中放入缓存，设置过期时间为1点，到了凌晨1点，这批热点商品的缓存都过期了，这时对于这批商品的查询请求会全部落到数据库头上，对于数据库而言产生了周期性的波峰压力，底层存储的调用量指数上升，整个服务极易出现问题。解决方案1、Redis高可用增加Redis服务器，一台出现问题之后，其他节点正常工作（异地多活）2、限流降级在缓存失效后，通过加锁或队列来控制读取数据库写缓存的线程数量3、数据预热在正式部署之前，把可能的数据先访问一遍，在发生高并发请求之前手动触发缓存key，并设置不同的缓存过期时间，让缓存失效的时间尽量均匀，而不是集中过期

Redis哨兵模式 哨兵模式Redis 2.8引入，Rentinel哨兵模式哨兵模式是一种特殊的模式，它是一个独立的进程，其原理是：哨兵通过发送命令，等待redis服务器响应，从而监控运行的多个redis实例。哨兵模式的作用：通过发送命令，让redis服务器返回监控的运行状态（包括主从所有节点）当哨兵检测到master宕机，会在自动将slave切换为master，然后通过发布订阅模式通知其他的从服务器，修改配置文件，让他们切换主服务器节点一般情况下使用多个哨兵进行监控，且哨兵之间互相监督：当master节点出现问题时，哨兵01先检测到问题，系统并不会马上进行failover，仅仅是哨兵01主观的认为主节点不可用，这个过程称为主观下线。当其他哨兵也检测到主节点服务不可用时，并且数量达到了一定的值，那么哨兵之间就会进行一次投票，投票的结果由一个哨兵发起，进行failover（故障转移）操作。当切换完成之后，就会通过发布订阅模式让各个哨兵把自己监控的节点进行master主节点切换（从故障节点切换为投票后票数最高的节点），这个过程称为客观下线。测试1、redis配置文件目录下新建sentinel.conf#sentinel monitor 被监控名 主机地址 端口 1 sentinel monitor myredis 127.0.0.1 6380 1 #master节点密码（必须配置） sentinel auth-pass myredis 123456 后面的数字1，代表当主机出现问题时，slave进行投票决定让谁成为主机，票数最多的成为主节点2、启动哨兵模式[root@suaxi bin]# redis-sentinel sconf/sentinel.conf 19524:X 03 Jan 2021 21:33:38.148 # oO0OoO0OoO0Oo Redis is starting oO0OoO0OoO0Oo 19524:X 03 Jan 2021 21:33:38.148 # Redis version=6.0.9, bits=64, commit=00000000, modified=0, pid=1 started 19524:X 03 Jan 2021 21:33:38.148 # Configuration loaded _._ _.-``__ ''-._ _.-`` `. `_. ''-._ Redis 6.0.9 (00000000/0) 64 bit .-`` .-```. ```\/ _.,_ ''-._ ( ' , .-` | `, ) Running in sentinel mode |`-._`-...-` __...-.``-._|'` _.-'| Port: 26379 | `-._ `._ / _.-' | PID: 19524 `-._ `-._ `-./ _.-' _.-' |`-._`-._ `-.__.-' _.-'_.-'| | `-._`-._ _.-'_.-' | http://redis.io `-._ `-._`-.__.-'_.-' _.-' |`-._`-._ `-.__.-' _.-'_.-'| | `-._`-._ _.-'_.-' | `-._ `-._`-.__.-'_.-' _.-' `-._ `-.__.-' _.-' `-._ _.-' `-.__.-' 19524:X 03 Jan 2021 21:33:38.149 # Sentinel ID is dce67565b60ad27529a7bbcfe2457faa3a306c4e 19524:X 03 Jan 2021 21:33:38.149 # +monitor master myredis 127.0.0.1 6379 quorum 1 19524:X 03 Jan 2021 21:33:38.151 * +slave slave 127.0.0.1:6380 127.0.0.1 6380 @ myredis 127.0.0.1 6 19524:X 03 Jan 2021 21:33:38.155 * +slave slave 127.0.0.1:6381 127.0.0.1 6381 @ myredis 127.0.0.1 6 19524:X 03 Jan 2021 21:34:20.556 # +sdown master myredis 127.0.0.1 6379 19524:X 03 Jan 2021 21:34:20.556 # +odown master myredis 127.0.0.1 6379 #quorum 1/1 19524:X 03 Jan 2021 21:34:20.556 # +new-epoch 3 19524:X 03 Jan 2021 21:34:20.556 # +try-failover master myredis 127.0.0.1 6379 19524:X 03 Jan 2021 21:34:20.560 # +vote-for-leader dce67565b60ad27529a7bbcfe2457faa3a306c4e 3 19524:X 03 Jan 2021 21:34:20.560 # +elected-leader master myredis 127.0.0.1 6379 19524:X 03 Jan 2021 21:34:20.560 # +failover-state-select-slave master myredis 127.0.0.1 6379 19524:X 03 Jan 2021 21:34:20.622 # +selected-slave slave 127.0.0.1:6380 127.0.0.1 6380 @ myredis 1279 19524:X 03 Jan 2021 21:34:20.622 * +failover-state-send-slaveof-noone slave 127.0.0.1:6380 127.0.0.yredis 127.0.0.1 6379 19524:X 03 Jan 2021 21:34:20.674 * +failover-state-wait-promotion slave 127.0.0.1:6380 127.0.0.1 63is 127.0.0.1 6379 19524:X 03 Jan 2021 21:34:21.037 # +promoted-slave slave 127.0.0.1:6380 127.0.0.1 6380 @ myredis 1279 19524:X 03 Jan 2021 21:34:21.037 # +failover-state-reconf-slaves master myredis 127.0.0.1 6379 19524:X 03 Jan 2021 21:34:21.104 * +slave-reconf-sent slave 127.0.0.1:6381 127.0.0.1 6381 @ myredis 6379 19524:X 03 Jan 2021 21:34:22.060 * +slave-reconf-inprog slave 127.0.0.1:6381 127.0.0.1 6381 @ myred.1 6379 如果之前端口为6379的主节点重新恢复后，不能再成为主节点，哨兵模式会将其分配为slave从节点优缺点优点：1、哨兵模式集群基于主从复制模式2、主从切换，系统可用性更好缺点：1、哨兵集群容量很大，在线扩容不方便2、哨兵模式配置繁杂sentinel.conf详细配置# Example sentinel.conf # 运行端口默认26379 port 26379 # 工作目录 dir /tmp # 监控的redis主节点 # master-name 可自定义命名的主节点名字，只能由字母A-Z、数字0-9、.-_等字符组成 # quorum 配置多少个哨兵节点同意认为master节点失联，这时客观认为主节点已失联（客观下线） # sentinel monitor <master-name> <ip> <redis-port> <quorum> sentinel monitor myredis 127.0.0.1 6379 1 # 设置哨兵连接主节点的密码 # sentinel auth-pass <master-name> <password> sentinel auth-pass myredis 123456 # 指定多少毫秒后，主节点没有应答哨兵监控，哨兵主观上认为主节点出现问题，默认为30秒 # sentinel down-after-milliseconds <master-name> <milliseconds> sentinel down-after-milliseconds myredis 30000 # 指定在发生failover[故障转移]主备切换时，有多少个slave节点同时对新的master节点进行同步 # 数字越小，完成failover的时间越长，数字越大，越多的slave节点会因replication而不可用 # 可将这个值设为1，来保证每次只有一个slave节点处于不能处理命令请求的状态 # sentinel parallel-syncs <master-name> <numslaves> sentinel parallel-syncs myredis 1 # 故障转移超时时间 failover-timeout，可用于以下几个方面 # 1.同一个sentinel对同一个master两次failover之间的时间间隔 # 2.当一个slave从一个出现问题的master那里同步数据开始计算时间，直到slave被纠正为向正确的master同步数据时 # 3.当想要取消一个正在进行failover所需要的时间 # 4.当进行failover时，配置所有slave指向新的master所需的最大时间（即使过了这个时间，slave依然会被配置到正确的master，此时不遵循parallel-syncs规则） # 默认为三分钟，单位毫秒 # sentinel failover-timeout <master-name> <milliseconds> sentinel failover-timeout myredis 180000 # SCRIPTS EXECUTION # 配置当某一事件发生时所需要执行的脚本，以此来通知管理员 # 1.若脚本执行后返回1，那么该脚本稍后会被再次执行，默认次数为10 # 2.若脚本执行后返回2，或者比2更高的一个返回值，脚本将不会被重复执行 # 3.如果脚本在执行过程中收到系统中断信号而被终止，则与返回值为1时的操作一致 # 一个脚本的最大执行时间为60s，将会被SIGKILL信号终止，之后重新执行 # 通知型脚本:当sentinel有任何警告级别的事件发生时（比如说redis实例的主观失效和客观失效等等），将会去调用这个脚本，这时这个脚本应该通过邮件，SMS等方式去通知系统管理员关于系统不正常运行的信息。调用该脚本时，将传给脚本两个参数，一个是事件的类型，一个是事件的描述。如果sentinel.conf配置文件中配置了这个脚本路径，那么必须保证这个脚本存在于这个路径，并且是可执行的，否则sentinel无法正常启动成功。 # sentinel notification-script <master-name> <script-path> sentinel notification-script myredis /var/redis/notify.sh # 客户端重新配置主节点参数脚本 # 当一个master由于failover而发生改变时，这个脚本将会被调用，通知相关的客户端关于master地址已经发生改变的信息。 # 以下参数将会在调用脚本时传给脚本: # <master-name> <role> <state> <from-ip> <from-port> <to-ip> <to-port> # 目前<state>总是“failover”, # <role>是“leader”或者“observer”中的一个。 # 参数 from-ip, from-port, to-ip, to-port是用来和旧的master和新的master(即旧的slave)通信的 # 这个脚本应该是通用的，能被多次调用，不是针对性的。 # sentinel client-reconfig-script <master-name> <script-path> sentinel client-reconfig-script myredis /var/redis/reconfig.sh

Redis主从复制 Redis主从复制默认情况下，每台redis服务器都是master主节点1、概念将一台redis服务器的数据，复制到其他redis服务器上，前者称为主节点（master/leader），后者称为从节点（slave/follower）,数据的复制是单向的，只能由主节点到从节点，Master以写为主，Slave以读为主。一般来说，一个主节点可以有多个从节点，但一个从节点，只能有一个主节点。2、作用数据冗余：实现了数据的热备份，是持久化之外的另一种数据冗余方式故障恢复：当主节点出现问题时，可以由从节点提供服务，实现快速的故障恢复（服务冗余）负载均衡：在主从复制的基础上，配合读写分离，在redis写数据时应用连接主节点，读数据时应用连接从节点，以此分担服务器负载（读多写少）高可用（集群）基石：主从复制是哨兵模式和集群的基础3、一主二从配置：此处以单机多集群为例：1、修改端口，2、修改日志文件名6380.log (为避免单机多集群情况下文件名重复，以下同理)，3、修改pid文件名redis_6380.pid，4、修改dump文件名dump6380.rdb,5、配置主节点地址replicaof 127.0.0.1 63796、配置主节点密码masterauth 1234566380、6381从节点以此类推需特别注意密码配置的方式：主节点redis.conf配置密码的方式为：requirepass 123456从机redis.conf配置密码的方式为：masterauth 123456如果主节点配置了密码，从机一定要配置，不然连不上主节点补充：也可以通过命令行配置（重启redis服务之后配置失效）127.0.0.1:6380> SLAVEOF 127.0.0.1 6379 #SLAVEOF host port OK 127.0.0.1:6380> info replication #查看节点信息 # Replication role:slave master_host:127.0.0.1 master_port:6379 #主节点 master_link_status:up master_last_io_seconds_ago:5 master_sync_in_progress:0 slave_repl_offset:0 slave_priority:100 slave_read_only:1 connected_slaves:0 master_replid:ed6d6cc071c41636468cecadd52b61d6c9b130af master_replid2:0000000000000000000000000000000000000000 master_repl_offset:0 second_repl_offset:-1 repl_backlog_active:1 repl_backlog_size:1048576 repl_backlog_first_byte_offset:1 repl_backlog_histlen:0 127.0.0.1:6380> 查看主节点信息：127.0.0.1:6379> info replication # Replication role:master #角色 connected_slaves:2 slave0:ip=127.0.0.1,port=6380,state=online,offset=42,lag=0 #从机信息1 slave1:ip=127.0.0.1,port=6381,state=online,offset=42,lag=1 #从机信息2 master_replid:ed6d6cc071c41636468cecadd52b61d6c9b130af master_replid2:0000000000000000000000000000000000000000 master_repl_offset:42 second_repl_offset:-1 repl_backlog_active:1 repl_backlog_size:1048576 repl_backlog_first_byte_offset:1 repl_backlog_histlen:42 127.0.0.1:6379> 4、复制原理1、Slave启动连接到Master后会发送一个sync同步命令2、Master接到命令，启动后台的存盘进程，同时收集所有接收到的修改数据集的命令，在后台进程执行完毕之后，master传送整个数据文件到slave，同时完成一次完全同步全量复制：slave接收到数据文件之后，将其存盘并加载到内存中增量复制：Master继续将收集到的新的修改命令依次传给Slave，完成同步5、层层节点

Redis发布订阅 Redis发布订阅发布订阅（pub/sub）是一种消息通信模式：发送者（pub）发送消息，订阅者（sub）接收消息命令命令及描述PSUBSCRIBE pattern [pattern...]订阅一个或多个符合给定模式的频道PUBSUB subcommand [argument [argument...]]查看订阅与发布系统状态PUBLISH channel message将信息发送到指定的频道PUNSUBSCRIBE [pattern [pattern...]]退订所有给定模式的频道SUBSCRIBE channel [channel...]订阅给定的一个或多个的频道信息UNSUBSCRIBE [channel[channel...]]退订指定的频道测试订阅者：127.0.0.1:6379> SUBSCRIBE test #订阅名为test的频道 Reading messages... (press Ctrl-C to quit) #订阅之后自动监听 1) "subscribe" 2) "test" 3) (integer) 1 # 监听消息，等待推送读取 1) "message" #消息 2) "test" #来自哪个频道 3) "nihao" #消息内容 1) "message" 2) "test" 3) "xiexie" 发布者：127.0.0.1:6379> PUBLISH test nihao #发布者发布消息到test频道 (integer) 1 127.0.0.1:6379> PUBLISH test xiexie (integer) 1 127.0.0.1:6379> 原理常用于微信公众号、微博、实时聊天等通过SUBSCRIBE命令订阅某频道后，redis-server里维护了一个字典，字典的键可以看作是一个个的频道，而字典的值则是一个链表，链表中保存了所有订阅这个频道（channel）的客户端。换句话说，就是将客户端添加到指定频道（channel）的订阅链表中。通过PUBLISH命令向订阅者发送消息，redis-server会使用给定的频道作为键，在它所维护的频道（channel）字典中查找订阅了这个频道的所有客户端的链表，遍历这个链表，将消息发布给所有订阅者。

Redis持久化 Redis持久化RDB（Redis DataBase）在指定的时间间隔内将内存中的数据集体写入磁盘（Sanpshot快照），恢复时将快照文件直接读取到内存中触发机制1、满足save规则的情况时2、执行flushall3、退出redis时，也会产生rdb文件如何恢复rdb文件将rdp文件放在redis的启动目录即可，redis启动时会自动检查dump.rdb文件，恢复其中的数据# 查看目录 127.0.0.1:6379> config get dir 1) "dir" 2) "/usr/local/bin" 127.0.0.1:6379> 优点：适合大规模的数据恢复对数据的完整性不敏感的业务缺点：操作需要一定的时间间隔，特殊情况下，最后一次的修改数据可能会丢失（宕机）fork进程时，会占用一定的空间AOF（Append Only File）以日志的形式记录每个写的操作，将Redis执行过程的所有指令记录下来（不记录读操作），只追加文件不允许改写文件，redis启动时会读取该文件重新构建数据，即：启动时根据日志文件的内容，将写操作从头到尾执行一次已完成数据恢复。redis.conf配置配置文件默认不开启aof，需手动设置为yes，重启redis即可生效重写规则当文件大小超过64mb时，会fork一个新的进程来重写文件（文件占用的内存空间会越来越大）redis-check-aofaof文件有误，redis启动失败如果aof文件中存在错误，可用redis-check-aof --fix命令修复这个文件修复完成之后重新启动redis即可优点：没修改一次，都会进行同步，数据的完整性更好效率更高缺点：每秒同步一次，极端情况下可能会丢失最后一秒的数据aof文件远大于rdb文件，修复速度也比rdb慢整体运行效率比rdb慢RDB与AOF对比：图片来源：狂神说Java