深入浅出boot2.0 第七章 redis

古城微笑少年丶 2023-02-15 00:23 42阅读 0赞

## 最新mybatis配置 ##

<dependency>
                <groupId>org.mybatis.spring.boot</groupId>
                <artifactId>mybatis-spring-boot-starter</artifactId>
                <version>2.1.2</version>
            </dependency>
            <dependency>
                <groupId>mysql</groupId>
                <artifactId>mysql-connector-java</artifactId>
                <scope>runtime</scope>
            </dependency>
    
    	如：
        <version>6.0.6</version>

#数据库配置
    #spring.datasource.url=jdbc:mysql://localhost:3306/test
    
    spring.datasource.url=jdbc:mysql://localhost:3306/test?serverTimezone=UTC&useUnicode=true&characterEncoding=utf8&useSSL=false
    #serverTimezone 必须要配置
    
    #可以不配置数据库驱动，Spring Boot会自己发现
    #spring.datasource.driver-class-name=com.mysql.cj.jdbc.Driver
    
    com.mysql.jdbc.Driver 是 mysql-connector-java 5中的，
    com.mysql.cj.jdbc.Driver 是 mysql-connector-java 6中的
    
    在设定时区的时候，如果设定serverTimezone=UTC，会比中国时间早8个小时，如果在中国，可以选择Asia/Shanghai或者Asia/Hongkong，
    serverTimezone=Asia/Shanghai

## redis介绍和xml (spring-data-redis) ##

*  加速系统，redis 和 mongoDB
 *  redis 是运行在内存中的数据库。支持7中数据类型
 *  开源，使用 ansic语言编写，遵循 BSD协议，基于内存可持久化的日志型。
 *  运行速度 是关系数据库的 几倍 到 几十倍。
 *  redis 1秒内 完成 10万次的 读写。
 *  将常用的数据类型存储在redis中，网站性能大幅提高
 *  正常网站的查询 和 更新 是 1:9 到 3:7
 *  当一个会员登录网站，可以把常用数据 一次性查询放入redis
 *  redis提供简单的事务，lua语言增加运算能力（原子性）
 *  spring 通过 spring-data-redis 对redis进行支持

<dependency>
    			<groupId>org.springframework.boot</groupId>
    			<artifactId>spring-boot-starter-data-redis</artifactId>
    			<exclusions>
    				
    				<exclusion>
    					<groupId>io.lettuce</groupId>
    					<artifactId>lettuce-core</artifactId>
    				</exclusion>
    			</exclusions>
    		</dependency>
    		
    		<dependency>
    			<groupId>redis.clients</groupId>
    			<artifactId>jedis</artifactId>
    		</dependency>

*  redis是 键值数据库，以字符串为中心
 *  包括：字符串，散列，列表（链表），集合，有序集合，基数 和 地理位置

### Spring-data-redis ###

*  redis连接驱动有 Jedis ，其他还有 Lettuce （少）， Jredis 和 Srp 不被推荐使用
 *  提供 RedisConnectionFactory ，生成 RedisConnection接口对象(Redis底层封装)
 *  Spring提供实现类 JedisConnection 去封装原有 jedis
 *  第一步肯定要配置这个工厂了，先配置redis连接池 （最大连接数，超时时间）
 *  RedisConnectionFactory —— JedisConnectionFactory —— 连接工厂管理redis连接池
 *  通过工厂获取到 RedisConnection
    
     *  AbstractRedisConnection——JedisConnection 封装jedis操作
     *  RedisClusterConnection——JedisClusterConection 集群支持
     *  StringRedisConnection——DefaultStringRedisConnection 字符串支持

### 创建redisConnectionFactory ###

//@Configuration
    public class RedisConfig { 
    
    	private RedisConnectionFactory connectionFactory = null;
    	
    	@Bean(name = "redisConnectionFactory")
    	public RedisConnectionFactory initConnectionFactory() { 
    		if (this.connectionFactory != null) { 
    			return this.connectionFactory;
    		}
    		JedisPoolConfig poolConfig = new JedisPoolConfig();
    		// 最大空闲数
    		poolConfig.setMaxIdle(50);
    		// 最大连接数
    		poolConfig.setMaxTotal(100);
    		// 最大等待毫秒数
    		poolConfig.setMaxWaitMillis(2000);
    		// 创建Jedis连接工厂
    		JedisConnectionFactory connectionFactory = new JedisConnectionFactory(poolConfig);
    		// 配置Redis连接服务器
    		RedisStandaloneConfiguration rsc = connectionFactory.getStandaloneConfiguration();
    		rsc.setHostName("192.168.10.128");
    		rsc.setPort(6379);
    		rsc.setPassword(RedisPassword.of("123456"));
    		this.connectionFactory = connectionFactory;
    		return connectionFactory;
    	}
    	
    	@Bean(name="redisTemplate")
    	public RedisTemplate<Object, Object> initRedisTemplate() { 
    	    RedisTemplate<Object, Object> redisTemplate = new RedisTemplate<>();
    	    redisTemplate.setConnectionFactory(initConnectionFactory());
    	    
    	    RedisSerializer<String> stringRedisSerializer = redisTemplate.getStringSerializer();
    	    redisTemplate.setKeySerializer(stringRedisSerializer);//key 序列化
    	    //setValueSerializer 也设置成string多好
    	    redisTemplate.setHashKeySerializer(stringRedisSerializer);//hash key序列化
    	    redisTemplate.setHashValueSerializer(stringRedisSerializer);//hash Value序列化
    	    
    	  return redisTemplate;
    	}
    }

*  redisTemplate 会自动从工程获取连接，关闭连接

### 测试redisTemplate ###

> keys k\* 命令行查看存储的key
> 
> 散列结构：hget hash field //hget map放入的键名字 hashMap的键 得到的是hvalue

public static void main(String[] args) { 
            //SpringApplication.run(DemoStartSevenApplication.class, args);
            AnnotationConfigApplicationContext context=new AnnotationConfigApplicationContext(RedisConfig.class);
            RedisTemplate bean = context.getBean(RedisTemplate.class);
            bean.opsForValue().set("key1","我的测试ke1");
            bean.opsForHash().put("hash","field","hvalue");
        }

### 自定义序列化 ###

*  redis是一种基于字符串存储的NoSQL，而java是基于对象的，
 *  java提供了序列化机制。只要实现 java.io.serializable接口 即可。将对象进行序列化 得到二进制字符串
 *  序列化器设置 RedisSerializer
    
     *  GenericToString Serializer
     *  Jackson2Json RedisSerializer
     *  GenericJackson2 RedisSerializer
     *  Jdk Serialization RedisSerializer (默认)
     *  OxmSerializer
     *  String RedisSerializer （常用）
     *  JacksonJson RedisSerializer (@Deprecated)
 *  RedisSerializer 有 bye\[\] serialize(T t) 和 deserialize (byte\[\] b)
 *  实现Serializable接口对象 ——redis序列化容器——二进制（存到redis）

RedisSerializer<String> stringRedisSerializer = redisTemplate.getStringSerializer();
            redisTemplate.setKeySerializer(stringRedisSerializer);//key 序列化
            redisTemplate.setValueSerializer(stringRedisSerializer);//value 序列化
            redisTemplate.setHashKeySerializer(stringRedisSerializer);//hash key序列化
            redisTemplate.setHashValueSerializer(stringRedisSerializer);//hash Value序列化
    
    		redisTemplate.setDefaultSerializer(stringRedisSerializer);
    	    redisTemplate.setStringSerializer(stringRedisSerializer);//字符串序列化，自动赋值为这个序列化器

bean.opsForValue().set("key1","我的测试ke1");
            bean.opsForHash().put("hash","field","hvalue");
            //是两条连接的操作， 为了克服这个问题spring提供了 RedisCallback和SessionCallback
    		//string操作最多，所以提供了 StringRedisTemplate 这个类继承 RedisTemplate

### spring 对redis数据类型的操作 ###

redis支持7种数据类型： redisTemplate

*  字符串 Value Operations rt.opsForValue().set(“key1”,“我的测试ke1”);
 *  散列 Hash Operations rt.opsForHash().put(“hash”,“field”,“hvalue”);
 *  列表（链表）List Operatios rt.opsForList();
 *  集合 Set Operatios bean.opsForSet();
 *  有序集合 ZSet Operations bean.opsForZSet();
 *  基数 HyperLog Operations bean.opsForHyperLogLog();
 *  地理位置 Geo Operations 地理位置 bean.opsForGeo();
 *  有时候需要对某一个键值对 做连续的操作，连续操作一个散列数据类型 或者 列表多次
 *  我们就可以对某个键进行多次操作
    
     *  BoundXXX operations
        
         *  Bound Value Operations rt.boundValueOps(“string”); //string是key
         *  Bound Hash Operations XXXXX
         *  Bound List Operations
         *  Bound Set Operations
         *  Bound ZSet Operations
         *  Bound HyperLog Operations
         *  Bound Geo Operations rt.boundGeoOps(“geo”);
            
                BoundHashOperations hashOps = stringRedisTemplate.boundHashOps("hash");
                        // 删除两个个字段
                        hashOps.delete("field1", "field2");
                        // 新增一个字段
                        hashOps.put("filed4", "value5");

### SessionCallback 和 RedisCallback接口 ###

*  作用是让redis回调 (使用同一条Redis连接进行回调，在同一条redis执行多个方法)
 *  同一条连接下执行多个Redis命令
 *  SessionCallback 提供了良好的封装（使用它）
 *  RedisCallback 接口比较底层

// 高级接口，比较友好，一般情况下，优先使用它
    	public static void useSessionCallback(RedisTemplate redisTemplate) { 
    	    redisTemplate.execute(new SessionCallback() { 
    	        @Override
    	        public Object execute(RedisOperations ro) 
    	                throws DataAccessException { 
    	            ro.opsForValue().set("key1", "value1");
    	            ro.opsForHash().put("hash", "field", "hvalue");
    	            return null;
    	        }
    	    });
    	}

// 需要处理底层的转换规则，如果不考虑改写底层，尽量不使用它
    public static void useRedisCallback(RedisTemplate redisTemplate) { 
        redisTemplate.execute(new RedisCallback() { 
            @Override
            public Object doInRedis(RedisConnection rc)
                    throws DataAccessException { 
                rc.set("key1".getBytes(), "value1".getBytes());
                rc.hSet("hash".getBytes(), "field".getBytes(), "hvalue".getBytes());
                return null;
            }
        });
    }

*  RedisCallback接口并不是那么友好，能够改一些底层的东西。比如序列化
 *  lambda

//报错，但可用
    	public static void useSessionCallback(RedisTemplate redisTemplate) { 
    	    redisTemplate.execute((RedisOperations ro) -> { 
    	        ro.opsForValue().set("key1", "value1");
    	        ro.opsForHash().put("hash", "field", "hvalue");
    	        return null;
    	    });
    	}
    
    
    	public static void useRedisCallback(RedisTemplate redisTemplate) { 
    	    redisTemplate.execute((RedisConnection rc) -> { 
    	        rc.set("key1".getBytes(), "value1".getBytes());
    	        rc.hSet("hash".getBytes(), "field".getBytes(), "hvalue".getBytes());
    	        return null;
    	    });
    	}

## spring boot 配置和使用 Redis ##

#数据库配置
    spring.datasource.url=jdbc:mysql://localhost:3306/spring_boot_chapter7
    spring.datasource.username=root
    spring.datasource.password=123456
    #可以不配置数据库驱动，Spring Boot会自己发现
    #spring.datasource.driver-class-name=com.mysql.jdbc.Driver
    
    
    #=========================数据库相关配置完毕===================================
    
    spring.datasource.tomcat.max-idle=10
    spring.datasource.tomcat.max-active=50
    spring.datasource.tomcat.max-wait=10000
    spring.datasource.tomcat.initial-size=5
    #设置默认的隔离级别为读写提交
    spring.datasource.tomcat.default-transaction-isolation=2
    
    #=========================数据源相关配置完毕===================================
    
    #mybatis配置
    mybatis.mapper-locations=classpath:com/springboot/chapter7/mapper/*.xml
    mybatis.type-aliases-package=com.springboot.chapter7.pojo
    
    #日志配置为DEBUG级别，这样日志最为详细
    logging.level.root=DEBUG  #全局日志为debug
    logging.level.org.springframework=DEBUG
    logging.level.org.org.mybatis=DEBUG
    
    #Redis配置
    spring.redis.jedis.pool.min-idle=5  #连接池属性
    spring.redis.jedis.pool.max-active=10
    spring.redis.jedis.pool.max-idle=10
    spring.redis.jedis.pool.max-wait=2000
    
    spring.redis.port=6379 #服务器属性
    spring.redis.host=192.168.11.128
    spring.redis.password=123456
    
    #redis连接超时时间，单位毫秒
    spring.redis.timeout=1000
    
    #缓存配置
    spring.cache.type=REDIS
    spring.cache.cache-names=redisCache
    
    spring.cache.redis.use-key-prefix=false
    spring.cache.redis.key-prefix=
    
    spring.cache.redis.cache-null-values=true
    
    spring.cache.redis.time-to-live=600000

*  会自动生成 RedisConnectioFactory，RedisTemplate，StringRedisTemplate (只使用字符串，不支持java对象)
 *  通过RedisTemplate 序列化器 来处理
    
        main的启动类上都行：
        	@Autowired
        	private RedisTemplate redisTemplate = null;
        
        	// 设置RedisTemplate的序列化器
        	private void initRedisTemplate() { 
        		RedisSerializer stringSerializer = redisTemplate.getStringSerializer();
        		redisTemplate.setKeySerializer(stringSerializer);
        		redisTemplate.setHashKeySerializer(stringSerializer);
        	}
        
        	// 自定义初始化方法
        	@PostConstruct
        	public void init() { 
        	    initRedisTemplate();
        	}

### 操作redis数据类型(字符和散列) ###

*  字符串
 *  散列
 *  列表
 *  集合
 *  有序列表
 *  大部分只需要一次操作所以使用RedisTemplate
 *  需要多次执行Redis命令，可以用 SessionCallback
    
        @RequestMapping("/stringAndHash")
            @ResponseBody
            public Map<String, Object> testStringAndHash() { 
                //设置了一个 key1
                redisTemplate.opsForValue().set("key1", "value1");
                // 注意这里使用了JDK的序列化器，所以Redis保存的时候不是整数，不能运算
                //设置了一个1
                redisTemplate.opsForValue().set("int_key", "1");
                //设置了一个1
                stringRedisTemplate.opsForValue().set("int", "1");
                // 使用运算。+1
                stringRedisTemplate.opsForValue().increment("int", 1);
                // 获取底层Jedis连接
                Jedis jedis = (Jedis) stringRedisTemplate.getConnectionFactory().getConnection().getNativeConnection();
                // 减一 操作,这个命令RedisTemplate不支持，所以笔者先获取底层的连接再操作
                jedis.decr("int");
                
                Map<String, String> hash = new HashMap<String, String>();
                hash.put("field1", "value1");//定义散列的第一个值
                hash.put("field2", "value2");//第二个值
                // 存入一个散列数据类型
                stringRedisTemplate.opsForHash().putAll("hash", hash);
                
                // 新增一个字段
                stringRedisTemplate.opsForHash().put("hash", "field3", "value3");
                
                // 绑定散列操作的key，这样可以连续对同一个散列数据类型进行操作
                BoundHashOperations hashOps = stringRedisTemplate.boundHashOps("hash");
                // 删除两个个字段
                hashOps.delete("field1", "field2");
                
                // 新增一个字段
                hashOps.put("filed4", "value5");
                
                Map<String, Object> map = new HashMap<String, Object>();
                map.put("success", true);
                
                return map;
            }

### 使用string 操作链表 ###

*  列表是一种链表结构
 *  查询性能不高，而增删节点的性能高。
 *  存在从左到右 或 从右到左的操作

@RequestMapping("/list")
        @ResponseBody
        public Map<String, Object> testList() { 
            // 插入两个列表,注意它们再链表的顺序
            // 链表从左到右顺序为v10,v8,v6,v4,v2
            stringRedisTemplate.opsForList().leftPushAll("list1", "v2", "v4", "v6", "v8", "v10");
            // 链表从左到右顺序为v1,v2,v3,v4,v5,v6
            stringRedisTemplate.opsForList().rightPushAll("list2", "v1", "v2", "v3", "v4", "v5", "v6");
            // 绑定list2链表操作
            BoundListOperations listOps = stringRedisTemplate.boundListOps("list2");
            // 从右边弹出一个成员 v6
            Object result1 = listOps.rightPop();
            // 获取定位元素，Redis从0开始计算,这里值为v2
            Object result2 = listOps.index(1);
            // 从左边插入链表。插入第一个
            listOps.leftPush("v0");
            // 求链表长度
            Long size = listOps.size();
            // 求链表下标区间成员，整个链表下标范围为0到size-1，这里不取最后一个元素
            List elements = listOps.range(0, size - 2);
            Map<String, Object> map = new HashMap<String, Object>();
            map.put("success", true);
            return map;
        }

### 使用String操作集合 ###

*  redis中是不允许成员重复的
 *  是一个散列的结构，而且是无序的。
 *  redis还提供了 交集，并集，差集
    
        @RequestMapping("/set")
        	@ResponseBody
        	public Map<String, Object> testSet() { 
        		// 请注意：这里v1重复2次，由于集合不允许重复，所以只是插入5个成员到集合中
        		stringRedisTemplate.opsForSet().add("set1", "v1", "v1", "v2", "v3", "v4", "v5");
        		stringRedisTemplate.opsForSet().add("set2", "v2", "v4", "v6", "v8");
        		// 绑定set1集合操作
        		BoundSetOperations setOps = stringRedisTemplate.boundSetOps("set1");
        		// 增加两个元素
        		setOps.add("v6", "v7");
        		// 删除两个元素
        		setOps.remove("v1", "v7");
        		// 返回所有元素
        		Set set1 = setOps.members();
        		// 求成员数
        		Long size = setOps.size();
        		// 求交集
        		Set inter = setOps.intersect("set2");
        		// 求交集，并且用新集合inter保存
        		setOps.intersectAndStore("set2", "inter");
        		// 求差集
        		Set diff = setOps.diff("set2");
        		// 求差集，并且用新集合diff保存
        		setOps.diffAndStore("set2", "diff");
        		// 求并集
        		Set union = setOps.union("set2");
        		// 求并集，并且用新集合union保存
        		setOps.unionAndStore("set2", "union");
        		Map<String, Object> map = new HashMap<String, Object>();
        		map.put("success", true);
        		return map;
        	}

### 使用string操作有序集合 ###

*  热门商品，或 最大的购买买家。 刷新需要及时，用数据库太慢
 *  有序集合也是一种散列表存储的方式
 *  有序性 只是靠它在数据结构中增加一个属性 score分数 得以支持
 *  Spring提供了 TypedTuple 接口，有两个方法。还提供了实现类 DefaultTypedTuple
 *  TypedTuple——getValue getScore
    
     *  DefaultTypedTuple —— score value
     *  value 保存有序集合的值 （买家）
     *  score 保存分数 （买家话费的金钱）
        
            Tuple
            英 /tjʊpəl; ˈtʌpəl/  美 /ˈtjʊpəl; ˈtʌpəl/  全球(美国)  
            n. [计] 元组，重数
    
        @RequestMapping("/zset")
            @ResponseBody
            public Map<String, Object> testZset() { 
                Set<ZSetOperations.TypedTuple<String>> typedTupleSet = new HashSet<>();
                for (int i = 1; i <= 9; i++) { 
                    // 分数
                    double score = i * 0.1;
                    // 创建一个TypedTuple对象，存入值和分数
                    ZSetOperations.TypedTuple<String> typedTuple = new DefaultTypedTuple<String>("value" + i, score);
                    typedTupleSet.add(typedTuple);
                }
                // 往有序集合插入元素
                stringRedisTemplate.opsForZSet().add("zset1", typedTupleSet);
                // 绑定zset1有序集合操作
                BoundZSetOperations<String, String> zsetOps = stringRedisTemplate.boundZSetOps("zset1");
                // 增加一个元素
                zsetOps.add("value10", 0.26);
                Set<String> setRange = zsetOps.range(1, 6);
                // 按分数排序获取有序集合
                Set<String> setScore = zsetOps.rangeByScore(0.2, 0.6);
                // 定义值范围
                RedisZSetCommands.Range range = new RedisZSetCommands.Range();
                range.gt("value3");// 大于value3
                // range.gte("value3");// 大于等于value3
                // range.lt("value8");// 小于value8
                range.lte("value8");// 小于等于value8
                // 按值排序，请注意这个排序是按字符串排序
                Set<String> setLex = zsetOps.rangeByLex(range);
                // 删除元素
                zsetOps.remove("value9", "value2");
                // 求分数
                Double score = zsetOps.score("value8");
                // 在下标区间下，按分数排序，同时返回value和score
                Set<ZSetOperations.TypedTuple<String>> rangeSet = zsetOps.rangeWithScores(1, 6);
                // 在分数区间下，按分数排序，同时返回value和score
                Set<ZSetOperations.TypedTuple<String>> scoreSet = zsetOps.rangeByScoreWithScores(1, 6);
                // 按从大到小排序
                Set<String> reverseSet = zsetOps.reverseRange(2, 8);
                Map<String, Object> map = new HashMap<String, Object>();
                map.put("success", true);
                return map;
            }
 *  TypeTuple保存集合的元素，默认从小到大排序

## redis的一些特殊用法 ##

*  事务
 *  流水线：大批量执行Redis命令的时候
 *  发布订阅
 *  Lua语 ：利用 Redis执行 Lua的原子性 来达到 数据一致性的目的

### 使用Redis事务 ###

*  命令组合 watch… multi … exec
 *  在一个 Redis 连接中执行 多个命令，考虑使用 SessionCallback接口
 *  watch命令是 可以监控 Redis的一些键，multi命令是开始事务
    
     *  开始事务后，客户端命令不会马上执行，而是放到一个队列里
     *  执行一些返回数据的命令，结果返回null
 *  exe命令的意义在于执行事务
    
     *  在队列命令执行前会判断 被 watch监控 的 redis的键 的数据是否发生过变化
     *  （重新赋值 都算变化过）
     *  如果变化过，那就取消事务。否则就执行事务。要么全部执行，要么全部不执行。
     *  且不会被其他客户端打断。
 *  watch命令 —— multi 命令开启事务 —— 命令进入队列 ——exec命令执行事务
 *  监控键值对 是否发生变化 （会判断watch命令的键值对是否发生变化）
    
     *  是 ，取消事务，取消监控的键值对
     *  否，执行事务，取消监控的键值对
 *  需要保证 RedisTemplate的键和散列结构 的 field 使用 字符串 序列化（STringRedisSerializer)
    
        @RequestMapping("/multi")
            @ResponseBody
            public Map<String, Object> testMulti() { 
                redisTemplate.opsForValue().set("key1", "value1");
                List list = (List) redisTemplate.execute((RedisOperations operations) -> { 
                    // 设置要监控key1
                    operations.watch("key1");
                    // 开启事务，在exec命令执行前，全部都只是进入队列
                    operations.multi();
                    operations.opsForValue().set("key2", "value2");
                    //operations.opsForValue().increment("key1", 1);// ①
                    // 获取值将为null，因为redis只是把命令放入队列，
                    Object value2 = operations.opsForValue().get("key2");
                    System.out.println("命令在队列，所以value为null【" + value2 + "】");
                    operations.opsForValue().set("key3", "value3");
                    Object value3 = operations.opsForValue().get("key3");
                    System.out.println("命令在队列，所以value为null【" + value3 + "】");
                    // 执行exec命令，将先判别key1是否在监控后被修改过，如果是不执行事务，否则执行事务
                    return operations.exec();// ②
                });
                System.out.println(list);
                Map<String, Object> map = new HashMap<String, Object>();
                map.put("success", true);
                return map;
            }
        
         			// 设置要监控key1
                    operations.watch("key1");
                    // 开启事务，在exec命令执行前，全部都只是进入队列
                    operations.multi();
        			// 执行exec命令
                    return operations.exec();// ②
 *  在 @2 处打断点，并且手动修改redis中的key1值，此时跳过断点，所有的不会被执行（在最后一步，会检查key1，如果和最开始不一致，本次不执行）
    
     *  因为 先使用 watch 监控了 key1。operations.watch(“key1”);
     *  之后的 multi 让之后的命令进入队列。
     *  在exec 方法运行前，我们手动改了 key1 。事务规则，没有修改过 才会执行事务。
 *  把 @1 处标记放开。 此处会报错，没法对 string类型 加1 ，但最终 key2 key3还是有值的。
    
     *  这就是和 数据库事务不一样
     *  redis事务 先让命令进入 队列，并没有检测 +1 命令是否执行成功
     *  只有 在 exec命令执行的时候 才会发现错误
     *  错误后面的依旧会执行
     *  为了解决这个问题，在执行事务前，都会严格的检查，避免这样的情况

### Redis流水线 ###

*  默认情况下，Redis客户端是一条条发送给 Redis服务器的。（性能不高）
 *  关系数据库可以批量，执行SQL，一次发送所有SQL
 *  Redis也可以，叫做流水线。pipline技术
 *  网络传输造成了瓶颈，使用了流水线大幅度 提升Redis性能
 *

Pipeline  
英 /ˈpaɪplaɪn/ 美 /ˈpaɪplaɪn/ 全球(英国)  
n. 管道；输油管；传递途径

- 
    
    ```java
    @RequestMapping("/pipeline")
    	@ResponseBody
    	public Map<String, Object> testPipeline() {
    		Long start = System.currentTimeMillis();
    		List list = (List) redisTemplate.executePipelined((RedisOperations operations) -> {
    			for (int i = 1; i <= 100000; i++) {
    				operations.opsForValue().set("pipeline_" + i, "value_" + i);
    				String value = (String) operations.opsForValue().get("pipeline_" + i);
    				if (i == 100000) {
    					System.out.println("命令只是进入队列，所以值为空【" + value + "】");
    				}
    			}
    			return null;
    		});
    		Long end = System.currentTimeMillis();
    		System.out.println("耗时：" + (end - start) + "毫秒。");
    		Map<String, Object> map = new HashMap<String, Object>();
    		map.put("success", true);
    		return map;
    	}
    
    10万条：耗时：417毫秒。
    1万条：90毫秒
     redisTemplate.executePipelined
    不使用流水线，每秒3万条。
    流水线大约能提升10倍的性能

*  这里还是用的 SessionCallback 接口

### 使用Redis发布订阅 ###

*  在企业分配任务后，可以通过邮件，短信或者微信，通知到相关责任人。
 *  Redis提供一个通道，让消息能发送到这个通道上。多个系统可以监听这个渠道（短信，微信，邮件）
 *  当一条消息发送到 渠道，渠道就会通知它的监听者。
 *  消息 发送到 渠道 ——短信，微信，邮件。 监听

#### 定义消息监听器 MessageListener ####

@Component
    public class RedisMessageListener implements MessageListener { 
        @Override //pattern渠道名称
        public void onMessage(Message message, byte[] pattern) { 
            // 消息体
            String body = new String(message.getBody());
            // 渠道名称
            String topic = new String(pattern); 
            System.out.println(body);
            System.out.println(topic);
        }
    }

#### 配置：需要工厂，定时任务，通道，消息监听器 ####

@SpringBootApplication(scanBasePackages = "com.springboot.chapter7")
    public class Chapter7Application { 
    
    	// RedisTemplate
    	@Autowired
    	private RedisTemplate redisTemplate = null;
    	
    	//redis 连接工厂
    	@Autowired
    	private RedisConnectionFactory connectionFactory = null;
    
    
    	// Redis消息监听器
    	@Autowired
    	private MessageListener redisMsgListener = null;
    	
    	// 任务池
    	private ThreadPoolTaskScheduler taskScheduler = null;
    	
    	/** * 创建任务池，运行线程等待处理Redis的消息 * @return */
    	@Bean
    	public ThreadPoolTaskScheduler initTaskScheduler() { 
    		if (taskScheduler != null) { 
    			return taskScheduler;
    		}
    	    //创建任务池
    		taskScheduler = new ThreadPoolTaskScheduler();
    	    //设置任务池的大小
    		taskScheduler.setPoolSize(20);
    		return taskScheduler;
    	}
    	
    	/** * 定义Redis的监听容器 * @return 监听容器 */
    	@Bean
    	public RedisMessageListenerContainer initRedisContainer() { 
    		RedisMessageListenerContainer container = new RedisMessageListenerContainer();
    		// 1，Redis连接工厂
    		container.setConnectionFactory(connectionFactory);
    		// 2，设置运行任务池。
    		container.setTaskExecutor(initTaskScheduler());
    		// 3，定义监听渠道，名称为topic1
    		Topic topic = new ChannelTopic("topic1");
    		// 4，使用监听器监听Redis的消息。（上面自定义的）
    		container.addMessageListener(redisMsgListener, topic);
    		return container;
    	}
    	
    	public static void main(String[] args) { 
    		SpringApplication.run(Chapter7Application.class, args);
    	}
    
    }

*  定义了一个任务池，设置了任务池的大小（20个）
 *  它将可以运行线程，等待redis消息的传入
 *  定义了接收 topic1 渠道的消息

#### 测试 ####

@Resource
        private RedisTemplate redisTemplate;
    
        @GetMapping("/myTest")
        public String myTest(){ 
            redisTemplate.convertAndSend("topic1","我的对象，哈哈哈，测试监听者");
            return "hahaha";
        }
    
    或者：publish topic1 哈哈哈
    都会进入：自定义消息监听器里面

## 使用Lua脚本 ##

*  Redis在 2.6 版本号 提供了 Lua脚本的支持
 *  执行Lua脚本具备 原子性。保证数据的一致性
 *  更强大的运算能力，比redis自身的事务，要好一些。
 *  有两种运行的方法
    
     *  直接发送Lua到服务器去执行
     *  先把Lua发给redis，Redis对 Lua缓存，返回一个 SHA1的 32位编码回来，
        
         *  之后只需发给SHA1和相关参数给Redis便可以执行了
 *  为什么要通过32位编码执行
    
     *  如果Lua脚本长，需要网络传递，传递的速度跟不上redis的执行速度
     *  如果只传递32位编码和参数，需要传递的消息就少了许多，可以极大的减少网络传输的内容。
 *  Spring提供了，RedisScript接口，还有实现类：DefaultRedisScript
    
        public interface RedisScript<T> { 
            String getSha1();//获取脚本的 sha1
            
            @Nullable
            Class<T> getResultType();//获取返回值。返回的java的类型
            
            String getScriptAsString();//获取脚本的字符串
        }
    
        @RequestMapping("/lua")
        	@ResponseBody
        	public Map<String, Object> testLua() { 
        		DefaultRedisScript<String> rs = new DefaultRedisScript<String>();
        		// 设置脚本
        		rs.setScriptText("return 'Hello Redis'");
        		// 定义返回类型，注意如果没有这个定义Spring不会返回结果
        		rs.setResultType(String.class);
        		RedisSerializer<String> stringSerializer = redisTemplate.getStringSerializer();
        		// 执行Lua脚本
        		String str = (String) redisTemplate.execute(rs, stringSerializer, stringSerializer, null);
        		Map<String, Object> map = new HashMap<String, Object>();
        		map.put("str", str);
        		return map;
        	}

## Spring缓存注解操作Redis ##

*  简化Redis使用，提供了缓存注解
 *  CacheManager
    
     *  EhCache CacheManager
     *  JCache CacheManager
     *  Redis CacheManager
     *  SimpleCacheManager
 *  需要配置缓存管理器，（缓存类型，超时时间）
 *  RedisCacheManager 最常用

### 缓存管理器配置 ###

#日志配置为DEBUG级别，这样日志最为详细
    logging.level.root=DEBUG
    logging.level.org.springframework=DEBUG
    logging.level.org.org.mybatis=DEBUG
    
    #Redis配置
    spring.redis.jedis.pool.min-idle=5
    spring.redis.jedis.pool.max-active=10
    spring.redis.jedis.pool.max-idle=10
    spring.redis.jedis.pool.max-wait=2000
    spring.redis.port=6379
    spring.redis.host=192.168.11.128
    spring.redis.password=123456
    
    #redis连接超时时间，单位毫秒
    spring.redis.timeout=1000
    
    
    
    spring.cache.caffeine.spec= #caffeine缓存配置细节
    spring.cache.couchbase.expiration=0ms
    spring.cache.ehcache.config= *
    spring.cache.infinispan.config= *
    spring.cache.jcache.config= #
    spring.cache.jcache.provider= *
    
    #是否允许redis缓存空值
    spring.cache.redis.cache-null-values=true 
    #是否启用键前缀
    spring.cache.redis.use-key-prefix=false
    #redis的键前缀
    spring.cache.redis.key-prefix=
    #缓存超时时间戳，配置为0，则不设置超时时间
    spring.cache.redis.time-to-live=600000
    
    #缓存配置。会自动推测
    spring.cache.type=REDIS
    #配置缓存的名称
    spring.cache.cache-names=redisCache

### 开启缓存 ###

@EnableCaching

### 配置mybatis ###

#### pom ####

<dependency>
    			<groupId>org.mybatis.spring.boot</groupId>
    			<artifactId>mybatis-spring-boot-starter</artifactId>
    			
    		</dependency>
    		<dependency>
    			<groupId>mysql</groupId>
    			<artifactId>mysql-connector-java</artifactId>
    			<scope>runtime</scope>
    		</dependency>

#### properties ####

#### pojo ####

@Alias("user")
    public class User implements Serializable { 
    	private static final long serialVersionUID = 7760614561073458247L;
    	private Long id;
    	private String userName;
    	private String note;
    }

#### mapper接口 ####

@Repository
    public interface UserDao { 
        // 获取单个用户
        User getUser(Long id);
        
        // 保存用户
        int insertUser(User user);
        
        // 修改用户
        int updateUser(User user);
        
        // 查询用户，指定MyBatis的参数名称
        List<User> findUsers(@Param("userName") String userName,
                             @Param("note") String note);
        
        // 删除用户
        int deleteUser(Long id); 
    }

#### 主类的包扫描 ####

//指定扫描的MyBatis Mapper
    @MapperScan(basePackages = "com.springboot.chapter7", annotationClass = Repository.class)

#### userMapper.xml ####

<?xml version="1.0" encoding="UTF-8" ?>
    <!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
    <mapper namespace="com.springboot.demostartseven.dao.UserDao">
    
    	<select id="getUser" parameterType="long" resultType="user">
    		select id, user_name as userName, note from t_user
    		where id = #{id}
    	</select>
    
    	<insert id="insertUser" useGeneratedKeys="true" keyProperty="id" parameterType="user">
    		insert into t_user(user_name, note)
    		values(#{userName}, #{note})
    	</insert>
    
    	<update id="updateUser">
    		update t_user
    		<set>
    			<if test="userName != null">user_name =#{userName},</if>
    			<if test="note != null">note =#{note}</if>
    		</set>
    		where id = #{id}
    	</update>
    
    	<select id="findUsers" resultType="user">
    		select id, user_name as userName, note from t_user
    		<where>
    			<if test="userName != null">
    				and user_name = #{userName}
    			</if>
    			<if test="note != null">
    				and note = #{note}
    			</if>
    		</where>
    	</select>
    
    	<delete id="deleteUser" parameterType="long">
    		delete from t_user where id = #{id}
    	</delete>
    </mapper>

*  useGeneratedKeys=“true” keyProperty=“id”
 *  MyBatis机会将数据库生成的主键 回填到POJO id中

#### mybatis配置思路 ####

1.pom  mybatis-starter 和 mysql驱动
    2.properties  连接配置， mapper.xml位置，pojo位置
    3.主类的main  配置mapper接口位置 @MapperScan(basePackages = "com.XX", annotationClass = Repository.class)
    4.dao接口 @Repository
    5.sql的xml配置

### service层使用缓存 ###

public interface UserService { 
        // 获取单个用户
        User getUser(Long id);
        
        // 保存用户
        User insertUser(User user);
        
        // 修改用户，指定MyBatis的参数名称
        User updateUserName(Long id, String userName);
        
        // 查询用户，指定MyBatis的参数名称
        List<User> findUsers(String userName, String note);    
        
        // 删除用户
        int deleteUser(Long id);  
    }

@Service
    public class UserServiceImpl implements UserService { 
    
    	@Autowired
    	private UserDao userDao = null;
    
    	// 插入用户，最后MyBatis会回填id，取结果id缓存用户
    	@Override
    	@Transactional
        //把结果放入redis注解
    	@CachePut(value = "redisCache", key = "'redis_user_'+#result.id")
    	public User insertUser(User user) { 
    		userDao.insertUser(user);
    		return user;
    	}
    
    	// 获取id，取参数id缓存用户
    	@Override
    	@Transactional
    	@Cacheable(value = "redisCache", key = "'redis_user_'+#id")
    	public User getUser(Long id) { 
    		return userDao.getUser(id);
    	}
    
    	// 更新数据后，充值缓存，使用condition配置项使得结果返回为null，不缓存
    	@Override
    	@Transactional
    	@CachePut(value = "redisCache", condition = "#result != 'null'", key = "'redis_user_'+#id")
    	public User updateUserName(Long id, String userName) { 
    		// 此处调用getUser方法，该方法缓存注解失效，
    		// 所以这里还会执行SQL，将查询到数据库最新数据
    		User user = this.getUser(id);
    		if (user == null) { 
    			return null;
    		}
    		user.setUserName(userName);
    		userDao.updateUser(user);
    		return user;
    
    	}
    
    	// 命中率低，所以不采用缓存机制
    	@Override
    	@Transactional
    	public List<User> findUsers(String userName, String note) { 
    		return userDao.findUsers(userName, note);
    	}
    
    	// 移除缓存
    	@Override
    	@Transactional
    	@CacheEvict(value = "redisCache", key = "'redis_user_'+#id", beforeInvocation = false)
    	public int deleteUser(Long id) { 
    		return userDao.deleteUser(id);
    	}
    }

@CachePut(value = "redisCache", key = "'redis_user_'+#result.id") //放入redis_user_11
    	
    	@Cacheable(value = "redisCache", key = "'redis_user_'+#id")
    	
    	@CachePut(value = "redisCache", condition = "#result != 'null'", key = "'redis_user_'+#id") //在缓存中通过定义的键去查询，查到就返回，否则执行该方法，返回结果。并放入缓存
    	
    	@CacheEvict(value = "redisCache", key = "'redis_user_'+#id", beforeInvocation = false)//删除。是否在方法之前就移除，默认为false

### action ###

@Controller
    @RequestMapping("/user")
    public class UserController { 
        
        @Autowired
        private UserService userService = null;
        
        @RequestMapping("/getUser")
        @ResponseBody
        public User getUser(Long id) { 
            return userService.getUser(id);
        }
        
        @RequestMapping("/insertUser")
        @ResponseBody
        public User insertUser(String userName, String note) { 
            User user = new User();
            user.setUserName(userName);
            user.setNote(note);
            userService.insertUser(user);
            return user;
        }
        
        @RequestMapping("/findUsers")
        @ResponseBody
        public List<User> findUsers(String userName, String note) { 
            return userService.findUsers(userName, note);
        }
        
        @RequestMapping("/updateUserName")
        @ResponseBody
        public Map<String, Object> updateUserName(Long id, String userName) { 
            User user = userService.updateUserName(id, userName);
            boolean flag = user != null;
            String message = flag? "更新成功" : "更新失败";
            return resultMap(flag, message);
        }
        
        @RequestMapping("/deleteUser")
        @ResponseBody
        public Map<String, Object> deleteUser(Long id) { 
            int result = userService.deleteUser(id);
            boolean flag = result == 1;
            String message = flag? "删除成功" : "删除失败";
            return resultMap(flag, message);
        }
        
        private Map<String, Object> resultMap(boolean success, String message) { 
            Map<String, Object> result = new HashMap<String, Object>();
            result.put("success", success);
            result.put("message", message);
            return result;
        }
    }

### 缓存注解 自调用 失效的问题 ###

@Override
    	@Transactional
    	@CachePut(value = "redisCache", condition = "#result != 'null'", key = "'redis_user_'+#id")
    	public User updateUserName(Long id, String userName) { 
    		// 此处调用getUser方法，该方法缓存注解失效，
    		// 所以这里还会执行SQL，将查询到数据库最新数据
    		User user = this.getUser(id);
    		if (user == null) { 
    			return null;
    		}
    		user.setUserName(userName);
    		userDao.updateUser(user);
    		return user;
    	}

*  String的缓存机制 也是 基于SpringAop原理，Aop是通过动态代理实现的

## 缓存的脏数据问题 ##

*  修改id为1的用户，更新到数据库，使用key\_1 保存到redis
 *  修改id为1的用户，更新到数据库，使用key\_2保存到redis
 *  这样 key\_1 缓存为脏数据了。 （说不通，此时 为什么 不用 key\_1保存呢？）
 *  如redis事务。 数据的读写 策略不一样
 *  比如排名，允许脏读，不允许一直都是错的。
 *  可以设置redis失效时间
 *  也可设置redis的超时时间（实时性高的，超时时间设置短点）
 *  对于数据的写操作，一般会从数据库中先读取最新的数据，在更新数据

## 自定义缓存管理器 ##

*  不希望 采用Spring boot机制 带来的 键命名方式
 *  不希望缓存永不超时
 *  有两种方法：
    
     *  通过配置消除缓存 的前缀 和 自定义 超时时间 的属性 来定制生成创建 缓存管理器
     *  不采用spring boot为我们生成的，而是通过自己的代码

#缓存配置
    spring.cache.type=REDIS
    spring.cache.cache-names=redisCache
    
    #是否允许redis的值为空
    spring.cache.redis.cache-null-values=true
    
    #是否启用redis前缀。 false是禁用前缀
    spring.cache.redis.use-key-prefix=false
    #redis的前缀为
    spring.cache.redis.key-prefix=hua-test
    
    #缓存超时时间
    spring.cache.redis.time-to-live=600 000   //600 秒。 10分钟过后redis就会超时。

keys * #查询redis存在的键值对
    get redis_user_1 #获取Id为1的用户信息
    ttl redis_user_1 #查询键的剩余超时秒数

@Autowired //连接工厂
    	private RedisConnectionFactory connectionFactory = null;
    
    	@Bean(name = "redisCacheManager" )
    	public RedisCacheManager initRedisCacheManager() { 
    		// Redis加锁的写入器
    		RedisCacheWriter writer= RedisCacheWriter.lockingRedisCacheWriter(connectionFactory);
            
    		// 启动Redis缓存的默认设置
    		RedisCacheConfiguration config = RedisCacheConfiguration.defaultCacheConfig();
            
    		// 设置JDK序列化器
    		config = config.serializeValuesWith(SerializationPair.fromSerializer(new JdkSerializationRedisSerializer()));
            
    		// 禁用前缀
    		config = config.disableKeyPrefix();
            
    		//设置10分钟超时
    		config = config.entryTtl(Duration.ofMinutes(10));
            //Duration.ofMinutes(10).getSeconds() 600秒
            
    		// 创建缓Redis存管理器
    		RedisCacheManager redisCacheManager = new RedisCacheManager(writer, config);
    		return redisCacheManager;
    	}