Redis - Redis 入门实战 - 《Java架构技术》

Jedis
高并发分布式锁

Jedis

Jedis是Java和Redis打交道的API客户端

<dependency>
<groupId>redis.clients</groupId>
<artifactId>jedis</artifactId>
<version>3.1.0</version>
</dependency>

注意修改：redis.conf [bind 0.0.0.0]允许任何IP访问，关闭防火墙或者开放6379端口
创建maven项目：

public class Test {
    public static void main(String[] args) {
        Jedis jedis = new Jedis("172.16.150.132", 6379);
        String ping = jedis.ping();
        System.out.println(ping);//返回 PONG 说明连接成功
    }
}

常用的API

package com.sufulu;

import redis.clients.jedis.Jedis;

import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Set;

public class RedisApiTest {

    private void testString() {
        Jedis jedis = new Jedis("172.16.150.132", 6379);
        String ping = jedis.ping();
        System.out.println(ping);//返回 PONG 说明连接成功
        //存入三条数据
        jedis.set("k1", "v1");
        jedis.set("k2", "v2");
        jedis.set("k3", "v3");

        //查询全部
        Set<String> keys = jedis.keys("*");
        Iterator<String> iterator = keys.iterator();
        while (iterator.hasNext()) {
            String k = iterator.next();
            //jedis.get(k)获取值
//            System.out.println(k + "->" + jedis.get(k));
        }
        //判断某个键是否存在
        Boolean k2Exists = jedis.exists("k2");
        System.out.println("k2Exists:" + k2Exists);//true
        //查看k1的过期时间
        System.out.println(jedis.ttl("k1"));//-1

        //设置多个键
        jedis.mset("k4", "v4", "k5", "v5");

        //String类型
        System.out.println("-------------------------------------------");

        //获取多个值
        System.out.println(jedis.mget("k1", "k2", "k3", "k4", "k5"));

        System.out.println("----------------------------------------");
    }

    private void testList() {
        Jedis jedis = new Jedis("172.16.150.132", 6379);

        //list
        jedis.lpush("list01", "l1", "l2", "l3", "l4", "l5");

        List<String> list01 = jedis.lrange("list01", 0, -1);
        for (String s : list01) {
            System.out.println(s);
        }

        System.out.println("----------------------------------");
    }

    private void testSet() {
        Jedis jedis = new Jedis("172.16.150.132", 6379);
        // set
        jedis.sadd("order", "001");
        jedis.sadd("order", "002");
        jedis.sadd("order", "003");

        Set<String> order = jedis.smembers("order");
        Iterator<String> iterator1 = order.iterator();
        while (iterator1.hasNext()) {
            String next = iterator1.next();
            System.out.println(next);
        }

        //删除
        jedis.srem("order", "002");
        System.out.println(jedis.smembers("order").size());
    }

    private void testHash() {
        Jedis jedis = new Jedis("172.16.150.132", 6379);
        jedis.hset("hash01", "username", "james");
        System.out.println(jedis.hget("hash01", "username"));

        HashMap<String, String> map = new HashMap<>();
        map.put("gender", "boy");
        map.put("address", "beijing");
        map.put("phone", "123131");
        jedis.hset("person", map);

        //获取多个属性的值
        List<String> list = jedis.hmget("person", "phone", "address");
        for (String s : list) {
            System.out.println(s);
        }
    }

    private void testZset() {
        Jedis jedis = new Jedis("172.16.150.132", 6379);

        jedis.zadd("zset01", 60d, "zs1");
        jedis.zadd("zset01", 70d, "zs2");
        jedis.zadd("zset01", 80d, "zs3");
        jedis.zadd("zset01", 90d, "zs4");
        jedis.zadd("zset01", 100d, "zs5");

        Set<String> zset01 = jedis.zrange("zset01", 0, -1);
        Iterator<String> iterator = zset01.iterator();
        while (iterator.hasNext()) {
            String next = iterator.next();
            System.out.println(next);
        }
    }

    public static void main(String[] args) {
        new RedisApiTest().testZset();
    }
}

事务控制

    public static void main(String[] args) throws InterruptedException {
        Jedis jedis = new Jedis("172.16.150.132", 6379);
        int yue = Integer.parseInt(jedis.get("yue"));
        int zhichu = 10;

        //监控余额
        jedis.watch("yue");
        //模拟网络延迟
        Thread.sleep(10000);//在执行后 去修改yue的值为10 模拟多线程操作

        if (yue < zhichu) {
            jedis.unwatch();//解除监控
            System.out.println("余额不足");
        } else {
            //开启事务
            Transaction transaction = jedis.multi();
            transaction.decrBy("yue", zhichu);//余额减少10元
            transaction.incrBy("zhichu", zhichu);//累计消费增加
            //执行事务
            transaction.exec();
            System.out.println("余额：" + jedis.get("yue"));
            System.out.println("累计支出:" + jedis.get("zhichu"));
        }
    }

JedisPool

redis的连接池技术。https://help.aliyun.com/document_detail/98726.html

<dependency>
<groupId>commons-pool</groupId>
<artifactId>commons-pool</artifactId>
<version>1.6</version>
</dependency>

封装连接池

public class JedisPoolUtils {
    private JedisPoolUtils() {
    }

    private volatile static JedisPool jedisPool = null;

    private volatile static Jedis jedis = null;

    //返回一个连接池
    private static JedisPool getInstance() {
        //双层检测锁
        if (jedisPool == null) {
            synchronized (JedisPoolUtils.class) {
                if (jedisPool == null) {
                    JedisPoolConfig jedisPoolConfig = new JedisPoolConfig();
                    jedisPoolConfig.setMaxTotal(1000);//最大连接数量
                    jedisPoolConfig.setMaxIdle(30);//最大等待数量
                    jedisPoolConfig.setMaxWaitMillis(60 * 1000);//最大等待时间
                    jedisPoolConfig.setTestOnBorrow(true);//后台运行
                    jedisPool = new JedisPool(jedisPoolConfig, "172.16.150.132", 6379);
                }
            }
        }
        return jedisPool;
    }

    //返回jedis
    public static Jedis getJedis() {
        if (jedis == null) {
            jedis = getInstance().getResource();
        }
        return jedis;
    }
}

高并发分布式锁

经典案例：秒杀、抢购优惠券等。

引入相关的依赖：

<dependencies>
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-webmvc</artifactId>
            <version>5.2.7.RELEASE</version>
        </dependency>
        <!-- 实现分布式锁的工具类 -->
        <dependency>
            <groupId>org.redisson</groupId>
            <artifactId>redisson</artifactId>
            <version>3.6.1</version>
        </dependency>
        <!-- spring操作Redis的工具类 -->
        <dependency>
            <groupId>org.springframework.data</groupId>
            <artifactId>spring-data-redis</artifactId>
            <version>2.3.2.RELEASE</version>
        </dependency>
        <!-- redis 客户端 -->
        <dependency>
            <groupId>redis.clients</groupId>
            <artifactId>jedis</artifactId>
            <version>3.1.0</version>
        </dependency>
        <!-- json解析工具 -->
        <dependency>
            <groupId>com.fasterxml.jackson.core</groupId>
            <artifactId>jackson-databind</artifactId>
            <version>2.9.8</version>
        </dependency>
    </dependencies>

spring核心配置文件进行配置：

<?xml version="1.0" encoding="UTF-8" ?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:context="http://www.springframework.org/schema/context"
       xsi:schemaLocation="
        http://www.springframework.org/schema/beans
        http://www.springframework.org/schema/beans/spring-beans.xsd
        http://www.springframework.org/schema/context
        http://www.springframework.org/schema/context/spring-context.xsd">

    <!-- 开启注解扫描 -->
    <context:component-scan base-package="controller"/>

    <!-- spring连接Redis的工具类 -->
    <bean id="stringRedisTemplate" class="org.springframework.data.redis.core.StringRedisTemplate">
        <!-- 连接工厂 -->
        <property name="connectionFactory" ref="connectionFactory">

        </property>
    </bean>

    <!-- connectionFactory 提供主机ip和port -->
    <bean id="connectionFactory" class="org.springframework.data.redis.connection.jedis.JedisConnectionFactory">
        <property name="hostName" value="172.16.150.132"/>
        <property name="port" value="6379"/>
    </bean>
</beans>

web.xml文件进行配置

<?xml version="1.0" encoding="UTF-8"?>
<web-app xmlns="http://xmlns.jcp.org/xml/ns/javaee"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://xmlns.jcp.org/xml/ns/javaee http://xmlns.jcp.org/xml/ns/javaee/web-app_4_0.xsd"
         version="4.0">
    <servlet>
        <servlet-name>dispatcherServlet</servlet-name>
        <servlet-class>org.springframework.web.servlet.DispatcherServlet</servlet-class>
        <init-param>
            <param-name>contextConfigLocation</param-name>
            <param-value>classpath:spring/spring.xml</param-value>
        </init-param>
        <load-on-startup>2</load-on-startup>
    </servlet>
    <servlet-mapping>
        <servlet-name>dispatcherServlet</servlet-name>
        <url-pattern>/</url-pattern>
    </servlet-mapping>
</web-app>

准备秒杀数据：

127.0.0.1:6379> set phone 10 # 设置10台手机
OK

编写controller类

@Controller
public class TestKill {

    @Autowired
    private StringRedisTemplate springRedisTemplate;

    @RequestMapping("kill")
    @ResponseBody
    public String kill() {
        //1. 从Redis获取 手机库存数量
        int phoneCount = Integer.parseInt(springRedisTemplate.opsForValue().get("phone"));
        //2. 判断手机的数量是否够秒杀的
        if (phoneCount > 0) {
            phoneCount--;
            //修改Redis的库存数据
            springRedisTemplate.opsForValue().set("phone", String.valueOf(phoneCount));
            System.out.println("库存减一，剩余：" + phoneCount);
        } else {
            return "count not";
        }

        return "over!";
    }
}

使用jmeter并发请求1S发送100个。后台打印的数据

打印结果如下：

解决可以在controller方法上添加synchronized 是可以解决上述问题的，但是synchronized 只能锁一个进程下的线程并发，如果分布式环境多个进程并发，这种方案就失效了。

测试多个进程并发访问

在虚拟机配置nginx,反向代理到本机的IP和端口，在本机启动两个Tomcat8001和8002端口，然后修改本机的host：添加虚拟机ip地址 ``www.zk.com

8002 tomcat请求：

8001 Tomcat请求：

可以明显发现问题。

实现分布式锁的思路

redis是单线程，命名具备原子性，使用setnx命令(判断key是否存在)实现锁，保存k-v
1. 如果k不存在，保存（当前线程加锁），执行完成后，删除k表示释放锁
2. 如果k存在，阻塞线程执行，表示有锁

类似的伪代码如下：

    @RequestMapping("kill")
    @ResponseBody
    public synchronized String kill() {
        //处理出现异常的情况，当程序异常在finally中释放锁，否则会造成死锁
        try {
            boolean b = setnx(lock laosun); //返回1/true 表示加锁成功
            if(!b){ //false/0 表示已经存在了，有锁
             return "已经有锁了 滚蛋"
            }
        //1. 从Redis获取 手机库存数量
        int phoneCount = Integer.parseInt(springRedisTemplate.opsForValue().get("phone"));
        //2. 判断手机的数量是否够秒杀的
        if (phoneCount > 0) {
            phoneCount--;
            //修改Redis的库存数据
            springRedisTemplate.opsForValue().set("phone", String.valueOf(phoneCount));
            System.out.println("库存减一，剩余：" + phoneCount);
        } else {
            System.out.println("库存不足");
            return "count not";
        }
        }finally{
        //释放锁
            delete(lock)
        }
        return "over!";
    }

如果加锁成功，在执行业务代码的过程中出现异常，导致没有删除k(释放锁失败)，那么就会造成死锁(后面的线程都无法执行)
1. 设置过期时间，例如10s后（setex），Redis自动删除
高并发下，由于时间段等因素导致服务器压力过大或过小，每个线程执行的时间不同。
1. 第一线程，执行需要13秒，执行到第10秒时，Redis自动过期了k释放锁
2. 第二个线程，执行需要7秒，加锁，执行到第3秒(锁被释放了，为什么？是被第一个线程的finally主动deleteKey释放掉了)
3. 。。。。连锁反应，当线程刚加的锁，就被其他线程释放掉了，周而复始，导致锁会永久失效
给每个线程加上唯一的标识UUID随机生成，释放的时候判断是否是当前的标识即可
那么问题又来了，过期时间如何设定呢？
1. 如果10秒太短不够用怎么办？设置60秒又太长浪费时间
2. 可以开启一个定时器线程，当过期时间小于总过期时间的1/3时，增长总过期时间

如果要自己实现分布式锁，太难了。

Redisson

Redis 是最流行的NoSQL数据库解决方案之一，而Java是世界上最流行的编程语言一致
Redisson 就是用于在Java程序中操作Redis的库，它使得我们可以在程序中轻松地使用Redis，Redisson在java.until中常用接口的基础上，提供了一系列具有分布式特性的工具类
```
      <dependency>
          <groupId>org.redisson</groupId>
          <artifactId>redisson</artifactId>
          <version>3.6.1</version>
      </dependency>
```
使用redisson: ```java package controller;

import org.redisson.Redisson; import org.redisson.config.Config; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Bean; import org.springframework.data.redis.core.StringRedisTemplate; import org.springframework.stereotype.Controller; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.ResponseBody;

@Controller public class TestKill {

@Autowired
private StringRedisTemplate springRedisTemplate;

@Autowired
private Redisson redisson;

@RequestMapping("kill")
@ResponseBody
public synchronized String kill() {
    //定义商品id，这里写死即可
    String productKey = "HUAWEI-P40";
    //通过redisson获得锁
    RLock lock = redisson.getLock(productKey);//底层源码就是集成了setnx setex等操作

    //上锁 过期时间为30秒
    lock.lock(30, TimeUnit.SECONDS);
    try {
        //1. 从Redis获取 手机库存数量
        int phoneCount = Integer.parseInt(springRedisTemplate.opsForValue().get("phone"));
        //2. 判断手机的数量是否够秒杀的
        if (phoneCount > 0) {
            phoneCount--;
            //修改Redis的库存数据
            springRedisTemplate.opsForValue().set("phone", String.valueOf(phoneCount));
            System.out.println("库存减一，剩余：" + phoneCount);
        } else {
            System.out.println("库存不足");
            return "count not";
        }
    } catch (Exception e) {
        e.printStackTrace();
    } finally {
        lock.unlock();
    }
    return "over!";
}

@Bean
public Redisson redisson() {
    Config config = new Config();
    //使用单个redis服务器 传递Redis服务器的IP和端口号，以及选择几号数据库
    config.useSingleServer().setAddress("redis://172.16.150.132:6379").setDatabase(0);
    //使用Redis集群 主从复制
    config.useClusterServers().setScanInterval(2000).addNodeAddress("redis://172.16.150.132:6379",
            "redis://172.16.150.131:6379", "redis://172.16.150.130:6379");
    return (Redisson) Redisson.create(config);
}

}

``` 然后启动两个Tomcat，使用Nginx负载均衡，使用jmeter做并发测试1S并发200个请求
8001Tomcat：

8002Tomcat:

可以看出来，两个Tomcat没有重复的数据。

:::tips 实现分布式锁的方案有很多，之前使用的Zookeeper的特点就是高可靠性，Redis的特点是高性能。目前分布式锁，应用最多的仍然是redis :::