title: Redis 数据类型和应用
date: 2020-06-24 10:45:38
categories:
  - 数据库
  - KV数据库
  - Redis
tags:
  - 数据库
  - KV数据库
  - Redis
  - 数据类型
permalink: /pages/ed757c/

127.0.0.1:6379> set hello world
OK
127.0.0.1:6379> get hello
"world"
127.0.0.1:6379> del hello
(integer) 1
127.0.0.1:6379> get hello
(nil)

127.0.0.1:6379> hset hash-key sub-key1 value1
(integer) 1
127.0.0.1:6379> hset hash-key sub-key2 value2
(integer) 1
127.0.0.1:6379> hset hash-key sub-key1 value1
(integer) 0
127.0.0.1:6379> hset hash-key sub-key3 value2
(integer) 0
127.0.0.1:6379> hgetall hash-key
1) "sub-key1"
2) "value1"
3) "sub-key2"
4) "value2"
127.0.0.1:6379> hdel hash-key sub-key2
(integer) 1
127.0.0.1:6379> hdel hash-key sub-key2
(integer) 0
127.0.0.1:6379> hget hash-key sub-key1
"value1"
127.0.0.1:6379> hgetall hash-key
1) "sub-key1"
2) "value1"

127.0.0.1:6379> rpush list-key item
(integer) 1
127.0.0.1:6379> rpush list-key item2
(integer) 2
127.0.0.1:6379> rpush list-key item
(integer) 3
127.0.0.1:6379> lrange list-key 0 -1
1) "item"
2) "item2"
3) "item"
127.0.0.1:6379> lindex list-key 1
"item2"
127.0.0.1:6379> lpop list-key
"item"
127.0.0.1:6379> lrange list-key 0 -1
1) "item2"
2) "item"

127.0.0.1:6379> sadd set-key item
(integer) 1
127.0.0.1:6379> sadd set-key item2
(integer) 1
127.0.0.1:6379> sadd set-key item3
(integer) 1
127.0.0.1:6379> sadd set-key item
(integer) 0
127.0.0.1:6379> smembers set-key
1) "item"
2) "item2"
3) "item3"
127.0.0.1:6379> sismember set-key item4
(integer) 0
127.0.0.1:6379> sismember set-key item
(integer) 1
127.0.0.1:6379> srem set-key item2
(integer) 1
127.0.0.1:6379> srem set-key item2
(integer) 0
127.0.0.1:6379> smembers set-key
1) "item"
2) "item3"

127.0.0.1:6379> zadd zset-key 728 member1
(integer) 1
127.0.0.1:6379> zadd zset-key 982 member0
(integer) 1
127.0.0.1:6379> zadd zset-key 982 member0
(integer) 0

127.0.0.1:6379> zrange zset-key 0 -1 withscores
1) "member1"
2) "728"
3) "member0"
4) "982"

127.0.0.1:6379> zrangebyscore zset-key 0 800 withscores
1) "member1"
2) "728"

127.0.0.1:6379> zrem zset-key member1
(integer) 1
127.0.0.1:6379> zrem zset-key member1
(integer) 0
127.0.0.1:6379> zrange zset-key 0 -1 withscores
1) "member0"
2) "982"

127.0.0.1:6379[15]> RPUSH 'sort-input' 23 15 110 7
(integer) 4
127.0.0.1:6379[15]> SORT 'sort-input'
1) "7"
2) "15"
3) "23"
4) "110"
127.0.0.1:6379[15]> SORT 'sort-input' alpha
1) "110"
2) "15"
3) "23"
4) "7"
127.0.0.1:6379[15]> HSET 'd-7' 'field' 5
(integer) 1
127.0.0.1:6379[15]> HSET 'd-15' 'field' 1
(integer) 1
127.0.0.1:6379[15]> HSET 'd-23' 'field' 9
(integer) 1
127.0.0.1:6379[15]> HSET 'd-110' 'field' 3
(integer) 1
127.0.0.1:6379[15]> SORT 'sort-input' by 'd-*->field'
1) "15"
2) "110"
3) "7"
4) "23"
127.0.0.1:6379[15]> SORT 'sort-input' by 'd-*->field' get 'd-*->field'
1) "1"
2) "3"
3) "5"
4) "9"

127.0.0.1:6379[15]> SET key value
OK
127.0.0.1:6379[15]> GET key
"value"
127.0.0.1:6379[15]> EXPIRE key 2
(integer) 1
127.0.0.1:6379[15]> GET key
(nil)

# 对不存在的 key 或者不存在的 offset 进行 GETBIT， 返回 0

redis> EXISTS bit
(integer) 0

redis> GETBIT bit 10086
(integer) 0


# 对已存在的 offset 进行 GETBIT

redis> SETBIT bit 10086 1
(integer) 0

redis> GETBIT bit 10086
(integer) 1

redis> BITCOUNT bit
(integer) 1

redis> PFADD  databases  "Redis"  "MongoDB"  "MySQL"
(integer) 1

redis> PFCOUNT  databases
(integer) 3

redis> PFADD  databases  "Redis"    # Redis 已经存在，不必对估计数量进行更新
(integer) 0

redis> PFCOUNT  databases    # 元素估计数量没有变化
(integer) 3

redis> PFADD  databases  "PostgreSQL"    # 添加一个不存在的元素
(integer) 1

redis> PFCOUNT  databases    # 估计数量增一
4

    public void articleVote(Jedis conn, String user, String article) {
        // 计算文章的投票截止时间。
        long cutoff = (System.currentTimeMillis() / 1000) - ONE_WEEK_IN_SECONDS;

        // 检查是否还可以对文章进行投票
        // （虽然使用散列也可以获取文章的发布时间，
        // 但有序集合返回的文章发布时间为浮点数，
        // 可以不进行转换直接使用）。
        if (conn.zscore("time:", article) < cutoff) {
            return;
        }

        // 从article:id标识符（identifier）里面取出文章的ID。
        String articleId = article.substring(article.indexOf(':') + 1);

        // 如果用户是第一次为这篇文章投票，那么增加这篇文章的投票数量和评分。
        if (conn.sadd("voted:" + articleId, user) == 1) {
            conn.zincrby("score:", VOTE_SCORE, article);
            conn.hincrBy(article, "votes", 1);
        }
    }

    public String postArticle(Jedis conn, String user, String title, String link) {
        // 生成一个新的文章ID。
        String articleId = String.valueOf(conn.incr("article:"));

        String voted = "voted:" + articleId;
        // 将发布文章的用户添加到文章的已投票用户名单里面，
        conn.sadd(voted, user);
        // 然后将这个名单的过期时间设置为一周（第3章将对过期时间作更详细的介绍）。
        conn.expire(voted, ONE_WEEK_IN_SECONDS);

        long now = System.currentTimeMillis() / 1000;
        String article = "article:" + articleId;
        // 将文章信息存储到一个散列里面。
        HashMap<String, String> articleData = new HashMap<String, String>();
        articleData.put("title", title);
        articleData.put("link", link);
        articleData.put("user", user);
        articleData.put("now", String.valueOf(now));
        articleData.put("votes", "1");
        conn.hmset(article, articleData);

        // 将文章添加到根据发布时间排序的有序集合和根据评分排序的有序集合里面。
        conn.zadd("score:", now + VOTE_SCORE, article);
        conn.zadd("time:", now, article);

        return articleId;
    }

    public List<Map<String, String>> getArticles(Jedis conn, int page, String order) {
        // 设置获取文章的起始索引和结束索引。
        int start = (page - 1) * ARTICLES_PER_PAGE;
        int end = start + ARTICLES_PER_PAGE - 1;

        // 获取多个文章ID。
        Set<String> ids = conn.zrevrange(order, start, end);
        List<Map<String, String>> articles = new ArrayList<>();
        // 根据文章ID获取文章的详细信息。
        for (String id : ids) {
            Map<String, String> articleData = conn.hgetAll(id);
            articleData.put("id", id);
            articles.add(articleData);
        }

        return articles;
    }

    public void addRemoveGroups(Jedis conn, String articleId, String[] toAdd, String[] toRemove) {
        // 构建存储文章信息的键名。
        String article = "article:" + articleId;
        // 将文章添加到它所属的群组里面。
        for (String group : toAdd) {
            conn.sadd("group:" + group, article);
        }
        // 从群组里面移除文章。
        for (String group : toRemove) {
            conn.srem("group:" + group, article);
        }
    }

    public List<Map<String, String>> getGroupArticles(Jedis conn, String group, int page, String order) {
        // 为每个群组的每种排列顺序都创建一个键。
        String key = order + group;
        // 检查是否有已缓存的排序结果，如果没有的话就现在进行排序。
        if (!conn.exists(key)) {
            // 根据评分或者发布时间，对群组文章进行排序。
            ZParams params = new ZParams().aggregate(ZParams.Aggregate.MAX);
            conn.zinterstore(key, params, "group:" + group, order);
            // 让Redis在60秒钟之后自动删除这个有序集合。
            conn.expire(key, 60);
        }
        // 调用之前定义的getArticles函数来进行分页并获取文章数据。
        return getArticles(conn, page, key);
    }

    public String checkToken(Jedis conn, String token) {
        // 尝试获取并返回令牌对应的用户。
        return conn.hget("login:", token);
    }

    public void updateToken(Jedis conn, String token, String user, String item) {
        // 获取当前时间戳。
        long timestamp = System.currentTimeMillis() / 1000;
        // 维持令牌与已登录用户之间的映射。
        conn.hset("login:", token, user);
        // 记录令牌最后一次出现的时间。
        conn.zadd("recent:", timestamp, token);
        if (item != null) {
            // 记录用户浏览过的商品。
            conn.zadd("viewed:" + token, timestamp, item);
            // 移除旧的记录，只保留用户最近浏览过的25个商品。
            conn.zremrangeByRank("viewed:" + token, 0, -26);
            conn.zincrby("viewed:", -1, item);
        }
    }

public static class CleanSessionsThread extends Thread {

    private Jedis conn;

    private int limit;

    private volatile boolean quit;

    public CleanSessionsThread(int limit) {
        this.conn = new Jedis("localhost");
        this.conn.select(15);
        this.limit = limit;
    }

    public void quit() {
        quit = true;
    }

    @Override
    public void run() {
        while (!quit) {
            // 找出目前已有令牌的数量。
            long size = conn.zcard("recent:");
            // 令牌数量未超过限制，休眠并在之后重新检查。
            if (size <= limit) {
                try {
                    sleep(1000);
                } catch (InterruptedException ie) {
                    Thread.currentThread().interrupt();
                }
                continue;
            }

            // 获取需要移除的令牌ID。
            long endIndex = Math.min(size - limit, 100);
            Set<String> tokenSet = conn.zrange("recent:", 0, endIndex - 1);
            String[] tokens = tokenSet.toArray(new String[tokenSet.size()]);

            // 为那些将要被删除的令牌构建键名。
            ArrayList<String> sessionKeys = new ArrayList<String>();
            for (String token : tokens) {
                sessionKeys.add("viewed:" + token);
            }

            // 移除最旧的那些令牌。
            conn.del(sessionKeys.toArray(new String[sessionKeys.size()]));
            conn.hdel("login:", tokens);
            conn.zrem("recent:", tokens);
        }
    }

}

    public void addToCart(Jedis conn, String session, String item, int count) {
        if (count <= 0) {
            // 从购物车里面移除指定的商品。
            conn.hdel("cart:" + session, item);
        } else {
            // 将指定的商品添加到购物车。
            conn.hset("cart:" + session, item, String.valueOf(count));
        }
    }

   while (!quit) {
        long size = conn.zcard("recent:");
        if (size <= limit) {
            try {
                sleep(1000);
            } catch (InterruptedException ie) {
                Thread.currentThread().interrupt();
            }
            continue;
        }

        long endIndex = Math.min(size - limit, 100);
        Set<String> sessionSet = conn.zrange("recent:", 0, endIndex - 1);
        String[] sessions = sessionSet.toArray(new String[sessionSet.size()]);

        ArrayList<String> sessionKeys = new ArrayList<String>();
        for (String sess : sessions) {
            sessionKeys.add("viewed:" + sess);
            // 新增加的这行代码用于删除旧会话对应用户的购物车。
            sessionKeys.add("cart:" + sess);
        }

        conn.del(sessionKeys.toArray(new String[sessionKeys.size()]));
        conn.hdel("login:", sessions);
        conn.zrem("recent:", sessions);
    }

    public String cacheRequest(Jedis conn, String request, Callback callback) {
        // 对于不能被缓存的请求，直接调用回调函数。
        if (!canCache(conn, request)) {
            return callback != null ? callback.call(request) : null;
        }

        // 将请求转换成一个简单的字符串键，方便之后进行查找。
        String pageKey = "cache:" + hashRequest(request);
        // 尝试查找被缓存的页面。
        String content = conn.get(pageKey);

        if (content == null && callback != null) {
            // 如果页面还没有被缓存，那么生成页面。
            content = callback.call(request);
            // 将新生成的页面放到缓存里面。
            conn.setex(pageKey, 300, content);
        }

        // 返回页面。
        return content;
    }

    public void scheduleRowCache(Jedis conn, String rowId, int delay) {
        // 先设置数据行的延迟值。
        conn.zadd("delay:", delay, rowId);
        // 立即缓存数据行。
        conn.zadd("schedule:", System.currentTimeMillis() / 1000, rowId);
    }

public class CacheRowsThread extends Thread {

    private Jedis conn;

    private boolean quit;

    public CacheRowsThread() {
        this.conn = new Jedis("localhost");
        this.conn.select(15);
    }

    public void quit() {
        quit = true;
    }

    @Override
    public void run() {
        Gson gson = new Gson();
        while (!quit) {
            // 尝试获取下一个需要被缓存的数据行以及该行的调度时间戳，
            // 命令会返回一个包含零个或一个元组（tuple）的列表。
            Set<Tuple> range = conn.zrangeWithScores("schedule:", 0, 0);
            Tuple next = range.size() > 0 ? range.iterator().next() : null;
            long now = System.currentTimeMillis() / 1000;
            if (next == null || next.getScore() > now) {
                try {
                    // 暂时没有行需要被缓存，休眠50毫秒后重试。
                    sleep(50);
                } catch (InterruptedException ie) {
                    Thread.currentThread().interrupt();
                }
                continue;
            }

            String rowId = next.getElement();
            // 获取下一次调度前的延迟时间。
            double delay = conn.zscore("delay:", rowId);
            if (delay <= 0) {
                // 不必再缓存这个行，将它从缓存中移除。
                conn.zrem("delay:", rowId);
                conn.zrem("schedule:", rowId);
                conn.del("inv:" + rowId);
                continue;
            }

            // 读取数据行。
            Inventory row = Inventory.get(rowId);
            // 更新调度时间并设置缓存值。
            conn.zadd("schedule:", now + delay, rowId);
            conn.set("inv:" + rowId, gson.toJson(row));
        }
    }

}

    public boolean canCache(Jedis conn, String request) {
        try {
            URL url = new URL(request);
            HashMap<String, String> params = new HashMap<>();
            if (url.getQuery() != null) {
                for (String param : url.getQuery().split("&")) {
                    String[] pair = param.split("=", 2);
                    params.put(pair[0], pair.length == 2 ? pair[1] : null);
                }
            }

            // 尝试从页面里面取出商品ID。
            String itemId = extractItemId(params);
            // 检查这个页面能否被缓存以及这个页面是否为商品页面。
            if (itemId == null || isDynamic(params)) {
                return false;
            }
            // 取得商品的浏览次数排名。
            Long rank = conn.zrank("viewed:", itemId);
            // 根据商品的浏览次数排名来判断是否需要缓存这个页面。
            return rank != null && rank < 10000;
        } catch (MalformedURLException mue) {
            return false;
        }
    }

    public void logRecent(Jedis conn, String name, String message, String severity) {
        String destination = "recent:" + name + ':' + severity;
        Pipeline pipe = conn.pipelined();
        pipe.lpush(destination, TIMESTAMP.format(new Date()) + ' ' + message);
        pipe.ltrim(destination, 0, 99);
        pipe.sync();
    }

    public static final int[] PRECISION = new int[] { 1, 5, 60, 300, 3600, 18000, 86400 };

    public void updateCounter(Jedis conn, String name, int count, long now) {
        Transaction trans = conn.multi();
        for (int prec : PRECISION) {
            long pnow = (now / prec) * prec;
            String hash = String.valueOf(prec) + ':' + name;
            trans.zadd("known:", 0, hash);
            trans.hincrBy("count:" + hash, String.valueOf(pnow), count);
        }
        trans.exec();
    }

    public List<Pair<Integer>> getCounter(
        Jedis conn, String name, int precision) {
        String hash = String.valueOf(precision) + ':' + name;
        Map<String, String> data = conn.hgetAll("count:" + hash);
        List<Pair<Integer>> results = new ArrayList<>();
        for (Map.Entry<String, String> entry : data.entrySet()) {
            results.add(new Pair<>(
                entry.getKey(),
                Integer.parseInt(entry.getValue())));
        }
        Collections.sort(results);
        return results;
    }

    public int ipToScore(String ipAddress) {
        int score = 0;
        for (String v : ipAddress.split("\\.")) {
            score = score * 256 + Integer.parseInt(v, 10);
        }
        return score;
    }

    public void importIpsToRedis(Jedis conn, File file) {
        FileReader reader = null;
        try {
            // 载入 csv 文件数据
            reader = new FileReader(file);
            CSVFormat csvFormat = CSVFormat.DEFAULT.withRecordSeparator("\n");
            CSVParser csvParser = csvFormat.parse(reader);
            int count = 0;
            List<CSVRecord> records = csvParser.getRecords();
            for (CSVRecord line : records) {
                String startIp = line.get(0);
                if (startIp.toLowerCase().indexOf('i') != -1) {
                    continue;
                }
                // 将 IP 地址转为整数值
                int score = 0;
                if (startIp.indexOf('.') != -1) {
                    score = ipToScore(startIp);
                } else {
                    try {
                        score = Integer.parseInt(startIp, 10);
                    } catch (NumberFormatException nfe) {
                        // 略过文件的第一行以及格式不正确的条目
                        continue;
                    }
                }

                // 构建唯一的城市 ID
                String cityId = line.get(2) + '_' + count;
                // 将城市 ID 及其对应的 IP 地址整数值添加到 ZSET
                conn.zadd("ip2cityid:", score, cityId);
                count++;
            }
        } catch (Exception e) {
            throw new RuntimeException(e);
        } finally {
            try {
                reader.close();
            } catch (Exception e) {
                // ignore
            }
        }
    }

    public void importCitiesToRedis(Jedis conn, File file) {
        Gson gson = new Gson();
        FileReader reader = null;
        try {
            // 加载 csv 信息
            reader = new FileReader(file);
            CSVFormat csvFormat = CSVFormat.DEFAULT.withRecordSeparator("\n");
            CSVParser parser = new CSVParser(reader, csvFormat);
            // String[] line;
            List<CSVRecord> records = parser.getRecords();
            for (CSVRecord record : records) {

                if (record.size() < 4 || !Character.isDigit(record.get(0).charAt(0))) {
                    continue;
                }

                // 将城市地理信息转为 json 结构，存入 HASH 结构中
                String cityId = record.get(0);
                String country = record.get(1);
                String region = record.get(2);
                String city = record.get(3);
                String json = gson.toJson(new String[] { city, region, country });
                conn.hset("cityid2city:", cityId, json);
            }
        } catch (Exception e) {
            throw new RuntimeException(e);
        } finally {
            try {
                reader.close();
            } catch (Exception e) {
                // ignore
            }
        }
    }

    public String[] findCityByIp(Jedis conn, String ipAddress) {
        int score = ipToScore(ipAddress);
        Set<String> results = conn.zrevrangeByScore("ip2cityid:", score, 0, 0, 1);
        if (results.size() == 0) {
            return null;
        }

        String cityId = results.iterator().next();
        cityId = cityId.substring(0, cityId.indexOf('_'));
        return new Gson().fromJson(conn.hget("cityid2city:", cityId), String[].class);
    }

# -----------------------------------------------------------
# Redis 职位搜索数据模型示例
# -----------------------------------------------------------

# （1）职位技能表：使用 set 存储
# job:001 职位添加 4 种技能
SADD job:001 skill:001
SADD job:001 skill:002
SADD job:001 skill:003
SADD job:001 skill:004

# job:002 职位添加 3 种技能
SADD job:002 skill:001
SADD job:002 skill:002
SADD job:002 skill:003

# job:003 职位添加 2 种技能
SADD job:003 skill:001
SADD job:003 skill:003

# 查看
SMEMBERS job:001
SMEMBERS job:002
SMEMBERS job:003

# （2）求职者技能表：使用 set 存储
SADD interviewee:001 skill:001
SADD interviewee:001 skill:003

SADD interviewee:002 skill:001
SADD interviewee:002 skill:002
SADD interviewee:002 skill:003
SADD interviewee:002 skill:004
SADD interviewee:002 skill:005

# 查看
SMEMBERS interviewee:001
SMEMBERS interviewee:002

# （3）求职者遍历查找自己符合要求的职位（返回结果为 empty 表示要求的技能全部命中）
# 比较职位技能清单和求职者技能清单的差异
SDIFF job:001 interviewee:001
SDIFF job:002 interviewee:001
SDIFF job:003 interviewee:001

SDIFF job:001 interviewee:002
SDIFF job:002 interviewee:002
SDIFF job:003 interviewee:002

# （4）用人公司根据遍历查找符合自己职位要求的求职者（返回结果为 empty 表示要求的技能全部命中）
# 比较职位技能清单和求职者技能清单的差异
SDIFF interviewee:001 job:001
SDIFF interviewee:002 job:001

SDIFF interviewee:001 job:002
SDIFF interviewee:002 job:002

SDIFF interviewee:001 job:003
SDIFF interviewee:002 job:003