在 Redis 中索引列

引言

与 memecached 不同，Redis 还可以用于持久化存储，而不仅仅是作为易失性缓存。事实上，Redis 是一个速度极快的数据库，如果使用得当，可以为您的应用程序提供显著更好的性能。当然，作为提醒，我想补充一点，使用 Redis 作为主要数据存储也存在风险，并且如果配置不当，这些风险会大大增加；因此，建议您在决定放弃“常规”SQL 数据库而改用 Redis 之前，进行充分的研究。

尽管它凭借其惊人的速度为您的应用程序带来了所有性能提升，但事实仍然是，Redis 本质上是一个键/值存储，并且不支持索引。因此，当您需要索引值以便稍后能够使用这些索引搜索和检索值时，可能会有些挑战。事实证明，我们可以通过使用 Redis 极其有用、原生提供的数据类型来绕过这种限制。在本文中，我们将探讨如何使用集合（sets）和有序集合（sorted sets）按日期对记录进行索引和排序，以及检索日期范围内的行。

背景

我使用 MySQL 已经有一段时间了，并且多年来我越来越喜欢它。就此而言，我将其用于各种企业应用程序，以及当我自由选择数据库用于 .NET 项目时。我不会很快放弃 MySQL；尽管如此，我在这里提到它，以便我们有一个适当的参考框架，在此基础上我可以说明以下内容。

我曾长期认为 Redis 只是与 memcached 相当（甚至可能更快）的产品。后来我偶然读到一篇 非常好的文章，其中提到 Redis 也可以用于持久化存储，这让我开始思考。于是，在一个周末，我发现自己——更多是出于好奇而不是其他原因——用 Redis 进行了一次快速的试验，结果相当令人满意。在此之后，我决定真正尝试一下：我一直在开发一个 .NET 应用程序，并且——由于项目尚处于早期阶段——我决定尝试将 Redis 作为主要的持久化存储引擎，从而完全抛弃 MySQL。我分别对这两个版本的代码进行了单元测试，Redis 版本和（之前的）MySQL 版本，结果令人震惊：性能提升了 5 倍！当然，这只是在我自己的情况下，您的体验可能有所不同。

话虽如此，Redis 的“问题”在于，它本质上是一个键/值存储，不支持索引。对于来自 SQL 世界的我来说，这是一个非常关键的要求，而这一点也通过 Redis 丰富的数据类型系统轻松解决了。Redis 不仅支持简单的标量值（简称为“string”，这其实是一个误称，因为它实际上可以存储比 string 更广泛的内容），还支持复合类型，如列表（lists）、集合（sets）、哈希（hashes），以及有序集合（sorted sets），这也是我将在本文中讨论的数据类型之一。

在此之前，我曾对如何索引日期感到非常困惑，尤其是当需要按日期范围检索记录时。事实证明，这可以通过使用 Redis 中的有序集合数据类型轻松实现。

关于 Redis 复杂类型

Redis 支持相当多的复杂类型。如前一节所述，我们有列表、集合、有序集合和哈希，尽管本文不会讨论列表和哈希。但是，我们将广泛使用其他两种类型，即集合和有序集合。

集合（Sets）正是其名称所示：一组唯一的值，它们之间没有顺序概念。而有序集合（Sorted sets）则略有不同。它们允许您存储带有分数的（scores）值，因此，有序集合的成员始终可以通过它们的分数，甚至分数范围来访问。事实证明，这在存储日期时非常有用。我们将存储的每个日期都转换为其“tick”值，从而获得一个有序的值集。然后，我们可以使用 Redis 的 ZRANGEBYSCORE 命令来检索落在特定范围内的值。

作为一个键-值存储，Redis 可以存储数据库中的几乎任何类型的值，只要它有一个唯一的键与之关联，无论该值是 string 类型还是复合类型。

Using the Code

代码效率不高：首先，我们将整个对象原样保存到所有索引中。“实际世界”的方法可能是将对象保存为标量“string”，并在各个索引中保存其 id。然而，不增加代码长度我无法做到这一点，所以我直接跳过了。这留给读者作为练习。

附件中包含几个文件。我们有一个 Person 类，它允许我们保存 Person 类对象，包含姓名、性别、国籍和出生日期等详细信息。我故意将该类设计得尽可能简单，以免处理不相关的细节。还有一个 static RedisAdaptor 类，其中包含用于保存 Person 对象和检索它们的辅助函数。

建议您查阅附件，以便更全面地理解代码。

外部库

就外部库而言，您可以在 这里找到不同语言/平台的 Redis 客户端库的全面列表。我使用了 StackExchange.Redis 库，因为它开源且许可条款非常友好，与 C# 部分的其他一些库相比。我还使用了 NewtonSoft JSON 库，可在 这里找到。后者库也对商业用途免费，但如果您愿意，也可以购买许可证。

主程序

在主程序中，我们有一个循环，遍历随意选择的年份 1971 的每一天，并创建一个新的 Person 类对象，其出生日期为当天的日期。至于性别，我们每创建一个对象就男女交替；对于国家，由于在这个虚构的例子中只指定了三个国家，每个人要么来自印度、美国，要么来自英国。对于姓名，我们使用一个随机生成的 string，实际上是一个 guid，去掉了所有非字母数字字符。

static void Main (string[] args) {
   const int YEAR = 1971;

   // We create one Person object for every single day in the given year.
   for (int month = 1; month <= 12; ++month) {
      for (int day = 1; day <= 31; ++day) {
         try {
         // Get any random name:
         string name = Util.GetAnyName ();
         // And a DoB:
         DateTime dob = new DateTime (YEAR, month, day);
         // As for the gender, let's alternate:
         Gender gender = Gender.FEMALE;
         if (day % 2 == 0) {
            gender = Gender.MALE;
         }
         // And the country, let's round-robin between all three:
         Country country = Country.INDIA;
         if (day % 3 == 1) {
            country = Country.USA;
         } else if (day % 3 == 2) {
            country = Country.GB;
         }

         // Create a new Person object:
         Person person = new Person (name, gender, country, dob);
         //Console.WriteLine ("Created new Person object: {0}", person);

         // We call the function that will store a new person in Redis:
            RedisAdaptor.StorePersonObject (person);
         } catch (Exception) {
            // If the control reaches here, it means the date was illegal.
            // So we just shrug your shoulders and move on to the next date.
            continue;
         }
      }
   }

   // At this point, we have 365 Person objects as a sorted set in our Redis database.

   // Next, let's take a date range and retrieve Person objects from within that range.
   DateTime fromDate = DateTime.Parse ("5-May-" + YEAR);
   DateTime toDate = DateTime.Parse ("7-May-" + YEAR);

   List<Person> persons = RedisAdaptor.RetrievePersonObjects (fromDate, toDate);

   Console.WriteLine ("Retrieved values in specified date range:");
      foreach (Person person in persons) {
      Console.WriteLine (person);
   }

   // Next, let's select some folks who are female AND from the USA.
   // This calls for a set intersection operation.
   List<Person> personsSelection = RedisAdaptor.RetrieveSelection (Gender.FEMALE, Country.USA);

   Console.WriteLine ("Retrieved values in selection:");
      foreach (Person person in personsSelection) {
         Console.WriteLine (person);
   }
}

RedisAdaptor 类有一个用于存储和索引传递给它的值的函数，以及分别用于按日期范围、按性别和按国家检索值的函数。我们还有一些 static 字段和常量用于此目的。

请注意，“索引”是在存储本身（自然地）完成的。在这种情况下，我们按性别、国家和出生日期进行索引。

static class RedisAdaptor {
   const string REDIS_HOST = "127.0.0.1";

   private static ConnectionMultiplexer _redis;

   // Date of birth key:
   const string REDIS_DOB_INDEX = "REDIS_DOB_INDEX";

   // Gender keys:
   const string REDIS_MALE_INDEX = "REDIS_MALE_INDEX";
   const string REDIS_FEMALE_INDEX = "REDIS_FEMALE_INDEX";

   // Country keys:
   const string REDIS_C_IN_INDEX = "REDIS_C_IN_INDEX";
   const string REDIS_C_USA_INDEX = "REDIS_C_USA_INDEX";
   const string REDIS_C_GB_INDEX = "REDIS_C_GB_INDEX";

   static RedisAdaptor () {
      // First, init the connection:
      _redis = ConnectionMultiplexer.Connect (REDIS_HOST);
   }

   public static void StorePersonObject (Person person) {
      // We first JSONize the object so that it's easier to save:
      string personJson = JsonConvert.SerializeObject (person);
      //Console.WriteLine ("JSONized new Person object: {0}", personJson);

      // And save it to Redis.

      // First, get the database object:
      IDatabase db = _redis.GetDatabase ();

      // Bear in mind that Redis is fundamentally a key-value store that does not provide
      // indexes out of the box.
      // We therefore work our way around this by creating and managing our own indexes.

      // The first index that we have is for gender.
      // We have two sets for this in Redis: one for males and the other for females.
      if (person.Gender == Gender.MALE) {
         db.SetAdd (REDIS_MALE_INDEX, personJson);
      } else {
         db.SetAdd (REDIS_FEMALE_INDEX, personJson);
      }

      // Next, we index by country.
      if (person.Country == Country.INDIA) {
         db.SetAdd (REDIS_C_IN_INDEX, personJson);
      } else if (person.Country == Country.USA) {
         db.SetAdd (REDIS_C_USA_INDEX, personJson);
      } else if (person.Country == Country.GB) {
         db.SetAdd (REDIS_C_GB_INDEX, personJson);
      }

      // Next, we need to create an index to be able to retrieve values that are in a particular
      // date range.

      // Since we need to index by date, we use the sorted set structure in Redis. Sorted sets
      // require a score (a real) to save a record. Therefore, in our case, we will use the
      // DoB's `ticks' value as the score.
      double dateTicks = (double) person.DoB.Ticks;

      db.SortedSetAdd (REDIS_DOB_INDEX, personJson, dateTicks);
   }

   public static List<Person> RetrievePersonObjects (DateTime fromDate, DateTime toDate) {
      // First. let's convert the dates to tick values:
      double fromTicks = fromDate.Ticks;
      double toTicks = toDate.Ticks;

      // And retrieve values from the sorted set.

      // First, get the database object:
      IDatabase db = _redis.GetDatabase ();

      RedisValue[] vals = db.SortedSetRangeByScore (REDIS_DOB_INDEX, fromTicks, toTicks);
      List<Person> opList = new List<Person> ();
      foreach (RedisValue val in vals) {
         string personJson = val.ToString ();
         Person person = JsonConvert.DeserializeObject<Person> (personJson);
         opList.Add (person);
      }

      return opList;
   }

   public static List<Person> RetrievePersonObjects (Gender gender) {
      // First, get the database object:
      IDatabase db = _redis.GetDatabase ();

      string keyToUse = gender == Gender.MALE ? REDIS_MALE_INDEX : REDIS_FEMALE_INDEX;

      RedisValue[] vals = db.SetMembers (keyToUse);

      List<Person> opList = new List<Person> ();
      foreach (RedisValue val in vals) {
         string personJson = val.ToString ();
         Person person = JsonConvert.DeserializeObject<Person> (personJson);
         opList.Add (person);
      }

      return opList;
   }

   public static List<Person> RetrievePersonObjects (Country country) {
      // First, get the database object:
      IDatabase db = _redis.GetDatabase ();

      string keyToUse = REDIS_C_IN_INDEX;
      if (country == Country.USA) {
         keyToUse = REDIS_C_USA_INDEX;
      } else if (country == Country.GB) {
         keyToUse = REDIS_C_GB_INDEX;
      }

      RedisValue[] vals = db.SetMembers (keyToUse);

      List<Person> opList = new List<Person> ();
      foreach (RedisValue val in vals) {
         string personJson = val.ToString ();
         Person person = JsonConvert.DeserializeObject<Person> (personJson);
         opList.Add (person);
      }

      return opList;
   }

   public static List<Person> RetrieveSelection (Gender gender, Country country) {
      // First, get the database object:
      IDatabase db = _redis.GetDatabase ();

      string keyToUseGender = gender == Gender.MALE ? REDIS_MALE_INDEX : REDIS_FEMALE_INDEX;
      string keyToUseCountry = REDIS_C_IN_INDEX;
      if (country == Country.USA) {
         keyToUseCountry = REDIS_C_USA_INDEX;
      } else if (country == Country.GB) {
         keyToUseCountry = REDIS_C_GB_INDEX;
      }

      RedisKey[] keys = new RedisKey[] { keyToUseGender, keyToUseCountry };

      RedisValue[] vals = db.SetCombine (SetOperation.Intersect, keys);

      List<Person> opList = new List<Person> ();
      foreach (RedisValue val in vals) {
         string personJson = val.ToString ();
         Person person = JsonConvert.DeserializeObject<Person> (personJson);
         opList.Add (person);
      }

      return opList;
   }
}

这个 static 类中定义的每个 const 键，如 REDIS_DOB_INDEX, REDIS_MALE_INDEX, REDIS_FEMALE_INDEX 等等，实际上是 Redis 存储中各个集合的键。

一旦我们保存了值并创建了它们的索引集合，我们就可以使用重载函数 RetrievePersonObjects 的各种版本，并传入日期范围、性别或国家参数来检索它们。

按性别检索非常直接：根据指定的性别，我们从两个性别集合中的一个中检索所需的值。按国家检索也是如此，其中我们为每个指定的国家都有三个唯一集合。

要按日期范围检索值，我们使用 Redis 数据库对象的 SortedSetRangeByScore 方法。它接受三个参数：第一个是有序集合本身的名称，其他两个是最小值和最大值。（您可以在 这里 阅读更多关于它的信息。）

使用 Redis 集合（sets）进行操作的一个更有趣的特性是它巧妙地使用了 集合论。您可以指定两个或多个集合，让 Redis 数据库对它们进行并集（union）、交集（intersection）或差集（difference）运算。在上面的最后一个代码段中，查看底部的 RetrieveSelection 函数，它声明了两个参数：Gender 和 Country。这个特定的函数根据同时满足 Person 的性别和他们的国籍的值。为此，我们使用了 Redis 数据库对象的 SetCombine 方法。

清理 Redis

随着您不断运行程序，您很快就会开始填充 Redis 实例的持久内存。如果您想删除 Redis 中已存储的先前值，我建议您从 Redis CLI 调用 flushdb （安装 Redis 和使用 CLI 的说明）来删除当前数据库中的所有键。您也可以使用更危险的 flushall 调用，但是，请注意，flushall 会删除所有数据库中的所有键！所以我请求您像使用枪一样谨慎使用 flushall：非常、非常小心。

关注点

嘿，如果您是 Redis 新手，我想补充一点，我们甚至还没有触及 Redis 能为您做的所有功能的表面。如果本文激起了您的好奇心，我建议您也了解一下开箱即用的 pub/sub 功能。

另一个非常巧妙的功能是 Redis 中可用的 HyperLogLog 这个东西 已经存在一段时间了，用于查找任何复杂类型的基数（或成员计数）。这本质上是一种具有 O(1) 复杂度的算法，另一个优点是它使用的空间量是固定的，无论成员数量如何。当然，权衡是它的准确性有点偏差，尽管这应该是可以接受的。

您可能还想查看为每种数据类型提供的非常精简的文档，以及更多内容，在此 处。

历史

• 第一版