redis 数据库的优化

　　redis是一个内存数据库，速度高的同时也加大消耗了内存资源，为了减少内存消耗，redis做了很多优化。

32bit 和64 bit

　　32bit相对于64bit有个很大的优势，就是指针空间占用很少，但是总内存不能超过4G。但是4G够了。

ziplist

　　对于redis的hashmap和zset来说，如果内部元素很少，依旧使用二维结构会很浪费空间。目前有个叫做ziplist的数据结构，该结构是一个紧凑的字节数组结构，每个元素都是紧挨着的。

intset

　　set集合的元素很少的时候，且都是整数的时候，会采用intset这个数据结构。这是一个紧凑的整形数组结构。如果加入了字符串就会转换为hashmap结构。

转换条件

　　hash-max-ziplist-entries 512 # hash 的元素个数超过 512 就必须用标准结构存储
hash-max-ziplist-value 64 # hash 的任意元素的 key/value 的长度超过 64 就必须用标准结构存储
list-max-ziplist-entries 512 # list 的元素个数超过 512 就必须用标准结构存储
list-max-ziplist-value 64 # list 的任意元素的长度超过 64 就必须用标准结构存储
zset-max-ziplist-entries 128 # zset 的元素个数超过 128 就必须用标准结构存储
zset-max-ziplist-value 64 # zset 的任意元素的长度超过 64 就必须用标准结构存储
set-max-intset-entries 512 # set 的整数元素个数超过 512 就必须用标准结构存储

redis 垃圾回收策略

　　当你从redis里面删除了1GB的key时，你会发现内存变化不大。因为操作系统回收数据是以页为单位的，即使这个页里面还有一个key,那么这个页也不会被回收。这个1Gb的key分散到各个页面。因此内存不会立即被回收。
　　当然你使用flushdb这个指令，可以强制回收。redis虽然无法保证立即回收已删除key的内存，但是他会优先使用尚未被回收的空闲内存。

redis内存分配算法

　　redis的内存分配是直接交给第三方库来管理，jemalloc这个facebook的第三方库来管理。因为内存管理很复杂，需要考虑到内存碎片,性能优化，效率等，为了redis的简单高效，于是使用第三方插件.

Redis的事务

　　基本每个成熟的数据库都有事务，redis也不例外。
　　redis通过multi(开始事务)，exec指事务的执行，discard表示事务的丢弃。
　　multi开始后，后面的指令不是直接执行，而是缓存在服务器redis的事务队列中，而服务器一旦收到exec指令，才开始执行整个事务队列，执行完毕后统一返回结果，由于redis的单线程特性，故不用担心其被其他指令打扰。
　　redis的事务没有原子性，中间的失败了，不仅不回滚，后面的反而继续执行。
　　我们可以用discard指令来丢弃redis事务队列里面的数据。
　　我们一般用pipeline来发送事务里面的指令一起发送，减少网络io次数。
　　但是redis里面无法做乘除法则，有时候需要乘除，那该怎么办呢？也就是说一个事务里面无法完成所有的操作，我们必须通过2个甚至多个事务来完成，这样就会有并发问题，因为这个不是串行的，可能2个事务之间，会有别的事务插进来。于是有了watch机制。就是开启事务之前，先watch要修改的key,然后在事务修改，当修改遇见问题则会报错，是一个乐观锁。

ps：读redis掘金小册

pipe管道命令

　　管道命令本质上是为了减少网络io交互产生的，其实现是在客户端实现的，客户端将多个请求并到一起发送，然后一起返回，有效减少了io次数。
　　这便是管道操作的本质，服务器根本没有任何区别对待，还是收到一条消息，执行一条消息，回复一条消息的正常的流程。客户端通过对管道中的指令列表改变读写顺序就可以大幅节省 IO 时间。管道中指令越多，效果越好。
　　我们可以用： redis-benchmark -t set -P 2 -q来进行压测，看看其QPS。

io操作真正的耗时点在哪里

１．客户端进程调用write将消息写到操作系统内核为套接字分配的发送缓冲send buffer。

２．客户端操作系统内核将发送缓冲的内容发送到网卡，网卡硬件将数据通过「网际路由」送到服务器的网卡。

３．服务器操作系统内核将网卡的数据放到内核为套接字分配的接收缓冲recv buffer。

４．服务器进程调用read从接收缓冲中取出消息进行处理。

５．服务器进程调用write将响应消息写到内核为套接字分配的发送缓冲send buffer。

６．服务器操作系统内核将发送缓冲的内容发送到网卡，网卡硬件将数据通过「网际路由」送到客户端的网卡。

７．客户端操作系统内核将网卡的数据放到内核为套接字分配的接收缓冲recv buffer。

８．客户端进程调用read从接收缓冲中取出消息返回给上层业务逻辑进行处理。

　　我们开始以为 write 操作是要等到对方收到消息才会返回，但实际上不是这样的。write 操作只负责将数据写到本地操作系统内核的发送缓冲然后就返回了。剩下的事交给操作系统内核异步将数据送到目标机器。但是如果发送缓冲满了，那么就需要等待缓冲空出空闲空间来，这个就是写操作 IO 操作的真正耗时。

　　我们开始以为 read 操作是从目标机器拉取数据，但实际上不是这样的。read 操作只负责将数据从本地操作系统内核的接收缓冲中取出来就了事了。但是如果缓冲是空的，那么就需要等待数据到来，这个就是读操作 IO 操作的真正耗时。

　　所以对于value = redis.get(key)这样一个简单的请求来说，write操作几乎没有耗时，直接写到发送缓冲就返回，而read就会比较耗时了，因为它要等待消息经过网络路由到目标机器处理后的响应消息,再回送到当前的内核读缓冲才可以返回。这才是一个网络来回的真正开销。

　　而管道操作就是只用等一个网路来回从而减少时间开销，实际上减少了多余的通信次数。

ps，读redi小册

redis的几个特性

　　单线程：没错redis是单线程的。
　　快：因为所有数据都在内存里面。
　　单线程如何高效处理大并发请求：多路复用。非阻塞io。其实现和java nio,netty都是基于多路复用实现的，通过select函数不断轮训请求，然后判断请求类型，就行不同操作，比如连接请求，读请求，写请求等(具体实现请看我的博客关于java nio的一章)
　　Redis 会将每个客户端套接字都关联一个指令队列。客户端的指令通过队列来排队进行顺序处理，先到先服务。
　　Redis 同样也会为每个客户端套接字关联一个响应队列。Redis 服务器通过响应队列来将指令的返回结果回复给客户端．
　　Redis 的定时任务会记录在一个称为最小堆的数据结构中。这个堆中，最快要执行的任务排在堆的最上方。在每个循环周期，Redis 都会将最小堆里面已经到点的任务立即进行处理。处理完毕后，将最快要执行的任务还需要的时间记录下来，这个时间就是select系统调用的timeout参数。因为 Redis 知道未来timeout时间内，没有其它定时任务需要处理，所以可以安心睡眠timeout的时间。

Redis 通信协议

　　Redis通信协议是一个文本协议RESP。优势是实现简单，解析性能好。
Redis协议将所传输的数据分为5个最小类型。单元结束统一加\r\n:

单行字符串 以 + 符号开头。
多行字符串 以 $ 符号开头，后跟字符串长度。
整数值 以 : 符号开头，后跟整数的字符串形式。
错误消息 以 - 符号开头。
数组 以 * 号开头，后跟数组的长度。
空串 用多行字符串表示，长度填 0。

　　set author codehole会被序列化成下面的字符串。

*3
$3
set
$6
author
$8
codehole

redis持久化

　　redis持久化有2个机制。一个是RDB(快照:全量备份，内存数据的2进制序列化形式)，一个是AOF(增量备份，记录内存数据的修改指令文本)。
　　快照原理：内存快照要求redis必须进行文件io操作，而这个操作是无法阻塞的。如果单线程在服务线上请求还要进行文件io操作，那么性能会变得很差，如果不阻塞线上的业务，便持久化边相应请求，持久化同时，内存数据结构还在变化，这怎么玩呢？redis采用多进程来进行处理。redis在持久化时fork一个子进程，快照持久化交给子进程来处理，父进程继续处理客户端请求。子进程刚产生时和父进程共享内存里面的代码段和数据段。所以不会导致内存突然变大。子进程做持久化，不会修改这个内存数据，只会对其结构不断遍历读取，然后序列化之后写到磁盘，但是父进程则是不断相应客户端请求，然后对内存数据不断修改。然后使用操作系统的写时复制机制，在内存里面，数据是一页一页的，当父进程修改内存数据时会把这个数据所在的那一页辅助一份出来，进行修改，等子进程顺利遍历完了，在替换合并。
　　AOF原理：就是一个日志存储着redis创建后的所有修改指令。redis是先执行指令在存日志的。同时redis长期运行会导致日志庞大，重启时间长，导致redis长期无法对外提供服务。所以需要瘦身，redis提供了bgrewriteaof 指令对aof日志瘦身，其原理是开辟一个子进程对内存进行遍历转换生成一系列的redis操作指令，序列化到一个新的aof日志文件里面，序列化后，再讲操作期间发生的增量aof日志加到新的aof日志文件里面，追加完毕即可代替旧的aof日志。aof日志以文件方式存在的，当程序对aof日志文件进行写操作时，实际上是将内容写到了内核为文件描述符分配的内存缓存，但是如果系统宕机了aof日志没来得及刷到磁盘，该如何。linux提供了fsync函数可以将日志文件强刷到磁盘。所以redis一般每隔一秒执行一次fsunc操作，使得尽可能减少数据丢失。
　　快照是通过开启子进程的方式进行的，它是一个比较耗资源的操作。

　　遍历整个内存，大块写磁盘会加重系统负载
　　AOF 的 fsync 是一个耗时的 IO 操作，它会降低 Redis 性能，同时也会增加系统 IO 负担
　　所以通常 Redis 的主节点是不会进行持久化操作，持久化操作主要在从节点进行。从节点是备份节点，没有来自客户端请求的压力，它的操作系统资源往往比较充沛。

　　但是如果出现网络分区，从节点长期连不上主节点，就会出现数据不一致的问题，特别是在网络分区出现的情况下又不小心主节点宕机了，那么数据就会丢失，所以在生产环境要做好实时监控工作，保证网络畅通或者能快速修复。另外还应该再增加一个从节点以降低网络分区的概率，只要有一个从节点数据同步正常，数据也就不会轻易丢失。

混合持久化

　　redis4.0之后将rdb文件的内容和增量aof文件存在一起，这里的aof日志则是自持久化到持久化结束的增量aof文件.这样redis重启的时候则是先加载rdb文件，在加载aof日志文件。

目的

　　公司框架为这个。且大多数同学不了解这个怎么玩。为了避免采坑太多，这里先学习总结下。

代码地址以及相关介绍

　　本代码使用spring+spring mvc+spring data jpa+hibernate+mysql(8.0之后的版本)
　　git地址：https://github.com/skydh/SpringJpaLearn
　　为了方便公司的小伙伴们看。
　　公司内部git地址：http://git.ipo.com/donghang846/spring-jpa-learn.git 　　

单表的3种简单查询模式以及注意点。

　　１．直接调用接口自带的方法。我们继承的是JpaRepository接口。主要有

count() 获取表多少数据。
findOne() 根据id获取数据。
save() 保存数据。
等。

　　注意点：

　　第一点.getone()和findone()作用一样都是根据id获取对象，但是getone是懒加载，没有配置的要报错，还有实体转json的时候也会报错。再看了源代码对接口的解释findOne返回实体，而getOne则是返回实体的引用。总而言之，最好用findOne.
　　
　　第二点：我们调用save方法时，如果主键不是自增的，我们必须在entity里面增加id,因为jpa会先根据这个id在数据库里面查询数据，如果有数据，那么生成的sql则是更新语句，也就是这个save方法同时承担了更新的职责。
　　
　　２．在对应的dao接口里面写好方法，调用这个接口即可调用sql获取数据。spring-data-jpa会根据方法的名字来自动生成sql语句，我们只需要按照方法定义的规则即可。规则如我网上找的表。

　　案例如下：

public interface UserDao extends JpaRepository<User, Serializable> {

/**
 * 根据name,id 找这个人
 * @param name
 * @param id
 * @return
 */
User findByNameAndId(String name,Integer id);

}

　　３.当上面２方法都不满足你的需求的时候，或者想自己写sql稳定精确一点的。可以自定义sql。案例如下：

public interface UserDao extends JpaRepository<User, Serializable> {

@Query(value="select * from User where address= :address" ,nativeQuery = true)
List<User> findUserByAddress(@Param("address")String address );
}

　　这里有几个注意点
　　１．这里我推荐使用　nativeQuery = true　方式，直接写sql，因为确定了，我们的数据库是mysql,所以我们直接写mysql的sql即可。
　　２．除了查询语句这种快照读之外，其余所有的语句都要加　@Transactional
　　@Modifying
　　例子如下：

@Transactional
@Modifying
@Query(value="update User set address= :address where name= :name")
void updateData(@Param("name")String name,@Param("address")String address);

　　
　　因为jpa对于非快照操作，要求必须都是事务操作。

多表查询

　　这边鉴于大多数同学都喜欢mybatis，喜欢写原生sql,这边采用EntityManager这个类来进行多表查询，相对于specification，确实利于优化和调试。
　　话不多说，上代码：

@Repository
public class UserDao {
@Autowired 
private EntityManager entityManager;
public void getUser() {
    StringBuilder sb = new StringBuilder();
    sb.append("select a.name as name ,b.name as name1 from user a inner join Orders b on a.id=b.user_id");
    Query query = entityManager.createNativeQuery(sb.toString());
    List<Object> list = query.getResultList();
    for (Object user : list) {
        System.out.println(user);
    }
}
}

　　如此可以多表查询，且很方便。我们可以在自己拼装sql的时候加逻辑判断。
　　几个注意点：
　　１．这个有的表的字段有重复名的我们必须起别名，不然报错。这边最好用有Native关键字的方法，便于我们使用原生sql.

Query query1 = entityManager.createNativeQuery(sb.toString(),UserEntity.class);

　　第一个返回的都是一个个Object对象，我们要自己拼装成自己需要的vo,但是当sql返回的都是一个表的数据时，我们可以在后面传这个这个类的Class对象。这样不用拼装sql了。但是坑爹的是，我们必须将这个entity的所有字段都要对应上才行。
　　由于上面要一个个对应字段，太麻烦了，这边写了一个工具类，帮助大家提高效率，和代码工整度。代码就不细说了，有问题联系我修改哈。

什么是接口限流

　　为什么要有接口限流？场景如下：１．接口限流，比如我们对外一个接口，我们要限制一个用户单位时间访问次数？２．系统要限定用户的某个行为在指定的时间A里只能允许发生 N 次。

实现方式

　　我们用zset来实现这个方案。我们将用户id和事件id生成这个zset的容器id,当发生一次时，我们将当前时间作为score,同时也作为value,放到集合里面。然后删除这个元素加入前限定时间A之外的数据。然后将集合的所有数据设置为过了A时间就过期.然后取出这个集合有多少数据，判断是否超标。

public boolean isActionAllowed(String userId, String     actionKey, int period, int maxCount) {
    String key = String.format("hist:%s:%s", userId, actionKey);
    long nowTs = System.currentTimeMillis();
    Pipeline pipe = jedis.pipelined();
    pipe.multi();
    pipe.zadd(key, nowTs, "" + nowTs);
    pipe.zremrangeByScore(key, 0, nowTs - period * 1000);
    Response<Long> count = pipe.zcard(key);
    pipe.expire(key, period + 1);
    pipe.exec();
    pipe.close();
    return count.get() <= maxCount;

}

漏斗限流策略

public class FunnelRateLimiter {
static class Funnel {
int capacity;//容量
float leakingRate;//漏水速度
int leftQuota;//剩余容量
long leakingTs;//上次漏水时间

public Funnel(int capacity, float leakingRate) {
  this.capacity = capacity;
  this.leakingRate = leakingRate;
  this.leftQuota = capacity;
  this.leakingTs = System.currentTimeMillis();
}

void makeSpace() {
  long nowTs = System.currentTimeMillis();
  long deltaTs = nowTs - leakingTs;
  int deltaQuota = (int) (deltaTs * leakingRate);
  if (deltaQuota < 0) { // 间隔时间太长，整数数字过大溢出
    this.leftQuota = capacity;
    this.leakingTs = nowTs;
    return;
  }
  if (deltaQuota < 1) { // 腾出空间太小，最小单位是1
    return;
  }
  this.leftQuota += deltaQuota;
  this.leakingTs = nowTs;
  if (this.leftQuota > this.capacity) {
    this.leftQuota = this.capacity;
  }
}
boolean watering(int quota) {
  makeSpace();
  if (this.leftQuota >= quota) {
    this.leftQuota -= quota;
    return true;
  }
  return false;
}
}
private Map<String, Funnel> funnels = new HashMap<>();
public boolean isActionAllowed(String userId, String actionKey, int capacity, float leakingRate) {
String key = String.format("%s:%s", userId, actionKey);
Funnel funnel = funnels.get(key);
if (funnel == null) {
  funnel = new Funnel(capacity, leakingRate);
  funnels.put(key, funnel);
}
return funnel.watering(1); // 需要1个quota
}
}

　　
　　Funnel 对象的 make_space 方法是漏斗算法的核心，其在每次灌水前都会被调用以触发漏水，给漏斗腾出空间来。能腾出多少空间取决于过去了多久以及流水的速率。Funnel 对象占据的空间大小不再和行为的频率成正比，它的空间占用是一个常量。
　　我们观察 Funnel 对象的几个字段，我们发现可以将 Funnel 对象的内容按字段存储到一个 hash 结构中，灌水的时候将 hash 结构的字段取出来进行逻辑运算后，再将新值回填到 hash 结构中就完成了一次行为频度的检测。

新方案

　　redis4.0出了一个新的模块。叫redis-cell。该模块也使用了漏斗算法，且提供了限流指令。
　　cl.throttle sd:sd 15 30 60
　　意思是15的容量，每60秒就丢弃30个元素。

由来

　　情景：当我们使用新闻客户端看新闻时，他会不断推送新的内容，他每次推荐时都要去重，那么问题来了，客户端如何实现推送去重呢？
　　当然不会采用记录用户看过的内容，第一数据量大，第二不断exists消耗性能太多。缓存也不行，时间久了数据量太大。
　　这个时候布隆过滤器出现了。

什么是布隆过滤器

　　这是一个不怎么精确地set结构，当你使用它的contains方法判断某个数据是否存在时，他可能误判，布隆过滤器是不精确的。只要参数设置合理，那么其精确度是可控的。其判断一个值存在时，其可能不存在，判断一个值不存在时，那么一定不存在。

用法

　　bf.add 添加元素，bf.exists,查询元素是否存在，其用法和set集合差不多，批量命令是bf.madd指令，批量查询则是bf.mexists指令。我们创建的布隆过滤器都是默认的过滤器，在我们第一次add的时候自动创建。，但是为了提供高精确率。我们可以用bf.reserve指令来修改其参数即可。第一个为其名字，第二个为
client.createFilter(“codehole”, 50000, 0.001);第二个参数为预计存放数据量，第三个为误差率。误差率越低，需要空间越大。

实现原理

　　底层是一个大型的位数组，和几个无偏的hash函数（无偏就是让hash值分布均匀）。添加值的时候，会用多个hash函数对key进行hash运算，然后对应数组上不同点，再把这些点全部置为1。询问key是否存在时，和add一样，也是用hash算法吧这些位置都计算出来，看看是否都为1，只要有一个为0，那么这个值不存在。当然如果都是1，也不一定都存在，可能是其他计算的点正好都覆盖了。

由来

　　PV和UV,统计pv很简单，对每个网页加一个独立的redis计数器即可，这个计数器+当天的日期，每来一个请求，就incrby一次即可。但是UV不一样要求一个用户只能算一次，或许你想到了可以用set,把每一个页面放到set里面，当一个请求来了，就将其sadd进去，也可以通过scard取出这个集合的大小，但是有个问题，那就是访问量很大的时候，这个set集合就会很大。而这个redis提供的HyperLogLog数据结构来解决该问题。该数据结构提供一个不精确去重计数方案。

使用方法

　　HyperLogLog提供了2个指令pfadd,pfcount 根据字面意思很好理解，一个是增加计数，一个是获取计数，pfadd的用法和set集合的asdd一样的，pfcount和scard用法一样的，直接获取计数值。pfmerge将２个HyperLogLog合并。
　　pfadd codehole user1
　　其底层实现是稠密矩阵。一个我也不懂的数据结构 　　

ps：读掘金小册笔记

位图的产生和作用

　　开发时我们常常需要存储一些bool型数据类型，比如用户的签到记录，签了为1，没签为0，如果用普通的key/value，每个用户要记录365个，当用户多的时候，需要的存储太多，为了解决这个问题，redis提供了位图数据结构，每天的签到记录只占据一个位，365天就是365位，也就是46个字节，其内容也是普通的字符串，也就是byte数组，可以用get,set获取和设置整个数组，也可以用getbite/setbite来将其作为数组来操作。

操作

　　setbit a 1 1;这个就是设置这个a这个数组第二位位1，getbite a 1,获取a这个byte数组的第二位的数字，如果直接用get a 则是获得到这个byte数组对应的Ascll值，用时我们也可以先set字符，然后getbite 其数组。总之，这2套操作是互通的。

统计查找

　　redis提供了位图统计指令bitcount和位图查找指令bitpos,bitcount是统计指定位置范围内1的个数，bitpos用来查找指定范围内出现的第一个0或者1。后面的2个参数指的是字符索引。比如bitcount w 0 0,表示w字节数组第一个字符中1的位数。比如bitcount w 0 1,则是查询前2个字符中1的位数。bitpos w 0,表示查询w这个字节数组第一个出现0的位置，bitpop w 1 2 2,表示查询第三个字符中第一个1出现的位置。

批量处理

　　１．前面getite都是单个单个执行的。如果想批量执行，就要用bitfield这个指令。
　　比如：bitfield w get u4 0,从第一个位置开始取4个数，结果是无符号数（u）,bitfield w get i3 2,从第三个数开始取3个数，结果是有符号数（i）.所谓有符号就是第一位是符号位，有符号数最多可以获取64位，无符号只能63位（redis协议中integer就是有符号数）
　　 bitfield w set u8 8 97 这个就是把第二个字符改成ａ，因为ａ的ASCII码的值是97.这就可以批量修改字节数组的值。

　　bitfield w incrby u4 2 1 这个就是从第三个位开始，对接下来的无符号数＋１，既然是自增操作，那么就可能出现溢出如果增加了正数就会上溢，如果增加的负数就会下溢出，redis默认的方法时折返，如果出现了溢出，就将溢出的符号位丢了。关于溢出，默认是折返
　　还有其他策略，饱和截断超过了范围就停留在最大值最小值那里，失败报错不执行。

ps：读掘金小册笔记

简单的异步消息队列

　　对于一些只有一组消费者的消息队列，使用redis可以轻松解决，redis的list结构可以很好的处理，这个list的基本操作rpush，lpush,rpop,lpop即可处理队列。

队列为空怎么破

　　客户端通过pop操作来获取消息，从而消费，如果队列空了，客户端就会进入pop的死循环，由于没数据，不停的pop，这种空循环不但对客户端影响很大，对redis的qps也有影响，我们则是通过sleep来解决这个问题，让线程睡一下即可，也就是暂停一秒左右。

队列延迟

　　这个让线程休眠会导致消息延迟变高，有个办法可以避免这个问题，不用让线程睡眠，那就是用blpop/brpop阻塞读，阻塞读在队列没有数据的时候，会进入到休眠状态，一旦数据到来，就会立刻醒过来。

空闲连接自动断开

　　如果这个线程一直阻塞在哪里，redis的客户端连接就会变成限制连接，服务器会主动断开，减少资源占用，这个时候blpop/brpop就会抛出异常。，因此编写客户端的时候要注意捕捉异常。

锁冲突处理

　　请求加锁失败时：
　　１．直接抛出异常。２．sleep一会再试。3.将请求移动到延时队列，过一会再试试。实现如下：
　　通过zset来实现延时队列，我们将消息序列化为value,到期时间作为score，然后多个线程轮训获取到期任务处理。