Redis 系列教程导航如何在ubuntu18.04上安装和保护redis如何连接到Redis数据库如何管理Redis数据库和Keys如何在Redis中管理副本和客户端如何在Redis中管理字符串如何在Redis中管理list如何在Redis中管理Hashes如何在Redis中管理Sets如何在Redis中管理Sorted Sets如何在Redis中运行事务如何使Redis中的Key失效如何解决Redis中的故障如何从命令行更改Redis的配置Redis数据类型简介Redis 中如何使用 lua 脚本Redis 常用命令指南介绍Redis 是一个开源的内存键值存储系统。它最受欢迎的功能之一是对复制（Replication）的支持：任何 Redis 服务器都可以将其数据复制到任意数量的副本中，从而实现高读取可伸缩性和强大的数据冗余性。此外，Redis 设计允许许多客户端（默认情况下最多 10,000 个）连接数据并与之交互，因此对于许多用户需要访问同一数据集的情况而言，它是一个不错的选择。本教程介绍了用于管理 Redis 客户端和副本的命令。如何使用本指南本指南以备有完整示例的速查表（Cheatsheet）形式编写。我们鼓励您跳至与您要完成的任务相关的任何部分。本指南中显示的命令已在运行 Redis 版本 4.0.9 的 Ubuntu 18.04 服务器上进行了测试。要设置类似的环境，您可以按照我们的指南 如何在 Ubuntu 18.04 上安装和保护 Redis 的 步骤 1 进行操作。我们将通过使用 Redis 命令行界面 redis-cli 运行它们来演示这些命令的行为。请注意，如果您使用其他 Redis 界面（例如 Redli），则某些命令的确切输出可能会有所不同。另外，您可以提供一个托管的 Redis 数据库实例来测试这些命令，但是请注意，根据数据库提供者所允许的控制级别，本指南中的某些命令可能无法按所述方式工作。要配置 DigitalOcean 托管数据库，请遵循我们的 托管数据库产品文档。然后，您必须 安装 Redli 或 设置 TLS 隧道 才能通过 TLS 连接到托管数据库。注意： Redis 项目在其文档和各种命令中使用术语“主（Master）”和“从（Slave）”来标识复制中的不同角色，尽管该项目的参与者 正在采取措施 在不引起兼容性问题的情况下 更改此语言。DigitalOcean 通常更喜欢使用替代术语“主要（Primary）”和“副本（Replica）”。本指南将在可能的情况下默认设置为“主要”和“副本”，但请注意，在某些情况下不可避免地会出现“主”和“从”这两个术语。管理副本Redis 最具特色的功能之一是其 内置的复制功能。使用复制时，Redis 创建主实例的精确副本。这些辅助实例在其连接断开时会随时重新连接至主要实例，并且始终旨在保持主要实例的精确副本。如果不确定当前连接的 Redis 实例是主实例还是副本，则可以通过运行以下 role 命令进行检查：role如果您使用 Redis Sentinel，则此命令将返回 master 或 slave，或者可能返回 sentinel。要将 Redis 实例动态指定为另一个实例的副本，请运行 replicaof 命令。此命令将预期的主服务器的主机名或 IP 地址和端口作为参数：replicaof hostname_or_IP port如果服务器已经是另一台主服务器的副本，它将停止复制旧服务器，并立即开始与新服务器同步。它还将丢弃旧数据集。要将副本提升为主副本，请运行以下 replicaof 命令：replicaof no one这将阻止实例复制主服务器，但不会丢弃已复制的数据集。在原始主数据库失败的情况下，此语法很有用。在 replicaof no one 发生故障的主数据库的副本上运行之后，以前的副本可以用作新的主数据库，并拥有自己的副本作为故障保护。注意： 在 5.0.0 之前的版本中，Redis 包含了此命令的一个名为 slaveof 的版本。管理客户端一个 客户端) 是任何机器或软件连接到服务器才能访问服务。Redis 附带了一些命令，可帮助跟踪和管理客户端连接。该 client list 命令返回一组有关当前客户端连接的可读信息：client listOutput "id=18165 addr=[2001:db8:0:0::12]:47460 fd=7 name=jerry age=72756 idle=0 flags=N db=0 sub=0 psub=0 multi=-1 qbuf=0 qbuf-free=0 obl=0 oll=0 omem=0 events=r cmd=ping id=18166 addr=[2001:db8:0:1::12]:47466 fd=8 name= age=72755 idle=5 flags=N db=0 sub=0 psub=0 multi=-1 qbuf=0 qbuf-free=0 obl=0 oll=0 omem=0 events=r cmd=info id=19381 addr=[2001:db8:0:2::12]:54910 fd=9 name= age=9 idle=0 flags=N db=0 sub=0 psub=0 multi=-1 qbuf=26 qbuf-free=32742 obl=0 oll=0 omem=0 events=r cmd=client "以下是每个字段的含义：字段含义id唯一的 64 位客户端 IDname客户端连接的名称，由先前的 client setname 命令定义addr客户端连接的地址和端口fd与客户端所连接的套接字相对应的 文件描述符age客户端连接的总时长，以秒为单位flags一组一个或多个单字符标志，可提供有关客户端的更详细的信息；有关更多详细信息，请参见 client list 命令文档db客户端连接到的当前数据库 ID 号（可以从 0 到 15）sub客户端订阅的频道数psub客户端的模式匹配订阅数multi客户端已在 事务中 排队的命令数（将显示 -1 客户端是否尚未开始事务或 0 是否仅已开始事务且未排队任何命令）qbuf客户端的查询缓冲区长度，0 这意味着它没有待处理的查询qbuf-free客户端查询缓冲区中的可用空间量，0 表示查询缓冲区已满obl客户端的输出缓冲区长度oll客户端的输出列表的长度，当客户端的缓冲区已满时，将在其中排队答复omem客户端的输出缓冲区使用的内存events客户端的文件描述符事件，可以是 r“可读”，w“可写”或两者兼有cmd客户端运行的最后一条命令设置客户端名称对于调试使用 Redis 的任何应用程序中的连接泄漏都很有用。每个新连接都将在开始时没有分配的名称，但 client setname 可用于为当前客户端连接创建一个新连接。尽管 Redis 通常将字符串长度限制为 512 MB，但是对客户端名称的长度没有限制。但是请注意，客户端名称不能包含空格：client setname elaine要检索客户端连接的名称，请使用以下 client getname 命令：client getnameOutput "elaine"要获取客户端的连接 ID，请使用以下 client id 命令：client id"19492"Redis 客户端 ID 永远不会重复，并且会单调递增。这意味着，如果一个客户端的 ID 大于另一个客户端，则它是在以后的时间建立的。阻止客户端并关闭客户端连接复制系统通常被描述为 同步 或 _异步_。同步复制：每当客户端添加或更改数据时，客户端都必须从一定数量的副本中收到某种确认，以使更改注册为已提交。这有助于防止节点发生数据冲突，但是这会增加延迟，因为客户端必须等待执行另一项操作，直到从一定数量的副本中收到回音为止。异步复制：一旦将数据写入本地存储，客户端就会看到确认操作已完成的确认。但是，这与副本实际写入数据的时间之间可能会有滞后。如果其中一个副本在写入更改之前失败，则该写入将永远丢失。因此，尽管异步复制允许客户端继续执行操作而没有等待副本引起的延迟，但它可能导致节点之间的数据冲突，并且可能需要数据库管理员方面的额外工作来解决这些冲突。由于注重性能和低延迟，Redis 默认情况下实现异步复制。但是，您可以使用该 wait 命令模拟同步复制。wait 在指定的时间（以毫秒为单位）内阻止当前客户端连接，直到所有先前的写命令成功传输并被指定数量的副本接受为止。此命令使用以下语法：wait number_of_replicas number_of_milliseconds例如，如果要在 30 毫秒的超时时间内至少 3 个副本注册所有先前的写入操作之前阻止客户端连接，则 wait 语法应如下所示：wait 3 30该 wait 命令返回一个整数，该整数表示已确认写入命令的副本数，即使并非每个副本都这样做：2要解除阻止以前通过 wait（brpop，或 xread）命令阻止的客户端连接，可以 client unblock 使用以下语法运行命令：client unblock client_id要暂时挂起当前连接到 Redis 服务器的每个客户端，可以使用以下 client pause 命令。这在需要以受控方式更改 Redis 设置的情况下很有用。例如，如果要将一个副本提升为主实例，则可以事先暂停每个客户端，以便升级副本并使客户端作为新的主实例连接到该副本，而不会丢失任何写操作。该 client pause 命令要求您指定要挂起客户端的时间（以毫秒为单位）。下面的示例将所有客户端暂停一秒钟：client pause 1000该 client kill 语法允许您基于许多不同的过滤器关闭单个连接或一组特定的连接。语法如下所示：client kill filter_1 value_1 ... filter_n value_n在 Redis 2.8.12 和更高版本中，可以使用以下过滤器：addr：允许您从指定的 IP 地址和端口关闭客户端连接client-id：允许您根据其唯一 ID 字段关闭客户端连接type：关闭一个给定类型，它可以是每一个客户端 normal，master，slave，或 pubsubskipme：此过滤器的值选项为 yes 和 no：如果 no 指定了，则调用该 client kill 命令的客户端不会被跳过，并且如果其他过滤器应用到该客户端，它将被杀死如果 yes 指定，则将跳过运行该命令的客户端，而 kill 命令将对该客户端无效。skipme 始终 yes 默认为结论本指南详细介绍了用于管理 Redis 客户端和副本的许多命令。如果您想在本指南中概述其他相关的命令、参数或过程，请在下面的评论中提出疑问或提出建议。有关 Redis 命令的更多信息，请参阅关于 如何管理 Redis 数据库的 系列教程。说明： 本文内容基于 Redis 4.0.9 版本及 Ubuntu 18.04 环境测试。请注意，Redis 5.0.0 及以上版本在复制命令上有所变更（如 slaveof 变更为 replicaof），部分命令行为在新版本中可能有所不同。
系列教程导航如何在 ubuntu18.04 上安装和保护 redis如何连接到 Redis 数据库如何管理 Redis 数据库和 Keys如何在 Redis 中管理副本和客户端如何在 Redis 中管理字符串如何在 Redis 中管理 list如何在 Redis 中管理 Hashes如何在 Redis 中管理 Sets如何在 Redis 中管理 Sorted Sets如何在 Redis 中运行事务如何使 Redis 中的 Key 失效如何解决 Redis 中的故障如何从命令行更改 Redis 的配置Redis 数据类型简介如何在 Redis 中管理字符串介绍Redis 是一个开源的内存中键值数据存储系统。在 Redis 中，字符串 是您可以创建和管理的最基本值类型。本教程概述了如何创建、检索字符串，以及如何操作字符串键所保存的值。如何使用本指南本指南采用备忘单（Cheatsheet）形式编写，包含完整示例。我们鼓励您直接跳至与您任务相关的部分。本指南中显示的命令已在运行 Redis 4.0.9 的 Ubuntu 18.04 服务器上进行了测试。要设置类似的环境，您可以按照指南 如何在 Ubuntu 18.04 上安装和保护 Redis 的步骤 1进行操作。我们将通过使用 Redis 命令行界面 redis-cli 运行它们来演示这些命令的行为。请注意，如果您使用其他 Redis 界面（例如 Redli），则某些命令的确切输出可能会有所不同。另外，您可以使用托管的 Redis 数据库实例来测试这些命令。但请注意，根据数据库提供者所允许的控制级别，本指南中的某些命令可能无法按所述方式工作。要配置 DigitalOcean 托管数据库，请遵循 托管数据库产品文档。然后，您必须 安装 Redli 或 设置 TLS 隧道 才能通过 TLS 连接到托管数据库。创建和管理字符串包含字符串的键只能包含一个值，您无法在一个键中存储多个字符串。但是，Redis 中的字符串是二进制安全的（binary-safe），这意味着 Redis 字符串可以保存任何类型的数据，从字母数字字符到 JPEG 图像均可。唯一的限制是字符串长度不能超过 512 MB。要创建一个字符串，请使用 set 命令。例如，以下 set 命令创建一个名为 key_Welcome1 的键，该键包含字符串 "Howdy"：set key_Welcome1 "Howdy"OK要在一个命令中设置多个字符串，请使用 mset：mset key_Welcome2 "there" key_Welcome3 "partners,"您还可以使用 append 命令创建字符串：append key_Welcome4 "welcome to Texas"若字符串创建成功，append 将输出一个整数，表示该字符串包含的字符数：16注意，append 也可以用来更改字符串的内容。有关详细信息，请参见下文 操作字符串 部分。检索字符串要检索字符串，请使用 get 命令：get key_Welcome1Howdy要使用一个命令检索多个字符串，请使用 mget：mget key_Welcome1 key_Welcome2 key_Welcome3 key_Welcome41) "Howdy" 2) "there" 3) "partners," 4) "welcome to Texas"对于传递给 mget 的任意键，若其不包含字符串值或根本不存在，该命令将返回 nil。操作字符串如果字符串由整数组成，则可以运行 incr 命令将其值增加一：set key_1 3 incr key_14同样，您可以使用 incrby 命令将数字字符串的值增加特定的增量：incrby key_1 1620decr 和 decrby 命令的工作方式相同，但它们会减少存储在数字字符串中的整数：decr key_119decrby key_1 163若字符串键已存在，append 会将值附加到现有内容的末尾，并返回新字符串的长度。为了说明这一点，以下命令将 ", y'all" 附加到键 key_Welcome4 所保存的字符串上，因此现在该字符串将显示为 "welcome to Texas, y'all"：append key_Welcome4 ", y'all"15您也可以将整数附加到包含数字值的字符串中。以下示例将 45 附加到包含整数 3 的 key_1，因此它将保持为 345。在这种情况下，append 还将返回字符串的新长度，而不是其新值：append key_1 453(注：此处原文输出为 3，可能是指长度变化或特定上下文，实际 345 长度为 3。.html)由于此键仍仅保留数字值，因此可以对其执行 incr 和 decr 操作。您可以在整数字符串后附加字母字符，但如果这样做，则在字符串上运行 incr 和 decr 将产生错误，因为字符串值不再是整数。结论本指南详细介绍了许多用于在 Redis 中创建和管理字符串的命令。如果您想在本指南中概述其他相关的命令、参数或过程，请在下面的评论中提出疑问或提出建议。有关 Redis 命令的更多信息，请参阅关于 如何管理 Redis 数据库的 系列教程。说明版本时效性说明：本文内容基于 Redis 4.0.9 及 Ubuntu 18.04 环境测试。虽然字符串基础命令在后续版本中保持高度兼容，但部分托管数据库服务或新版本 Redis 可能在安全配置（如 TLS 连接）及默认权限上有所调整，请根据实际情况参考官方文档。
系列教程导航如何在 Ubuntu 18.04 上安装和保护 Redis如何连接到 Redis 数据库如何管理 Redis 数据库和 Keys如何在 Redis 中管理副本和客户端如何在 Redis 中管理字符串如何在 Redis 中管理 List如何在 Redis 中管理 Hashes如何在 Redis 中管理 Sets如何在 Redis 中管理 Sorted Sets如何在 Redis 中运行事务如何使 Redis 中的 Key 失效如何解决 Redis 中的故障如何从命令行更改 Redis 的配置Redis 数据类型简介介绍Redis 是一个开源的内存中键值数据存储系统。默认情况下，Redis 的键（Key）是持久化的，这意味着除非手动删除，否则 Redis 服务器将一直存储它们。但在某些场景下，您可能希望键在设定的一段时间后自动删除，即让键具有可变性（Volatile）。本教程将说明如何设置键的过期时间、如何检查键的剩余存活时间，以及如何取消键的过期设置。如何使用本指南本指南以备有完整示例的备忘单形式编写。您可以直接跳转至与您任务相关的任何部分。本指南中显示的命令已在运行 Redis 4.0.9 版本的 Ubuntu 18.04 服务器上进行了测试。要设置类似的环境，您可以按照我们的指南 如何在 Ubuntu 18.04 上安装和保护 Redis 的 步骤 1 进行操作。我们将通过使用 redis-cli（Redis 命令行界面）运行命令来演示其行为。请注意，如果您使用其他 Redis 客户端（例如 Redli），某些命令的确切输出可能会有所不同。另外，您也可以使用托管的 Redis 数据库实例来测试这些命令。但请注意，根据数据库提供者允许的控制级别，本指南中的某些命令可能无法按所述方式工作。要配置 DigitalOcean 托管数据库，请遵循我们的 托管数据库产品文档。然后，您必须 安装 Redli 或 设置 TLS 隧道 才能通过 TLS 连接到托管数据库。设置键的过期时间您可以使用 expire 命令设置现有键的过期时间。该命令接受键名和直到过期的秒数作为参数。为了证明这一点，请运行以下两个命令。第一条命令创建一个名为 key_melon 的字符串键，值为 "cantaloupe"；第二条命令将其设置为在 450 秒后超时：set key_melon "cantaloupe" expire key_melon 450如果成功设置了超时，expire 命令将返回 (integer.html) 1。如果设置超时失败（例如键不存在），它将返回 (integer.html) 0。或者，您可以使用 expireat 命令将键设置为在特定的时间点过期。它使用 Unix 时间戳 作为参数，而不是到期前的秒数。Unix 时间戳是自 _Unix 纪元_（即 1970 年 1 月 1 日 UTC 00:00:00）以来的秒数。可以使用许多在线工具来查找特定日期和时间的 Unix 时间戳，例如 EpochConverter 或 UnixTimestamp.com。例如，要设置 key_melon 为在 2025 年 5 月 1 日格林尼治标准时间晚上 8:30 到期（对应的 Unix 时间戳为 1746131400），可以使用以下命令：expireat key_melon 1746131400请注意，如果传递给 expireat 的时间戳已经发生，它将立即删除该键。检查键的剩余过期时间在任何时候，您都可以检查键距离过期还有多少时间（以秒为单位）。可以使用 ttl 命令，其全称为 "Time To Live"：ttl key_melon输出示例：(integer.html) 433有关更详细的信息，可以运行 pttl 命令，它将返回直到键过期的时间（以毫秒为单位）：pttl key_melon输出示例：(integer.html) 431506如果键尚未设置为过期，ttl 和 pttl 都将返回 (integer.html) -1。如果键不存在，两者都将返回 (integer.html) -2。使键持久化如果键已设置为过期，任何覆盖键内容的命令（例如 set 或 getset）都会清除键的超时值。若要手动清除键的超时设置使其永久保存，请使用 persist 命令：persist key_melon如果成功完成，persist 命令将返回 (integer.html) 1，表明该键将不再过期。结论本指南详细介绍了用于在 Redis 中操作和检查键持久性的许多命令。如果您想在本指南中概述其他相关的命令、参数或过程，请在下面的评论中提出疑问或提出建议。有关 Redis 命令的更多信息，请参阅关于 如何管理 Redis 数据库的 系列教程。说明：本文内容基于 Redis 4.0.9 版本及 Ubuntu 18.04 环境测试。不同版本的 Redis 可能在命令行为或输出格式上略有差异，请以官方文档为准。
本系列教程如何在 ubuntu18.04 上安装和保护 redis如何连接到 Redis 数据库如何管理 Redis 数据库和 Keys如何在 Redis 中管理副本和客户端如何在 Redis 中管理字符串如何在 Redis 中管理 list如何在 Redis 中管理 Hashes如何在 Redis 中管理 Sets如何在 Redis 中管理 Sorted Sets如何在 Redis 中运行事务如何使 Redis 中的 Key 失效如何解决 Redis 中的故障如何从命令行更改 Redis 的配置Redis 数据类型简介在 Redis 中怎么使用 SetsRedis 是一个开源的内存中键值数据存储。集合 (Sets.html) 是存储在 Redis 给定键 (Key.html) 下的字符串集合。当保存在集合中时，单个记录值称为_成员 (Member.html)_。与列表 (List.html) 不同，集合是无序的，并且不允许重复的值。本教程说明了如何创建集合、检索和删除成员，以及比较不同集合的成员。如何使用本指南本指南以备有完整示例的速查表形式编写。我们鼓励您跳至与您要完成的任务相关的任何部分。本指南中显示的命令已在运行 Redis 版本 4.0.9 的 Ubuntu 18.04 服务器上进行了测试。要设置类似的环境，您可以按照我们的指南 如何在 Ubuntu 18.04 上安装和保护 Redis 的步骤 1进行操作。我们将通过使用 redis-cli 运行它们来演示这些命令的行为。请注意，如果您使用其他 Redis 界面（例如 Redli），则某些命令的确切输出可能会有所不同。另外，您可以提供一个托管的 Redis 数据库实例来测试这些命令，但是请注意，根据数据库提供者所允许的控制级别，本指南中的某些命令可能无法按所述方式工作。要配置 DigitalOcean 托管数据库，请遵循我们的 托管数据库产品文档。然后，您必须 安装 Redli 或 设置 TLS 隧道 才能通过 TLS 连接到托管数据库。创建集合 (Creating Sets.html)sadd 命令允许您创建一个集合并向其中添加一个或多个成员。下面的例子将创建一组在一个名为键 key_horror 与成员 "Frankenstein" 和 "Godzilla"：sadd key_horror "Frankenstein" "Godzilla"如果成功，sadd 将返回一个整数，显示它添加到集合中的成员数量：Output (integer.html) 2如果您尝试将集合的成员添加到已经持有非集合值的键中，它将返回错误。在此块中的第一个命令创建一个 列表 命名的 key_action 一个元素，"Shaft"。下一条命令尝试将 set 成员添加 "Shane" 到列表中，但是由于数据类型冲突而产生错误：rpush key_action "Shaft" sadd key_action "Shane"Output (error.html) WRONGTYPE Operation against a key holding the wrong kind of value请注意，集合不允许同一成员出现多次：sadd key_comedy "It's" "A" "Mad" "Mad" "Mad" "Mad" "Mad" "World"Output (integer.html) 4即使此 sadd 命令指定了八个成员，它也会丢弃四个重复的 "Mad" 成员，从而将大小设置为 4。从集合中检索成员 (Retrieving Members from Sets.html)在本节中，我们将介绍一些 Redis 命令，这些命令返回有关集合中持有的成员的信息。要练习此处概述的命令，请运行以下命令，这将创建一个由六个成员组成的集合，该集合的键为 key_stooges：sadd key_stooges "Moe" "Larry" "Curly" "Shemp" "Joe" "Curly Joe"要从集合中返回每个成员，请运行 smembers 命令，然后输入要检查的密钥：smembers key_stoogesOutput 1) "Curly" 2) "Moe" 3) "Larry" 4) "Shemp" 5) "Curly Joe" 6) "Joe"要检查特定值是否是集合的成员，请使用以下 sismember 命令：sismember key_stooges "Harpo"如果元素 "Harpo" 是 key_stooges 集合的成员，sismember 则将返回 1。否则，它将返回 0：Output (integer.html) 0要查看给定集合中有多少成员（换句话说，找到给定集合的_基数 (Cardinality.html)_），请运行 scard：scard key_stoogesOutput (integer.html) 6要从集合中返回随机元素，请运行 srandmember：srandmember key_stoogesOutput "Larry"要从集合中返回多个随机，不同的元素，可以在 srandmember 命令后加上要检索的元素数量：srandmember key_stooges 3Output 1) "Larry" 2) "Moe" 3) "Curly Joe"如果您向传递一个负数 srandmember，则该命令可以多次返回相同的元素：srandmember key_stooges -3Output 1) "Shemp" 2) "Curly Joe" 3) "Curly Joe"srandmember 尽管所使用的随机元素函数的性能在较大的数据集中有所改善，但它并不是完全随机的。有关更多详细信息，请参见 命令的官方文档。从集合中删除成员Redis 带有用于从一组删除成员三个命令：spop，srem，和 smove。spop 从中随机选择指定数量的成员并返回它们，类似于 srandmember，但是从集合中删除它们。它接受包含集合的键名和要从集合中删除的成员数作为参数。如果您未指定数字，spop 则默认为返回并删除单个值。以下示例命令将从 key_stooges 上一节创建的集合中删除并返回两个随机选择的元素：spop key_stooges 2Output 1) "Shemp" 2) "Larry"srem 允许您从集合中删除一个或多个特定成员，而不是随机成员：srem key_stooges "Joe" "Curly Joe"而不是返回从集合中删除的成员，而是 srem 返回一个整数，显示已删除的成员数：Output (integer.html) 2使用 smove 一个成员从一组移动到另一个。此命令以该顺序接受源集，目标集和要移动的成员作为参数。请注意，一次 smove 只能移动一位成员：smove key_stooges key_jambands "Moe"如果命令成功移动了成员，它将返回 (integer.html) 1：Output (integer.html) 1如果 smove 失败，它将返回 (integer.html) 0。请注意，如果目标键尚不存在，请 smove 在将成员移入之前创建它。比较集合Redis 还提供了许多命令，可以找到集合之间的差异和相似性。为了演示如何工作的，这部分将引用一个名为三套 presidents，kings 和 beatles。如果您想亲自尝试本节中的命令，请创建这些集合并使用以下 sadd 命令填充它们：sadd presidents "George" "John" "Thomas" "James" sadd kings "Edward" "Henry" "John" "James" "George" sadd beatles "John" "George" "Paul" "Ringo"sinter 比较不同的集合并返回_集合相交 (Intersection.html)_ 或出现在每个集合中的值：sinter presidents kings beatlesOutput 1) "John" 2) "George"sinterstore 执行类似的功能，但不是返回相交成员，而是在包含这些相交成员的指定目的地创建了一个新集合。请注意，如果目的地已经存在，sinterstore 将覆盖其内容：sinterstore new_set presidents kings beatles smembers new_setOutput 1) "John" 2) "George"sdiff 返回_集合差异 (Difference.html)_ — 由以下每个集合的第一个指定集合的​​差异得出的成员：sdiff presidents kings beatlesOutput 1) "Thomas"换句话说，sdiff 查看第一个给定集合中的每个成员，然后将其与每个连续集合中的成员进行比较。第一组中也出现在以下组中的任何成员都将被删除，并 sdiff 返回其余成员。可以将其视为从第一组中删除后续组的成员。sdiffstore 执行与相似的功能 sdiff，但不返回集合差，而是在给定的目的地创建一个包含集合差的新集合：sdiffstore new_set beatles kings presidents smembers new_setOutput 1) "Paul" 2) "Ringo"像一样 sinterstore，sdiffstore 将覆盖目标键（如果已存在）。sunion 返回_集合并集 (Union.html)_ 或包含您指定的每个集合的每个成员的集合：sunion presidents kings beatlesOutput 1) "Thomas" 2) "George" 3) "Paul" 4) "Henry" 5) "James" 6) "Edward" 7) "John" 8) "Ringo"sunion 将结果视为新集，因为它只允许出现任何给定成员。sunionstore 执行类似的功能，但是在给定的目的地创建一个包含集合并集的新集合，而不仅仅是返回结果：sunionstore new_set presidents kings beatlesOutput (integer.html) 8与 sinterstore 和一样 sdiffstore，sunionstore 如果目标键已经存在，它将覆盖目标键。结论本指南详细介绍了用于在 Redis 中创建和管理集的许多命令。如果您想在本指南中概述其他相关的命令，参数或过程，请在下面的评论中提出疑问或提出建议。有关 Redis 命令的更多信息，请参阅关于 如何管理 Redis 数据库的 系列教程。说明本教程基于 Redis 4.0.9 版本编写，部分命令行为在新版本中可能保持一致，但建议参考官方最新文档以获取更新特性。
Redis 通讯协议 (RESP.html)Redis 的通信协议全称为 Redis Serialization Protocol（简称 RESP）。该协议是二进制安全（binary-safe）的，具有以下显著特性：实现简单、快速解析以及可读性好。RESP 是 Redis 客户端与服务端之间进行通信的标准协议。客户端向服务端发送命令请求，服务端则根据命令类型返回相应的数据。需要注意的是，协议中的每一部分均以回车换行符 \r\n 结尾。命令协议示例以下是一个 SET 命令的原始协议格式示例。客户端发送三条数据（命令名、键、值），服务端回复操作结果。*3\r\n // 消息一共有三行（3 个参数） $3\r\n // 第一行字节数为 3 set\r\n // 第一行的消息内容 $3\r\n // 第二行字节数为 3 abc\r\n // 第二行的消息内容 $6\r\n // 第三行字节数为 6 123456\r\n // 第三行的消息内容 +OK\r\n // 服务端回复：操作成功客户端交互演示在实际使用中，我们可以通过 redis-cli 或 telnet 工具观察客户端与服务端的交互过程。1. 使用 redis-clidaichangyadeMacBook-Pro:~ daichangya$ redis-cli -h localhost -p 6379 localhost:6379> set abc 123456 OK localhost:6379> get abc "123456" localhost:6379> quit2. 使用 telnet 查看原始协议通过 telnet 可以直接看到底层的协议交互内容（注意输入需符合 RESP 格式）：daichangyadeMacBook-Pro:~ daichangya$ telnet localhost 6379 Trying ::1... Connected to localhost. Escape character is '^]'. *2 $3 get $3 abc $6 123456 quit +OK Connection closed by foreign host.说明本文描述的为经典 RESP 协议（通常指 RESP2），该版本在 Redis 6.0 之前是默认协议。Redis 6.0 虽引入了 RESP3，但 RESP2 至今仍被广泛支持且兼容大多数客户端。
《Redis 设计与实现》笔记 - Cluster 节点关于 Redis 的集群，官网介绍比较全面，参考：https://redis.io/topics/cluster-spec。设计目标高性能与线性可扩展性（High performance and linear scalability）支持高达 1000 个节点的线性扩展。没有代理（proxies），使用异步复制，并且在进行赋值时没有合并操作。可接受程度的写安全（Acceptable degree of write safety）系统努力（使用最优的方式）保持来自与大多数 master 节点建立连接的客户端的写操作。通常存在小窗口，其中确认的写操作可能会丢失。当客户端在一个小的分区中时，丢失写操作的窗口会更大。可用性（Availability）Redis 集群支持网络分区——其中大部分主节点都可访问，并且不可访问的各 master 节点对应的从节点中至少有一个可访问。采用副本迁移（replicas migration）：有多个从节点的主节点会提供一个从节点给没有从节点的主节点。正是 Redis Cluster 作为 Redis 的分布式解决方案，有效解决了 Redis 在分布式方面的需求。当遇到单机内存、并发、流量等瓶颈时，可以采用 Cluster 架构达到负载均衡的目的。背景知识点：数据分布理论分布式数据库首要解决把整个数据集按照分区规则映射到多个节点的问题，即把数据集划分到多个节点上，每个节点负责整个数据的一个子集。所谓的数据分区就是将一个较大的数据集分布在不同的节点上进行储存。常见的数据分区方式包括：节点取余原理：根据 key 的 hash 值和节点数取模的方式计算出节点 ID，然后向对应的节点提交数据。缺点：节点的增删会造成大量的数据迁移。一致性哈希原理：对于任何的哈希函数，都有其取值范围。我们可以用环形结构来标识范围。通过哈希函数，每个节点都会被分配到环上的一个位置，每个键值也会被映射到环上的一个位置，然后顺时针找到相邻的节点。缺点：节点调整会造成数据分布不均等的问题。虚拟槽（Virtual Slot）原理：虚拟槽分区巧妙地使用了哈希空间，使用分散度良好的哈希函数把所有的数据映射到一个固定范围内的整数集合，整数定义为槽（slot）。比如 Redis Cluster 槽的范围是 0 ～ 16383。优势：槽是集群内数据管理和迁移的基本单位。采用大范围的槽的主要目的是为了方便数据的拆分和集群的扩展，每个节点负责一定数量的槽。节点 (Node.html)在单机版的 Redis 中，每个 Master 之间是没有任何通信的。要组建一个真正可工作的集群，我们必须将各个独立的节点连接起来，构成一个包含多个节点的集群。连接各个节点的工作可以使用 CLUSTER MEET 命令来完成，该命令的格式如下：CLUSTER MEET <ip> <port>向一个节点 node 发送 CLUSTER MEET 命令，可以让 node 节点与 ip 和 port 所指定的节点进行握手（handshake）。当握手成功时，node 节点就会将 ip 和 port 所指定的节点添加到当前所在的集群中。2.1 启动节点一个节点就是一个运行在集群模式下的 Redis 服务器。Redis 服务器在启动时会根据 cluster_enabled 配置选项是否为 yes 来决定是否开启服务器的集群模式。.... // 集群开启，则初始化 cluster if (server.cluster_enabled.html) clusterInit(); ...// 初始化集群 void clusterInit(void.html) { int saveconf = 0; // 初始化配置 server.cluster = zmalloc(sizeof(clusterState.html)); server.cluster->myself = NULL; server.cluster->currentEpoch = 0; // 初始为 FAIL 状态 server.cluster->state = CLUSTER_FAIL; // master 节点数 server.cluster->size = 1; server.cluster->todo_before_sleep = 0; server.cluster->nodes = dictCreate(&clusterNodesDictType,NULL.html); server.cluster->nodes_black_list = dictCreate(&clusterNodesBlackListDictType,NULL.html); server.cluster->failover_auth_time = 0; server.cluster->failover_auth_count = 0; server.cluster->failover_auth_rank = 0; server.cluster->failover_auth_epoch = 0; server.cluster->cant_failover_reason = CLUSTER_CANT_FAILOVER_NONE; server.cluster->lastVoteEpoch = 0; server.cluster->stats_bus_messages_sent = 0; server.cluster->stats_bus_messages_received = 0; memset(server.cluster->slots,0, sizeof(server.cluster->slots.html)); clusterCloseAllSlots(); /* Lock the cluster config file to make sure every node uses * its own nodes.conf. */ // 配置文件上锁 if (clusterLockConfig(server.cluster_configfile.html) == C_ERR) exit(1.html); /* Load or create a new nodes configuration. */ // 载入或创建一个新的节点配置文件 if (clusterLoadConfig(server.cluster_configfile.html) == C_ERR) { /* No configuration found. We will just use the random name provided * by the createClusterNode() function. */ // 没找到配置文件，随机创建一个集群节点 myself = server.cluster->myself = createClusterNode(NULL,CLUSTER_NODE_MYSELF|CLUSTER_NODE_MASTER.html); serverLog(LL_NOTICE,"No cluster configuration found, I'm %.40s", myself->name); // 添加到当前集群节点的配置中 (// 该 nodes 维护的是一张 nodeName -> node 的 hash 表.html) clusterAddNode(myself.html); saveconf = 1; } // 写配置文件 if (saveconf.html) clusterSaveConfigOrDie(1.html); /* We need a listening TCP port for our cluster messaging needs. */ server.cfd_count = 0; /* Port sanity check II * The other handshake port check is triggered too late to stop * us from trying to use a too-high cluster port number. */ // 检查端口号是否合法 (<=65535.html) if (server.port > (65535-CLUSTER_PORT_INCR.html)) { serverLog(LL_WARNING, "Redis port number too high. " "Cluster communication port is 10,000 port " "numbers higher than your Redis port. " "Your Redis port number must be " "lower than 55535."); exit(1.html); } // 将该集群节点的端口和 fd 绑定 ( 打开 cluster 通道的 非阻塞监听端口.html) if (listenToPort(server.port+CLUSTER_PORT_INCR, server.cfd,&server.cfd_count) == C_ERR) { exit(1.html); } else { int j; for (j = 0; j < server.cfd_count; j++.html) { // 为所有集群的 fd 设置可读事件的处理函数 clusterAcceptHandler // 可以根据系统平台选择合适的 事件模型 (如：Linux 上的 epoll，具体查看 aeApiAddEvent 源码.html) if (aeCreateFileEvent(server.el, server.cfd[j], AE_READABLE, clusterAcceptHandler, NULL) == AE_ERR) serverPanic("Unrecoverable error creating Redis Cluster " "file event."); } } /* The slots -> keys map is a sorted set. Init it. */ // 创建槽映射到键的有序集合 server.cluster->slots_to_keys = zslCreate(); /* Set myself->port to my listening port, we'll just need to discover * the IP address via MEET messages. */ // 设置集群端口 myself->port = server.port; // 没有正在进行手动的故障转移 server.cluster->mf_end = 0; // 重置与手动故障转移的状态 resetManualFailover(); }从该部分源码可以看出，Redis Cluster 初始化部分的核心主要是 nodes.conf 的加载，以及 cluster bus 通道的监听服务的启动这两部分。节点会继续使用 redisServer 结构来保存服务器的状态，使用 redisClient 结构来保存客户端的状态。至于那些只有在集群模式下才会用到的数据，节点将它们保存到了 cluster.h/clusterNode 结构、cluster.h/clusterLink 结构，以及 cluster.h/clusterState 结构里面。下面分别介绍。2.2 集群数据结构clusterNode 结构保存了一个节点的当前状态，比如节点的创建时间、节点的名字、节点当前的配置纪元、节点的 IP 地址和端口号等等。typedef struct clusterNode { // 创建节点的时间 mstime_t ctime; /* Node object creation time. */ // 节点的名字，由 40 个十六进制字符组成 char name[CLUSTER_NAMELEN]; /* Node name, hex string, sha1-size */ // 节点标识 // 使用各种不同的标识值记录节点的角色（比如主节点或者从节点）， // 以及节点目前所处的状态（比如在线或者下线）。 int flags; /* CLUSTER_NODE_... */ // 节点当前的配置纪元，用于实现故障转移 uint64_t configEpoch; /* Last configEpoch observed for this node */ // 由这个节点负责处理的槽 //(一共有 REDIS_CLUSTER_SLOTS / 8 个字节长，每个字节的每个位记录了一个槽的保存状态位，值为 1 表示槽正由本节点处理，值为 0 则表示槽并非本节点处理.html) unsigned char slots[CLUSTER_SLOTS/8]; /* slots handled by this node */ // 该节点负责处理的槽数量 int numslots; /* Number of slots handled by this node */ // 如果本节点是主节点，那么用这个属性记录从节点的数量 int numslaves; /* Number of slave nodes, if this is a master */ // 指针数组，指向各个从节点 struct clusterNode **slaves; /* pointers to slave nodes */ // 指向主节点，即使是从节点也可以为 NULL struct clusterNode *slaveof; /* pointer to the master node. Note that it may be NULL even if the node is a slave if we don't have the master node in our tables. */ // 最后一次发送 PING 命令的时间 mstime_t ping_sent; /* Unix time we sent latest ping */ // 接收到 PONG 的时间 mstime_t pong_received; /* Unix time we received the pong */ // 最后一次被设置为 FAIL 状态的时间 mstime_t fail_time; /* Unix time when FAIL flag was set */ // 最后一次给某个从节点投票的时间 mstime_t voted_time; /* Last time we voted for a slave of this master */ // 最后一次从这个节点接收到复制偏移量的时间 mstime_t repl_offset_time; /* Unix time we received offset for this node */ // 孤立的主节点迁移的时间 mstime_t orphaned_time; /* Starting time of orphaned master condition */ // 该节点已知的复制偏移量 long long repl_offset; /* Last known repl offset for this node. */ // 节点的 IP 地址 char ip[NET_IP_STR_LEN]; /* Latest known IP address of this node */ // 节点的端口号 int port; /* Latest known port of this node */ // 保存连接节点所需的有关信息 clusterLink *link; /* TCP/IP link with this node */ // 保存下线报告的链表 list *fail_reports; /* List of nodes signaling this as failing */ } clusterNode;clusterNode 结构的 link 属性是一个 clusterLink 结构，该结构保存了连接节点所需的有关信息，比如套接字描述符，输入缓冲区和输出缓冲区：/* clusterLink encapsulates everything needed to talk with a remote node. */ // clusterLink 包含了与其他节点进行通讯所需的全部信息 typedef struct clusterLink { // 连接创建的时间 mstime_t ctime; /* Link creation time */ // TCP 套接字描述符 int fd; /* TCP socket file descriptor */ // 输出缓冲区，保存着等待发送给其他节点的消息（message）。 sds sndbuf; /* Packet send buffer */ // 输入缓冲区，保存着从其他节点接收到的消息。 sds rcvbuf; /* Packet reception buffer */ // 与这个连接相关联的节点，如果没有的话就为 NULL struct clusterNode *node; /* Node related to this link if any, or NULL */ } clusterLink;redisClient 结构和 clusterLink 结构都有自己的套接字描述符和输入、输出缓冲区。这两个结构的区别在于：redisClient 结构中的套接字和缓冲区是用于连接客户端的，而 clusterLink 结构中的套接字和缓冲区则是用于连接节点的。最后，每个节点都保存着一个 clusterState 结构，这个结构记录了在当前节点的视角下，集群目前所处的状态。typedef struct clusterState { // 指向当前节点的指针 clusterNode *myself; /* This node */ // 集群当前的配置纪元，用于实现故障转移 uint64_t currentEpoch; // 集群当前的状态：是在线还是下线 int state; /* CLUSTER_OK, CLUSTER_FAIL, ... */ // 集群中至少负责一个槽的主节点个数 int size; /* Num of master nodes with at least one slot */ // 集群节点名单（包括 myself 节点） // 字典的键为节点的名字，字典的值为 clusterNode 结构 dict *nodes; /* Hash table of name -> clusterNode structures */ // 防止重复添加节点的黑名单 dict *nodes_black_list; /* Nodes we don't re-add for a few seconds. */ // 导出槽数据到目标节点，该数组记录这些节点 clusterNode *migrating_slots_to[CLUSTER_SLOTS]; // 导入槽数据到目标节点，该数组记录这些节点 clusterNode *importing_slots_from[CLUSTER_SLOTS]; // 槽和负责槽节点的映射 (负责处理各个槽的节点.html) // 例如 slots[i] = clusterNode_A 表示槽 i 由节点 A 处理 clusterNode *slots[CLUSTER_SLOTS]; // 槽映射到键的有序集合 ( 跳表中以槽作为分值，键作为成员，对槽进行有序排序，当有需要对槽进行区间 range 操作时，会很方便 .html) zskiplist *slots_to_keys; /* The following fields are used to take the slave state on elections. */ // 以下这些域被用于进行故障转移选举 // 上次执行选举或者下次执行选举的时间 mstime_t failover_auth_time; /* Time of previous or next election. */ // 节点获得的投票数量 int failover_auth_count; /* Number of votes received so far. */ // 如果为真，表示本节点已经向其他节点发送了投票请求 int failover_auth_sent; /* True if we already asked for votes. */ // 该从节点在当前请求中的排名 int failover_auth_rank; /* This slave rank for current auth request. */ // 当前选举的纪元 uint64_t failover_auth_epoch; /* Epoch of the current election. */ // 从节点不能执行故障转移的原因 int cant_failover_reason; /* Why a slave is currently not able to failover. See the CANT_FAILOVER_* macros. */ /* Manual failover state in common. */ /* 共用的手动故障转移状态 */ // 如果为 0，表示没有正在进行手动的故障转移。否则表示手动故障转移的时间限制 mstime_t mf_end; /* Manual failover time limit (ms unixtime.html). It is zero if there is no MF in progress. */ /* Manual failover state of master. */ /* 主服务器的手动故障转移状态 */ clusterNode *mf_slave; /* Slave performing the manual failover. */ /* Manual failover state of slave. */ // 从节点记录手动故障转移时的主节点偏移量 long long mf_master_offset; /* Master offset the slave needs to start MF or zero if stil not received. */ // 非零值表示手动故障转移能开始 int mf_can_start; /* If non-zero signal that the manual failover can start requesting masters vote. */ /* The followign fields are used by masters to take state on elections. */ /* 以下这些域由主服务器使用，用于记录选举时的状态 */ // 集群最近一次投票的纪元 uint64_t lastVoteEpoch; /* Epoch of the last vote granted. */ // 调用 clusterBeforeSleep() 所做的一些事，以各个 flag 来记录 int todo_before_sleep; /* Things to do in clusterBeforeSleep(). */ // 通过 cluster 连接发送的消息数量 long long stats_bus_messages_sent; /* Num of msg sent via cluster bus. */ // 通过 cluster 接收到的消息数量 long long stats_bus_messages_received; /* Num of msg rcvd via cluster bus.*/ } clusterState;2.3 CLUSTER MEET 命令的实现通过向节点 A 发送 CLUSTER MEET 命令，客户端可以让接收命令的节点 A 将另一个节点 B 添加到节点 A 当前所在的集群里面：CLUSTER MEET <ip> <port>收到命令的节点 A 将与节点 B 进行握手（handshake），以此来确认彼此的存在，并为将来的进一步通信打好基础：节点 A 会为节点 B 创建一个 clusterNode 结构，并将该结构添加到自己的 clusterState.nodes 字典里面。之后，节点 A 将根据 CLUSTER MEET 命令给定的 IP 地址和端口号，向节点 B 发送一条 MEET 消息（message）。如果一切顺利，节点 B 将接收到节点 A 发送的 MEET 消息，节点 B 会为节点 A 创建一个 clusterNode 结构，并将该结构添加到自己的 clusterState.nodes 字典里面。之后，节点 B 将向节点 A 返回一条 PONG 消息。如果一切顺利，节点 A 将接收到节点 B 返回的 PONG 消息，通过这条 PONG 消息节点 A 可以知道节点 B 已经成功地接收了自己发送的 MEET 消息。之后，节点 A 将向节点 B 返回一条 PING 消息。如果一切顺利，节点 B 将接收到节点 A 返回的 PING 消息，通过这条 PING 消息节点 B 可以知道节点 A 已经成功地接收到了自己返回的 PONG 消息，握手完成。之后，节点 A 会将节点 B 的信息通过 Gossip 协议 传播给集群中的其他节点，让其他节点也与节点 B 进行握手。最终，经过一段时间之后，节点 B 会被集群中的所有节点认识。上面是书上介绍的流程，下面看下源码的实现。2.3.1 A 节点发送 MEET 给 B 节点由客户端发起命令：cluster meet <ip> <port>。当节点接收到客户端的 cluster meet 命令后会调用对应的函数来处理命令，该命令的执行函数是 clusterCommand() 函数。// CLUSTER 命令的实现 void clusterCommand(client *c.html) { // 不能在非集群模式下使用该命令 if (server.cluster_enabled == 0.html) { addReplyError(c,"This instance has cluster support disabled".html); return; } // CLUSTER MEET <ip> <port> 命令 // 与给定地址的节点建立连接 if (!strcasecmp(c->argv[1]->ptr,"meet".html) && c->argc == 4) { long long port; // 获取端口 if (getLongLongFromObject(c->argv[3], &port.html) != C_OK) { addReplyErrorFormat(c,"Invalid TCP port specified: %s", (char*.html)c->argv[3]->ptr); return; } // 尝试与给定地址的节点进行连接 if (clusterStartHandshake(c->argv[2]->ptr,port.html) == 0 && errno == EINVAL) { // 连接失败 addReplyErrorFormat(c,"Invalid node address specified: %s:%s", (char*.html)c->argv[2]->ptr, (char*.html)c->argv[3]->ptr); } else { // 连接成功回复 ok addReply(c,shared.ok); } .... }该函数先根据 cluster meet <ip> <port> 命令传入的参数，获取要与目标节点建立连接的节点地址，然后根据节点地址执行 clusterStartHandshake() 函数来开始执行握手操作。该函数代码如下：/* Start an handshake with the specified address if there is not one * already in progress. Returns non-zero if the handshake was actually * started. On error zero is returned and errno is set to one of the * following values: * 如果还没有与指定的地址进行过握手，那么进行握手。 * 返回 1 表示握手已经开始， * 返回 0 并将 errno 设置为以下值来表示意外情况： * EAGAIN - There is already an handshake in progress for this address. 已经有握手在进行中了。 * EINVAL - IP or port are not valid. ip 或者 port 参数不合法。 */ int clusterStartHandshake(char *ip, int port.html) { clusterNode *n; char norm_ip[NET_IP_STR_LEN]; struct sockaddr_storage sa; /* IP sanity check */ // 检查地址是否非法 if (inet_pton(AF_INET,ip, &(((struct sockaddr_in *.html)&sa)->sin_addr))) { sa.ss_family = AF_INET; } else if (inet_pton(AF_INET6,ip, &(((struct sockaddr_in6 *.html)&sa)->sin6_addr))) { sa.ss_family = AF_INET6; } else { errno = EINVAL; return 0; } /* Port sanity check */ // port 合法性检查 if (port <= 0 || port > (65535-CLUSTER_PORT_INCR.html)) { errno = EINVAL; return 0; } /* Set norm_ip as the normalized string representation of the node * IP address. */ // 设置 norm_ip 作为节点地址的标准字符串表示形式 memset(norm_ip,0,NET_IP_STR_LEN.html); if (sa.ss_family == AF_INET.html) inet_ntop(AF_INET, (void*.html)&(((struct sockaddr_in *.html)&sa)->sin_addr), norm_ip,NET_IP_STR_LEN); else inet_ntop(AF_INET6, (void*.html)&(((struct sockaddr_in6 *.html)&sa)->sin6_addr), norm_ip,NET_IP_STR_LEN); // 判断当前地址是否处于握手状态，如果是，则设置 errno 并返回，该函数被用来避免重复和相同地址的节点进行握手 if (clusterHandshakeInProgress(norm_ip,port.html)) { errno = EAGAIN; return 0; } /* Add the node with a random address (NULL as first argument to * createClusterNode()). Everything will be fixed during the * handshake. */ // 为 node 设置一个随机的名字，当握手完成时会为其设置真正的名字 // 创建一个随机名字的节点 n = createClusterNode(NULL,CLUSTER_NODE_HANDSHAKE|CLUSTER_NODE_MEET.html); // 设置地址 memcpy(n->ip,norm_ip,sizeof(n->ip.html)); n->port = port; // 添加到集群中 clusterAddNode(n.html); return 1; }该函数先判断传入的地址是否非法，如果非法会设置 errno，然后会调用 clusterHandshakeInProgress() 函数来判断是否要进行握手的节点也处于握手状态，以避免重复和相同地址的目标节点进行握手。然后创建一个随机名字的目标节点，并设置该目标节点的状态。然后调用 clusterAddNode() 函数将该目标节点添加到集群中，也就是 server.cluster->nodes 字典，该字典的键是节点的名字，值是指向 clusterNode() 结构的指针。此时 A 节点并没有将 meet 消息发送给指定地址的目标节点，而是设置集群中目标节点的状态。而发送 meet 消息则是在 clusterCron() 函数中执行。/* This is executed 10 times every second */ // 集群的周期性执行函数。每秒执行 10 次，100ms 执行一次 void clusterCron(void.html) { dictIterator *di; dictEntry *de; int update_state = 0; // 没有从节点从属的主节点个数 int orphaned_masters; /* How many masters there are without ok slaves. */ // 所有主节点附属的最多的从节点数量 int max_slaves; /* Max number of ok slaves for a single master. */ // 如果 myself 是从节点，该从节点对应的主节点下有多少从节点 int this_slaves; /* Number of ok slaves for our master (if we are slave.html). */ mstime_t min_pong = 0, now = mstime(); clusterNode *min_pong_node = NULL; // 静态变量，表示该函数执行的计数器 static unsigned long long iteration = 0; mstime_t handshake_timeout; // 记录一次迭代 iteration++; /* Number of times this function was called so far. */ /* The handshake timeout is the time after which a handshake node that was * not turned into a normal node is removed from the nodes. Usually it is * just the NODE_TIMEOUT value, but when NODE_TIMEOUT is too small we use * the value of 1 second. */ // 获取握手状态超时的时间，最低为 1s // 如果一个处于握手状态的节点如果没有在该超时时限内变成一个普通的节点，那么该节点从节点字典中被删除 handshake_timeout = server.cluster_node_timeout; if (handshake_timeout < 1000.html) handshake_timeout = 1000; /* Check if we have disconnected nodes and re-establish the connection. */ // 检查是否当前集群中有断开连接的节点和重新建立连接的节点 di = dictGetSafeIterator(server.cluster->nodes.html); // 遍历所有集群中的节点，如果有未建立连接的节点，那么发送 PING 或 PONG 消息，建立连接 while((de = dictNext(di.html)) != NULL) { clusterNode *node = dictGetVal(de.html); // 跳过当前节点以及没有地址的节点 if (node->flags & (CLUSTER_NODE_MYSELF|CLUSTER_NODE_NOADDR.html)) continue; /* A Node in HANDSHAKE state has a limited lifespan equal to the * configured node timeout. */ // 如果仍然 node 节点处于握手状态，但是从建立连接开始到现在已经超时 if (nodeInHandshake(node.html) && now - node->ctime > handshake_timeout) { // 从集群中删除该节点，遍历下一个节点 clusterDelNode(node.html); continue; } // 为未创建连接的节点创建连接 if (node->link == NULL.html) { int fd; mstime_t old_ping_sent; clusterLink *link; // myself 节点连接这个 node 节点 fd = anetTcpNonBlockBindConnect(server.neterr, node->ip, node->port+CLUSTER_PORT_INCR, NET_FIRST_BIND_ADDR); // 连接出错，跳过该节点 if (fd == -1.html) { /* We got a synchronous error from connect before * clusterSendPing() had a chance to be called. * If node->ping_sent is zero, failure detection can't work, * so we claim we actually sent a ping now (that will * be really sent as soon as the link is obtained). */ // 如果 ping_sent 为 0，察觉故障无法执行，因此要设置发送 PING 的时间，当建立连接后会真正的的发送 PING 命令 if (node->ping_sent == 0.html) node->ping_sent = mstime(); serverLog(LL_DEBUG, "Unable to connect to " "Cluster Node [%s]:%d -> %s", node->ip, node->port+CLUSTER_PORT_INCR, server.neterr); continue; } // 为 node 节点创建一个连接对象 link = createClusterLink(node.html); // 设置连接对象的属性 link->fd = fd; // 为 node 设置连接对象 node->link = link; // 监听该连接的可读事件，设置可读时间的读处理函数 aeCreateFileEvent(server.el,link->fd,AE_READABLE, clusterReadHandler,link); /* Queue a PING in the new connection ASAP: this is crucial * to avoid false positives in failure detection. * * If the node is flagged as MEET, we send a MEET message instead * of a PING one, to force the receiver to add us in its node * table. */ /* 向新连接的节点发送 PING 命令，防止节点被识进入下线*/ // 备份旧的发送 PING 的时间 old_ping_sent = node->ping_sent; // 如果 node 节点指定了 MEET 标识，那么发送 MEET 命令，否则发送 PING 命令 clusterSendPing(link, node->flags & CLUSTER_NODE_MEET ? CLUSTERMSG_TYPE_MEET : CLUSTERMSG_TYPE_PING); // 如果不是第一次发送 PING 命令，要将发送 PING 的时间还原，等待被 clusterSendPing() 更新 if (old_ping_sent.html) { /* If there was an active ping before the link was * disconnected, we want to restore the ping time, otherwise * replaced by the clusterSendPing() call. */ node->ping_sent = old_ping_sent; } /* We can clear the flag after the first packet is sent. * If we'll never receive a PONG, we'll never send new packets * to this node. Instead after the PONG is received and we * are no longer in meet/handshake status, we want to send * normal PING packets. */ // 发送 MEET 消息后，清除 MEET 标识 // 如果没有接收到 PONG 回复，那么不会在向该节点发送消息 // 如果接收到了 PONG 回复，取消 MEET/HANDSHAKE 状态，发送一个正常的 PING 消息 node->flags &= ~CLUSTER_NODE_MEET; serverLog(LL_DEBUG,"Connecting with Node %.40s at %s:%d", node->name, node->ip, node->port+CLUSTER_PORT_INCR); } } dictReleaseIterator(di.html); ... }以 A 节点举例来解释周期性函数中发送 MEET 消息的代码：遍历集群中所有的节点，跳过操作当前 A 节点和没有指定地址的节点，然后判断处于握手状态的节点是否在建立连接的过程中超时，如果超时则会删除该节点。如果还没有创建连接，那么 A 节点会与当前这个目标节点建立 TCP 连接，并获取套接字 fd，根据这个套接字，就可以创建 clusterLink 结构的连接对象，并将这个连接对象保存到当前这个目标节点。A 节点创建完连接后，首先会监听与目标节点建立的 fd 的可读事件，并设置对应的处理程序 clusterReadHandler()，因为当发送 MEET 消息给目标节点后，要接收目标节点回复的 PING。接下来，A 节点就调用 clusterSendPing() 函数发送 MEET 消息给目标节点。MEET 消息是特殊的 PING 消息，只用于通知新节点的加入，而 PING 消息还需要更改一些时间信息，以便进行故障检测。最后无论如何都要取消 CLUSTER_NODE_MEET 标识，但是没有取消 CLUSTER_NODE_HANDSHAKE 该标识，表示仍处于握手状态，但是已经发送了 MEET 消息了。2.3.2 目标节点处理 MEET 消息并回复 PONG 消息当 A 节点将 MEET 消息发送给目标节点 B 之前，就设置了 clusterReadHandler() 函数为处理接收的 PONG 消息。当时目标节点 B 如何接收到 MEET 消息，并且回复 PONG 消息给 A 节点呢？在集群模式下，每个节点初始化时调用的 clusterInit 时，会监听节点的端口等待客户端的连接，并且会将该监听的套接字 fd 保存到 server.cfd 数组中，然后创建文件事件，监听该套接字 fd 的可读事件，并设置可读事件处理函数 clusterAcceptHandler()，等待客户端发送数据。// 监听事件处理器 (集群的 fd 所设置可读事件的处理函数.html) #define MAX_CLUSTER_ACCEPTS_PER_CALL 1000 void clusterAcceptHandler(aeEventLoop *el, int fd, void *privdata, int mask.html) { int cport, cfd; int max = MAX_CLUSTER_ACCEPTS_PER_CALL; char cip[NET_IP_STR_LEN]; clusterLink *link; UNUSED(el.html); UNUSED(mask.html); UNUSED(privdata.html); /* If the server is starting up, don't accept cluster connections: * UPDATE messages may interact with the database content. */ // 如果当前节点正在载入数据，则直接返回。不接收集群的连接 if (server.masterhost == NULL && server.loading) return; // 最大每次调用接收 1000 个连接 while(max--.html) { // TCP 连接的 accept cfd = anetTcpAccept(server.neterr, fd, cip, sizeof(cip.html), &cport); if (cfd == ANET_ERR.html) { if (errno != EWOULDBLOCK.html) serverLog(LL_VERBOSE, "Error accepting cluster node: %s", server.neterr); return; } // 设置 fd 为非阻塞模式 anetNonBlock(NULL,cfd.html); // 禁用 nagle 算法 anetEnableTcpNoDelay(NULL,cfd.html); /* Use non-blocking I/O for cluster messages. */ serverLog(LL_VERBOSE,"Accepted cluster node %s:%d", cip, cport.html); /* Create a link object we use to handle the connection. * It gets passed to the readable handler when data is available. * Initiallly the link->node pointer is set to NULL as we don't know * which node is, but the right node is references once we know the * node identity. */ // 当连接成功后，为其创建一个连接对象，但是不关联连接的节点 link = createClusterLink(NULL.html); link->fd = cfd; // 监听该连接的可读事件，并设置处理函数为 clusterReadHandler aeCreateFileEvent(server.el,cfd,AE_READABLE,clusterReadHandler,link.html); } }clusterAcceptHandler() 函数，该函数实际上就是 accept() 函数，接收 A 节点的连接，然后监听该连接上的可读事件，设置可读事件的处理函数为 clusterReadHandler()，等待 A 节点发送数据。当 A 节点发送 MEET 消息给目标节点时，触发目标节点执行 clusterReadHandler() 函数来处理消息。clusterReadHandler 函数主要工作就是解析 cluster bus 上接收的数据并进行消息分包，然后对消息进行处理。而对于消息的分包首先需要了解一下消息结构，Redis Cluster 节点之间通信的消息结构定义如下：#define CLUSTER_PROTO_VER 0 /* Cluster bus protocol version. */ // 用来表示集群消息的结构（消息头，header） typedef struct { // "RCmb"的签名 char sig[4]; /* Siganture "RCmb" (Redis Cluster message bus.html). */ // 消息的总长 uint32_t totlen; /* Total length of this message */ // 协议版本，当前为 0 uint16_t ver; /* Protocol version, currently set to 0. */ // 未使用的 2 字节 uint16_t notused0; /* 2 bytes not used. */ // 消息类型 uint16_t type; /* Message type */ // 只在发送 PING、PONG 和 MEET 消息时使用 // 消息正文包含的节点信息数量 uint16_t count; /* Only used for some kind of messages. */ // 消息发送者的配置纪元 uint64_t currentEpoch; /* The epoch accordingly to the sending node. */ // 如果消息发送者是一个主节点，那么这里记录的是消息发送者的配置纪元 // 如果消息发送者是一个从节点，那么这里记录的是消息发送者正在复制的主节点的配置纪元 uint64_t configEpoch; /* The config epoch if it's a master, or the last epoch advertised by its master if it is a slave. */ // 节点的复制偏移量 uint64_t offset; /* Master replication offset if node is a master or processed replication offset if node is a slave. */ // 发送消息的节点的 name（ID） char sender[CLUSTER_NAMELEN]; /* Name of the sender node */ // 消息发送者目前的槽指派信息 unsigned char myslots[CLUSTER_SLOTS/8]; // 如果消息发送者是一个从节点，那么这里记录的是消息发送者正在复制的主节点的名字 // 如果消息发送者是一个主节点，那么这里记录的是空 ( REDIS_NODE_NULL_NAME 一个 40 字节长，值全为 0 的字节数组.html) char slaveof[CLUSTER_NAMELEN]; // 32 字节未使用 char notused1[32]; /* 32 bytes reserved for future usage. */ // 发送消息节点的端口 uint16_t port; /* Sender TCP base port */ // 发送消息节点的标识 uint16_t flags; /* Sender node flags */ // 消息发送者所处集群的状态 unsigned char state; /* Cluster state from the POV of the sender */ // 消息的标识 unsigned char mflags[3]; /* Message flags: CLUSTERMSG_FLAG[012]_... */ // 消息的数据 union clusterMsgData data; } clusterMsg;从上面结构可以看到消息分包，主要解析前 8 个字节，分别为：char sig[4]：消息签名，对于 cluster 消息，固定为字符序列 RCmb。uint32_t totlen：消息总长度。其他结构成员都是在处理消息时使用的，后续讲解消息处理流程时进行分析。/* Read data. Try to read the first field of the header first to check the * full length of the packet. When a whole packet is in memory this function * will call the function to process the packet. And so forth. */ // 读事件处理器 // 首先读入内容的头，以判断读入内容的长度 // 如果内容是一个 whole packet，那么调用函数来处理这个 packet。 void clusterReadHandler(aeEventLoop *el, int fd, void *privdata, int mask.html) { char buf[sizeof(clusterMsg.html)]; ssize_t nread; clusterMsg *hdr; clusterLink *link = (clusterLink*.html) privdata; unsigned int readlen, rcvbuflen; UNUSED(el.html); UNUSED(mask.html); // 循环从 fd 读取数据 (尽可能地多读数据.html) while(1.html) { /* Read as long as there is data to read. */ // 获取连接对象的接收缓冲区的长度，表示一次最多能多大的数据量 rcvbuflen = sdslen(link->rcvbuf.html); // 如果接收缓冲区的长度小于 8 字节 (头信息.html)，就无法读入消息的总长 if (rcvbuflen < 8.html) { /* First, obtain the first 8 bytes to get the full message * length. */ readlen = 8 - rcvbuflen; } else {// 能够读入完整数据信息 /* Finally read the full message. */ hdr = (clusterMsg*.html) link->rcvbuf; // 如果是 8 个字节 if (rcvbuflen == 8.html) { /* Perform some sanity check on the message signature * and length. */ // 如果前四个字节不是"RCmb"签名，释放连接 if (memcmp(hdr->sig,"RCmb",4.html) != 0 || ntohl(hdr->totlen.html) < CLUSTERMSG_MIN_LEN) { serverLog(LL_WARNING, "Bad message length or signature received " "from Cluster bus."); handleLinkIOError(link.html); return; } } // 记录已经读入的内容长度 readlen = ntohl(hdr->totlen.html) - rcvbuflen; if (readlen > sizeof(buf.html)) readlen = sizeof(buf.html); } // 读入内容 nread = read(fd,buf,readlen.html); // 没有内容可读 if (nread == -1 && errno == EAGAIN.html) return; /* No more data ready. */ // 读错误，释放连接 if (nread <= 0.html) { /* I/O error... */ serverLog(LL_DEBUG,"I/O error reading from node link: %s", (nread == 0.html) ? "connection closed" : strerror(errno.html)); handleLinkIOError(link.html); return; } else { /* Read data and recast the pointer to the new buffer. */ // 将读到的数据追加到连接对象的接收缓冲区中 link->rcvbuf = sdscatlen(link->rcvbuf,buf,nread.html); hdr = (clusterMsg*.html) link->rcvbuf; rcvbuflen += nread; } /* Total length obtained? Process this packet. */ // 检查已读入内容的长度，看是否整条信息已经被读入了 if (rcvbuflen >= 8 && rcvbuflen == ntohl(hdr->totlen.html)) { // 如果读到的数据有效，处理读到接收缓冲区的数据 if (clusterProcessPacket(link.html)) { // 处理成功，则设置新的空的接收缓冲区 sdsfree(link->rcvbuf.html); link->rcvbuf = sdsempty(); } else { return; /* Link no longer valid. */ } } } }之前在介绍 clusterLink 对象时，每个连接对象都有一个 link->rcvbuf 接收缓冲区和 link->sndbuf 发送缓冲区，因此这个函数就是从 fd 将数据读到 link 的接收缓冲区，然后进行是否读完整的判断。如果完整的读完数据，就调用 clusterProcessPacket() 函数来处理读到的数据，这里会处理 MEET 消息。/* When this function is called, there is a packet to process starting * at node->rcvbuf. Releasing the buffer is up to the caller, so this * function should just handle the higher level stuff of processing the * packet, modifying the cluster state if needed. * * 当这个函数被调用时，说明 node->rcvbuf 中有一条待处理的信息。 * 信息处理完毕之后的释放工作由调用者处理，所以这个函数只需负责处理信息就可以了。 * * The function returns 1 if the link is still valid after the packet * was processed, otherwise 0 if the link was freed since the packet * processing lead to some inconsistency error (for instance a PONG * received from the wrong sender ID). * 如果函数返回 1，那么说明处理信息时没有遇到问题，连接依然可用。 * 如果函数返回 0，那么说明信息处理时遇到了不一致问题 * （比如接收到的 PONG 是发送自不正确的发送者 ID 的），连接已经被释放。 */ int clusterProcessPacket(clusterLink *link.html) { // 连接的输入（接收）缓冲区 clusterMsg *hdr = (clusterMsg*.html) link->rcvbuf; // 消息的总长度 uint32_t totlen = ntohl(hdr->totlen.html); // 消息的类型 uint16_t type = ntohs(hdr->type.html); // 通过 Cluster 接收到的消息数量加 1 server.cluster->stats_bus_messages_received++; serverLog(LL_DEBUG,"--- Processing packet of type %d, %lu bytes", type, (unsigned long.html) totlen); /* Perform sanity checks */ // 检查消息包的合法性 // 至少包含一个签名、版本、总长、消息正文包含的节点信息数量 if (totlen < 16.html) return 1; /* At least signature, version, totlen, count. */ // 总长度大于接收缓冲区的大小 if (totlen > sdslen(link->rcvbuf.html)) return 1; // 目前版本号为 0，不处理其他版本 if (ntohs(hdr->ver.html) != CLUSTER_PROTO_VER) { /* Can't handle messages of different versions. */ return 1; } // 获取发送消息节点的标识 uint16_t flags = ntohs(hdr->flags.html); uint64_t senderCurrentEpoch = 0, senderConfigEpoch = 0; clusterNode *sender; // 如果消息是 PING、PONG 或者 MEET if (type == CLUSTERMSG_TYPE_PING || type == CLUSTERMSG_TYPE_PONG || type == CLUSTERMSG_TYPE_MEET) { // 消息正文包含的节点信息数量 uint16_t count = ntohs(hdr->count.html); uint32_t explen; /* expected length of this packet */ // 计算消息包应该的长度 explen = sizeof(clusterMsg.html)-sizeof(union clusterMsgData.html); explen += (sizeof(clusterMsgDataGossip.html)*count); // 总长度和计算的长度不相同返回 1 if (totlen != explen.html) return 1; // 如果消息是 FAIL } else if (type == CLUSTERMSG_TYPE_FAIL.html) { // 计算消息包应该的长度 uint32_t explen = sizeof(clusterMsg.html)-sizeof(union clusterMsgData.html); explen += sizeof(clusterMsgDataFail.html); // 总长度和计算的长度不相同返回 1 if (totlen != explen.html) return 1; // 如果消息是 PUBLISH } else if (type == CLUSTERMSG_TYPE_PUBLISH.html) { uint32_t explen = sizeof(clusterMsg.html)-sizeof(union clusterMsgData.html); explen += sizeof(clusterMsgDataPublish.html) - 8 + ntohl(hdr->data.publish.msg.channel_len.html) + ntohl(hdr->data.publish.msg.message_len.html); if (totlen != explen.html) return 1; // 如果消息是故障有关的 } else if (type == CLUSTERMSG_TYPE_FAILOVER_AUTH_REQUEST || type == CLUSTERMSG_TYPE_FAILOVER_AUTH_ACK || type == CLUSTERMSG_TYPE_MFSTART) { uint32_t explen = sizeof(clusterMsg.html)-sizeof(union clusterMsgData.html); if (totlen != explen.html) return 1; // 如果消息是 UPDATE 的 } else if (type == CLUSTERMSG_TYPE_UPDATE.html) { uint32_t explen = sizeof(clusterMsg.html)-sizeof(union clusterMsgData.html); explen += sizeof(clusterMsgDataUpdate.html); if (totlen != explen.html) return 1; } /* Check if the sender is a known node. */ // 如果消息是 UPDATE 的 // 从集群中查找 sender 节点 sender = clusterLookupNode(hdr->sender.html); // 节点存在，并且不是 HANDSHAKE 节点 // 那么个更新节点的配置纪元信息 if (sender && !nodeInHandshake(sender.html)) { /* Update our curretEpoch if we see a newer epoch in the cluster. */ // 如果 sender 的纪元大于集群的纪元，更新集群的纪元 senderCurrentEpoch = ntohu64(hdr->currentEpoch.html); senderConfigEpoch = ntohu64(hdr->configEpoch.html); // 更新集群的当前纪元 if (senderCurrentEpoch > server.cluster->currentEpoch.html) server.cluster->currentEpoch = senderCurrentEpoch; /* Update the sender configEpoch if it is publishing a newer one. */ // 更新 sender 的配置纪元 if (senderConfigEpoch > sender->configEpoch.html) { sender->configEpoch = senderConfigEpoch; // 更新配置和状态 clusterDoBeforeSleep(CLUSTER_TODO_SAVE_CONFIG| CLUSTER_TODO_FSYNC_CONFIG); } /* Update the replication offset info for this node. */ // 更新 sender 的复制偏移量和更新复制偏移量的时间 sender->repl_offset = ntohu64(hdr->offset.html); sender->repl_offset_time = mstime(); /* If we are a slave performing a manual failover and our master * sent its offset while already paused, populate the MF state. */ // 如果当前节点是正在执行手动故障转移的从节点，该当前节点的主节点正是 sender 节点 // 并且主节点发送复制偏移量时已经暂停手动故障转移 if (server.cluster->mf_end && nodeIsSlave(myself.html) && myself->slaveof == sender && hdr->mflags[0] & CLUSTERMSG_FLAG0_PAUSED && server.cluster->mf_master_offset == 0) { // 设置当前从节点已经复制的偏移量 server.cluster->mf_master_offset = sender->repl_offset; serverLog(LL_WARNING, "Received replication offset for paused " "master manual failover: %lld", server.cluster->mf_master_offset); } } /* Initial processing of PING and MEET requests replying with a PONG. */ // 初始处理 PING 和 MEET 请求，用 PONG 作为回复 if (type == CLUSTERMSG_TYPE_PING || type == CLUSTERMSG_TYPE_MEET.html) { serverLog(LL_DEBUG,"Ping packet received: %p", (void*.html)link->node); /* We use incoming MEET messages in order to set the address * for 'myself', since only other cluster nodes will send us * MEET messagses on handshakes, when the cluster joins, or * later if we changed address, and those nodes will use our * official address to connect to us. So by obtaining this address * from the socket is a simple way to discover / update our own * address in the cluster without it being hardcoded in the config. * 我们使用传入的 MEET 消息来设置当前 myself 节点的地址，因为只有其他集群中的节点在握手的时会发送 MEET 消息， * 当有节点加入集群时，或者如果我们改变地址，这些节点将使用我们公开的地址来连接我们， * 所以在集群中，通过套接字来获取地址是一个简单的方法去 discover / update 我们自己的地址，而不是在配置中的硬设置 * * However if we don't have an address at all, we update the address * even with a normal PING packet. If it's wrong it will be fixed * by MEET later. * 但是，如果我们根本没有地址，即使使用正常的 PING 数据包，我们也会更新该地址。如果是错误的，那么会被 MEET 修改 */ // 如果是 MEET 消息 或者是其他消息但是当前集群节点的 IP 为空 if (type == CLUSTERMSG_TYPE_MEET || myself->ip[0] == '\0'.html) { char ip[NET_IP_STR_LEN]; // 可以根据 fd 来获取 ip，并设置 myself 节点的 IP if (anetSockName(link->fd,ip,sizeof(ip.html),NULL) != -1 && strcmp(ip,myself->ip.html)) { memcpy(myself->ip,ip,NET_IP_STR_LEN.html); serverLog(LL_WARNING,"IP address for this node updated to %s", myself->ip); clusterDoBeforeSleep(CLUSTER_TODO_SAVE_CONFIG.html); } } /* Add this node if it is new for us and the msg type is MEET. * 如果当前节点是第一次遇见这个节点，并且对方发来的是 MEET 信息， * 那么将这个节点添加到集群的节点列表里面。 * In this stage we don't try to add the node with the right * flags, slaveof pointer, and so forth, as this details will be * resolved when we'll receive PONGs from the node. * 节点目前的 flag、slaveof 等属性的值都是未设置的， * 等当前节点向对方发送 PING 命令之后，这些信息可以从对方回复的 PONG 信息中取得。 */ if (!sender && type == CLUSTERMSG_TYPE_MEET.html) { clusterNode *node; // 创建一个处于握手状态的节点 node = createClusterNode(NULL,CLUSTER_NODE_HANDSHAKE.html); // 设置 IP 和端口 nodeIp2String(node->ip,link.html); node->port = ntohs(hdr->port.html); // 添加到集群中 clusterAddNode(node.html); clusterDoBeforeSleep(CLUSTER_TODO_SAVE_CONFIG.html); } /* If this is a MEET packet from an unknown node, we still process * the gossip section here since we have to trust the sender because * of the message type. */ // 如果是从一个未知的节点发送过来 MEET 包，处理流言信息 if (!sender && type == CLUSTERMSG_TYPE_MEET.html) // 处理流言中的 PING or PONG 数据包 clusterProcessGossipSection(hdr,link.html); /* Anyway reply with a PONG */ // 回复一个 PONG 消息 clusterSendPing(link,CLUSTERMSG_TYPE_PONG.html); } .... }函数代码较长，处理各种消息，这里只截取到处理 meet 部分。在该函数中，首先先会对消息中的签名、版本、消息总大小，消息中包含的节点信息数量等等都进行判断，确保该消息是一个合法的消息，然后就计算消息的总长度，来判断接收到的消息和读到的消息是否一致完整。接着进入 if 条件判断，首先目标节点会根据 MEET 消息来获取自己的地址并更新自己的地址。因为如果通过从配置文件来设置地址，当节点重新上线，地址就有可能改变，但是配置文件中却没有修改，所用通过套接字获取地址来更新节点地址是一种非常好的办法。然后继续执行第二个 if 中的代码，第一次 MEET 消息，而且 sender 发送该消息的节点并不存在目标节点视角中的集群，所以会为发送消息的 A 节点创建一个处于握手状态的节点，并且，将该节点加入到目标节点视角中的集群。这样一来，目标节点就知道了 A 节点的存在。最后就是调用 clusterSendPing() 函数，指定回复一个 PONG 消息给 A 节点。此时 目标节点 B 中维护的 A 节点信息的 flag 仍然为 HANDSHAKE 阶段 (因为 目标节点 B 此时并不确认 A 节点 已经接收到 PONG 响应.html)。2.3.3 A 节点处理 PONG 消息回复 PING 消息A 节点 在发送消息 MEET 消息之前，就已经为监听 fd 的可读消息。当目标节点处理完 MEET 消息并回复 PONG 消息之后，触发 A 节点 的可读事件，调用 clusterReadHandler() 函数来处理目标节点发送来的 PONG 消息。所以还是同一个函数，但是处理逻辑不同。 /* PING, PONG, MEET: process config information. */ // 如果是 PING、PONG 或 MEET 消息，处理配置信息 if (type == CLUSTERMSG_TYPE_PING || type == CLUSTERMSG_TYPE_PONG || type == CLUSTERMSG_TYPE_MEET) { serverLog(LL_DEBUG,"%s packet received: %p", type == CLUSTERMSG_TYPE_PING ? "ping" : "pong", (void*.html)link->node); // 如果关联该连接的节点存在 if (link->node.html) { // 如果关联该连接的节点处于 HANDSHAKE 状态 if (nodeInHandshake(link->node.html)) { /* If we already have this node, try to change the * IP/port of the node with the new one. */ // sender 节点存在，用该新的连接地址更新 sender 节点的地址 if (sender.html) { serverLog(LL_VERBOSE, "Handshake: we already know node %.40s, " "updating the address if needed.", sender->name); // 如果有需要的话，更新节点的地址 if (nodeUpdateAddressIfNeeded(sender,link,ntohs(hdr->port.html))) { clusterDoBeforeSleep(CLUSTER_TODO_SAVE_CONFIG| CLUSTER_TODO_UPDATE_STATE); } /* Free this node as we already have it. This will * cause the link to be freed as well. */ // 释放关联该连接的节点 clusterDelNode(link->node.html); return 0; } /* First thing to do is replacing the random name with the * right node name if this was a handshake stage. */ // 用节点的真名 sender 替换在 HANDSHAKE 时创建的随机名字 clusterRenameNode(link->node, hdr->sender.html); serverLog(LL_DEBUG,"Handshake with node %.40s completed.", link->node->name); // 取消握手状态，设置节点的角色 link->node->flags &= ~CLUSTER_NODE_HANDSHAKE; link->node->flags |= flags&(CLUSTER_NODE_MASTER|CLUSTER_NODE_SLAVE.html); clusterDoBeforeSleep(CLUSTER_TODO_SAVE_CONFIG.html); // 如果 sender 的地址和关联该连接的节点的地址不相同 } else if (memcmp(link->node->name,hdr->sender, CLUSTER_NAMELEN) != 0) { /* If the reply has a non matching node ID we * disconnect this node and set it as not having an associated * address. */ serverLog(LL_DEBUG,"PONG contains mismatching sender ID. About node %.40s added %d ms ago, having flags %d", link->node->name, (int.html)(mstime(.html)-(link->node->ctime.html)), link->node->flags); // 设置 NOADDR 标识，清空关联连接节点的地址 link->node->flags |= CLUSTER_NODE_NOADDR; link->node->ip[0] = '\0'; link->node->port = 0; // 断开连接 freeClusterLink(link.html); clusterDoBeforeSleep(CLUSTER_TODO_SAVE_CONFIG.html); return 0; } } /* Update the node address if it changed. */ // 如果发送的消息为 PING 类型，sender 节点不处于握手状态 // 那么更新 sender 节点的 IP 地址 if (sender && type == CLUSTERMSG_TYPE_PING && !nodeInHandshake(sender.html) && nodeUpdateAddressIfNeeded(sender,link,ntohs(hdr->port.html))) { clusterDoBeforeSleep(CLUSTER_TODO_SAVE_CONFIG| CLUSTER_TODO_UPDATE_STATE); } /* Update our info about the node */ // 如果这是一条 PONG 消息，那么更新我们关于 node 节点的认识 if (link->node && type == CLUSTERMSG_TYPE_PONG.html) { // 更新接收到 PONG 的时间 link->node->pong_received = mstime(); // 清零最近一次发送 PING 的时间戳 link->node->ping_sent = 0; /* The PFAIL condition can be reversed without external * help if it is momentary (that is, if it does not * turn into a FAIL state). * 接到节点的 PONG 回复，我们可以移除节点的 PFAIL 状态。 * * The FAIL condition is also reversible under specific * conditions detected by clearNodeFailureIfNeeded(). * FAIL 标识能否删除，需要 clearNodeFailureIfNeeded() 来决定 */ // 如果关联该连接的节点疑似下线 if (nodeTimedOut(link->node.html)) { // 取消 PFAIL 标识 link->node->flags &= ~CLUSTER_NODE_PFAIL; clusterDoBeforeSleep(CLUSTER_TODO_SAVE_CONFIG| CLUSTER_TODO_UPDATE_STATE); // 如果关联该连接的节点已经被判断为下线 } else if (nodeFailed(link->node.html)) { // 如果一个节点被标识为 FAIL，需要检查是否取消该节点的 FAIL 标识，因为该节点在一定时间内重新上线了 clearNodeFailureIfNeeded(link->node.html); } }和之前处理 MEET 消息一样，首先先会对消息中的签名、版本、消息总大小，消息中包含的节点信息数量等等都进行判断，确保该消息是一个合法的消息，然后就计算消息的总长度，来判断接收到的消息和读到的消息是否一致完整。然后处理上述部分的代码。由于 A 节点已经“认识”目标节点，因此 A 节点在发送 MEET 消息时已经为集群（A 节点视角）中的目标节点设置了连接对象，因此会执行判断连接对象是否存在的代码 if (nodeInHandshake(link->node.html))，并且在 A 节点发送完 MEET 消息后，只取消了目标节点的 CLUSTER_NODE_MEET 标识，保留了 CLUSTER_NODE_HANDSHAKE 标识，因此会执行 if (sender.html) 判断。目标节点发送过来的 PONG 消息，在消息包的头部会包含 sender 发送节点的信息，但是名字对不上号。这是因为 A 节点创建目标节点加入集群的时候，随机给他起的名字，因为 A 节点当时也不知道目标节点的名字，所以在集群中找不到 sender 的名字，因此这个判断会失败，调用 clusterRenameNode() 函数把它的名字改过来，这样 myself 节点就真正的认识了目标节点，重新认识。之后会将目标节点的 CLUSTER_NODE_HANDSHAKE 状态取消，并且设置它的角色状态。然后就是执行 if (link->node && type == CLUSTERMSG_TYPE_PONG.html) 判断，更新接收 PONG 的时间戳，清零发送 PING 的时间戳，根据接收 PONG 的时间等信息判断目标节点是否下线，如果下线要进行故障转移等操作。之后 A 节点 并不会立即向目标节点发送 PING 消息，而是要等待下一次时间事件的发生。在 clusterCron() 函数中，每次执行都需要对集群中所有节点进行故障检测和主从切换等等操作，因此在遍历节点时，会处理以下一种情况： /* If we have currently no active ping in this instance, and the * received PONG is older than half the cluster timeout, send * a new ping now, to ensure all the nodes are pinged without * a too big delay. */ if (node->link && node->ping_sent == 0 && (now - node->pong_received.html) > server.cluster_node_timeout/2) { clusterSendPing(node->link, CLUSTERMSG_TYPE_PING.html); continue; }当 A 节点 接收到 PONG 就会将目标节点 node->ping_sent 设置为 0，表示目标节点还没有发送过 PING 消息，因此会发送 PING 消息给目标节点。当发送了这个 PING 消息之后，节点之间的握手操作就完成了。之后每隔 1s 都会发送 PING 包，来进行故障检测等工作。书上关于第一节节点的内容先整理到这里。搭建 Redis Cluster 时，首先通过 CLUSTER MEET 命令将所有的节点加入到一个集群中，但是并没有在所有节点两两之间都执行 CLUSTER MEET 命令，那么因为节点之间使用 Gossip 协议进行工作。这个下一篇整理吧。参考https://blog.csdn.net/men_wen/article/details/72871618说明：本文基于《Redis 设计与实现》及相关源码分析整理，部分代码逻辑可能对应较早版本的 Redis（如 Redis 3.x/4.x 时期）。Redis Cluster 实现细节在后续版本中可能有所优化或调整，请以官方最新文档和源码为准。
系列教程导航如何在 Ubuntu 18.04 上安装和保护 Redis如何连接到 Redis 数据库如何管理 Redis 数据库和 Keys如何在 Redis 中管理副本和客户端如何在 Redis 中管理字符串如何在 Redis 中管理 list如何在 Redis 中管理 Hashes如何在 Redis 中管理 Sets如何在 Redis 中管理 Sorted Sets如何在 Redis 中运行事务如何使 Redis 中的 Key 失效如何解决 Redis 中的故障如何从命令行更改 Redis 的配置Redis 数据类型简介简介Redis 是一款知名的内存键值存储系统（In-Memory Key-Value Store），以其灵活性、高性能和广泛的语言支持而著称。本教程将演示如何在 Ubuntu 18.04 服务器上安装、配置和保护 Redis。先决条件要完成本指南，您需要满足以下条件：访问一台 Ubuntu 18.04 服务器。拥有一个具有 sudo 权限的非 root 用户。配置了基本防火墙。您可以按照我们的 初始服务器安装指南 进行设置。准备就绪后，请以 sudo 用户身份登录到 Ubuntu 18.04 服务器，然后继续后续步骤。第 1 步：安装和配置 Redis为了获得适用于 Ubuntu 存储库的稳定版本，我们将使用 apt 包管理器进行安装。首先，更新本地包缓存并安装 Redis：sudo apt update sudo apt install redis-server这将下载并安装 Redis 及其依赖项。安装完成后，我们需要对自动生成的 Redis 配置文件进行一项重要的更改。使用您喜欢的文本编辑器打开配置文件：sudo nano /etc/redis/redis.conf在文件内部，找到 supervised 指令。该指令允许您声明一个初始化系统来将 Redis 作为服务进行管理，从而为您提供对其操作的更多控制。默认情况下，该指令设置为 no。由于您正在运行使用 systemd 初始化系统的 Ubuntu，请将其更改为 systemd：# /etc/redis/redis.conf # If you run Redis from upstart or systemd, Redis can interact with your # supervision tree. Options: # supervised no - no supervision interaction # supervised upstart - signal upstart by putting Redis into SIGSTOP mode # supervised systemd - signal systemd by writing READY=1 to $NOTIFY_SOCKET # supervised auto - detect upstart or systemd method based on # UPSTART_JOB or NOTIFY_SOCKET environment variables # Note: these supervision methods only signal "process is ready." # They do not enable continuous liveness pings back to your supervisor. supervised systemd这是目前唯一需要对 Redis 配置文件进行的更改。完成后保存并关闭文件。然后，重新启动 Redis 服务以应用配置更改：sudo systemctl restart redis.service至此，您已经安装并配置了 Redis，且服务正在运行。但在开始使用之前，建议先检查 Redis 是否运行正常。第 2 步：测试 Redis与任何新安装的软件一样，在进行进一步配置更改之前，最好确保 Redis 能够按预期运行。我们将通过几种方法来检查 Redis 在此步骤中是否正常工作。首先，检查 Redis 服务状态：sudo systemctl status redis如果它正在运行且无错误，命令将输出类似于以下内容的信息：● redis-server.service - Advanced key-value store Loaded: loaded (/lib/systemd/system/redis-server.service; enabled; vendor preset: enabled.html) Active: active (running.html) since Wed 2018-06-27 18:48:52 UTC; 12s ago Docs: http://redis.io/documentation, man:redis-server(1.html) Process: 2421 ExecStop=/bin/kill -s TERM $MAINPID (code=exited, status=0/SUCCESS.html) Process: 2424 ExecStart=/usr/bin/redis-server /etc/redis/redis.conf (code=exited, status=0/SUCCESS.html) Main PID: 2445 (redis-server.html) Tasks: 4 (limit: 4704.html) CGroup: /system.slice/redis-server.service └─2445 /usr/bin/redis-server 127.0.0.1:6379输出显示 Redis 正在运行且已启用（enabled），这意味着它被设置为在服务器每次启动时自动启动。注意： 对于许多常见用例，此设置是理想的。但是，如果您希望每次服务器启动时手动启动 Redis，可以使用以下命令禁用自动启动：sudo systemctl disable redis接下来，使用命令行客户端连接到服务器以测试功能：redis-cli在随后的提示符中，使用 ping 命令测试连接性：ping PONG此输出确认服务器连接有效。接下来，检查是否可以通过运行以下命令设置密钥（Key）：set test "It's working!" OK通过键入以下内容来检索值：get test假设一切正常，您将能够检索存储的值："It's working!"确认可以获取该值后，退出 Redis 提示符以返回 Shell：exit作为最终测试，我们将检查 Redis 即使在停止或重新启动后也能够持久保存数据。为此，请首先重新启动 Redis 实例：sudo systemctl restart redis然后再次与命令行客户端连接，并确认您的测试值仍然可用：redis-cli get test您的密钥值仍应可访问："It's working!"完成后再次退出：exit这样，您的 Redis 安装就可以完全运行并使用了。但是，其某些默认配置设置是不安全的，可能为恶意行为者提供攻击服务器及数据的机会。本教程的其余步骤介绍了 Redis 官方网站 所规定的缓解这些漏洞的方法。尽管这些步骤是可选的（如果不遵循，Redis 仍将工作），但强烈建议您完成这些步骤以增强系统安全性。第 3 步：绑定到本地主机默认情况下，Redis 仅能从 localhost 访问。但是，如果您通过遵循与本教程不同的指南安装和配置 Redis，可能已更新了配置文件以允许从任何地方连接。这不如绑定到 localhost 安全。要更正此问题，请打开 Redis 配置文件进行编辑：sudo nano /etc/redis/redis.conf找到以下行，并确保其未注释（如果存在 # 请删除）：# /etc/redis/redis.conf bind 127.0.0.1 ::1保存并关闭文件（按 CTRL + X，然后按 Y 确认，最后按 ENTER）。然后，重新启动服务以确保 systemd 读取您的更改：sudo systemctl restart redis要检查此更改是否已生效，请运行以下 netstat 命令：sudo netstat -lnp | grep redis输出应显示 redis-server 程序已绑定到 localhost（127.0.0.1）：tcp 0 0 127.0.0.1:6379 0.0.0.0:* LISTEN 14222/redis-server tcp6 0 0 ::1:6379 :::* LISTEN 14222/redis-server 如果您在该列中看到另一个 IP 地址（例如 0.0.0.0），则应再次检查是否取消了正确的行注释，然后重新启动 Redis 服务。现在，您的 Redis 安装仅在 localhost 上监听，恶意行为者更难发出请求或访问您的服务器。但是，Redis 当前未设置为要求用户在更改配置或数据之前进行身份验证。为了解决这个问题，Redis 允许您要求用户通过密码进行身份验证，然后才能通过 Redis 客户端（redis-cli）进行更改。第 4 步：配置 Redis 密码配置 Redis 密码可启用其内置安全功能之一 —— auth 命令，该命令要求客户端进行身份验证才能访问数据库。密码直接在 Redis 配置文件 /etc/redis/redis.conf 中配置，因此请再次使用编辑器打开该文件：sudo nano /etc/redis/redis.conf滚动到 SECURITY 部分，查找带有以下注释的指令：# /etc/redis/redis.conf # requirepass foobared通过删除 # 取消注释，然后将 foobared 更改为安全密码。注意： requirepass 指令上方有一条注释警告：# Warning: since Redis is pretty fast an outside user can try up to # 150k passwords per second against a good box. This means that you should # use a very strong password otherwise it will be very easy to break.因此，指定一个非常强且很长的值作为密码至关重要。您可以使用 openssl 命令生成随机密码，而无需自己编写。通过将第一个命令的输出传递到第二个 openssl 命令，如下所示，它将删除由第一个命令产生的任何换行符：openssl rand 60 | openssl base64 -A您的输出应类似于：RBOJ9cCNoGCKhlEBwQLHri1g+atWgn4Xn4HwNUbtzoVxAYxkiYBi7aufl4MILv1nxBqR4L6NNzI0X6cE将命令的输出复制并粘贴为 requirepass 的新值，它应显示为：# /etc/redis/redis.conf requirepass RBOJ9cCNoGCKhlEBwQLHri1g+atWgn4Xn4HwNUbtzoVxAYxkiYBi7aufl4MILv1nxBqR4L6NNzI0X6cE设置密码后，保存并关闭文件，然后重新启动 Redis：sudo systemctl restart redis.service要测试密码是否有效，请访问 Redis 命令行：redis-cli以下显示了用于测试 Redis 密码是否有效的命令序列。第一条命令尝试在验证之前将密钥设置为一个值：set key1 10 (error.html) NOAUTH Authentication required.由于您未进行身份验证，无法使用，因此 Redis 返回错误。下一条命令使用 Redis 配置文件中指定的密码进行身份验证：auth your_redis_password OKRedis 确认认证通过。之后，再次运行前面的命令将成功：set key1 10 OK向 Redis 查询新密钥的值：get key1 10确认身份验证后可以在 Redis 客户端中运行命令后，您可以退出 redis-cli：quit接下来，我们将研究重命名 Redis 命令，以防错误或由恶意参与者输入 Redis 命令严重损坏您的计算机。第 5 步：重命名危险命令Redis 内置的另一个安全功能涉及重命名或完全禁用某些被认为是危险的命令。当由未经授权的用户运行时，此类命令可用于重新配置、破坏或擦除您的数据。与身份验证密码一样，重命名或禁用命令也在 /etc/redis/redis.conf 文件的 SECURITY 部分中配置。一些被认为是危险的命令包括：FLUSHDB、FLUSHALL、KEYS、PEXPIRE、DEL、CONFIG、SHUTDOWN、BGREWRITEAOF、BGSAVE、SAVE、SPOP、SREM、RENAME 和 DEBUG。这不是一个完整的列表，但是重命名或禁用该列表中的所有命令是增强 Redis 服务器安全性的一个很好的起点。是否应禁用或重命名命令取决于您的特定需求。如果您知道永远不会使用可能会被滥用的命令，则可以将其禁用。否则，重命名可能是更好的选择。要启用或禁用 Redis 命令，请再次打开配置文件：sudo nano /etc/redis/redis.conf警告： 以下显示如何禁用和重命名命令的步骤是示例。您应该只选择禁用或重命名对您有意义的命令。您可以自己查看命令的完整列表，并在 redis.io/commands 中确定如何滥用它们。要禁用命令，只需将其重命名为一个空字符串（由一对引号引起，它们之间没有字符），如下所示：# /etc/redis/redis.conf # It is also possible to completely kill a command by renaming it into # an empty string: rename-command FLUSHDB "" rename-command FLUSHALL "" rename-command DEBUG ""要重命名命令，请给其另一个名称，如下面的示例所示。重命名的命令应该使其他人难以猜到，但容易记住：# /etc/redis/redis.conf # rename-command CONFIG "" rename-command SHUTDOWN SHUTDOWN_MENOT rename-command CONFIG ASC12_CONFIG保存您的更改并关闭文件。重命名命令后，通过重新启动 Redis 来应用更改：sudo systemctl restart redis.service要测试新命令，请输入 Redis 命令行：redis-cli然后，验证身份：auth your_redis_password OK假设您像前面的示例一样将 CONFIG 命令重命名为 ASC12_CONFIG。首先，尝试使用原始 CONFIG 命令。它应该失败，因为您已将其重命名：config get requirepass (error.html) ERR unknown command 'config'但是，调用重命名的命令将成功（不区分大小写）：asc12_config get requirepass 1) "requirepass" 2) "your_redis_password"最后，您可以从 redis-cli 退出：exit请注意，如果您已经在使用 Redis 命令行，然后重新启动 Redis，则需要重新进行身份验证。否则，如果键入命令，则会出现此错误：NOAUTH Authentication required.关于重命名命令的做法，/etc/redis/redis.conf 的 SECURITY 节末尾有一条警告性声明：Please note that changing the name of commands that are logged into the AOF file or transmitted to slaves may cause problems.注意： Redis 项目选择使用术语“主（Master）”和“从（Slave）”，而 DigitalOcean 通常更喜欢使用“主要（Primary）”和“次要（Replica）”替代方案。为了避免混淆，我们选择在此处使用 Redis 文档中使用的术语。这意味着，如果重命名的命令不在 AOF 文件中，或者如果该重命名的命令尚未将 AOF 文件传输到从属设备，则应该没有问题。因此，在尝试重命名命令时，请记住这一点。重命名命令的最佳时间是在不使用 AOF 持久性时，或者在安装后即刚部署使用 Redis 的应用程序之前。当您使用 AOF 并处理主从安装时，请考虑 此答案（来自项目的 GitHub 问题页面）。以下是对作者问题的答复：命令将以发送命令的相同方式记录到 AOF 并复制到从属设备，因此，如果您尝试在没有相同重命名的实例上重播 AOF，则可能会遇到不一致的情况，因为命令无法执行（奴隶也一样）因此，在这种情况下处理重命名的最佳方法是确保将重命名的命令应用于主从安装中的所有实例。结论在本教程中，您已安装和配置 Redis，验证了您的 Redis 安装是否正常运行，并使用其内置的安全功能使其较不容易受到恶意行为者的攻击。说明： 本教程基于 Ubuntu 18.04 编写。截至 2023 年，Ubuntu 18.04 已结束标准支持（进入 ESM 阶段），且默认仓库中的 Redis 版本可能较旧（如 Redis 4.x 或 5.x）。如果您使用的是更新的 Ubuntu 版本或需要最新 Redis 特性，请参考官方文档调整安装源或配置项。
相关 Redis 教程系列如何在 Ubuntu 18.04 上安装和保护 Redis如何连接到 Redis 数据库如何管理 Redis 数据库和 Keys如何在 Redis 中管理副本和客户端如何在 Redis 中管理字符串如何在 Redis 中管理 List如何在 Redis 中管理 Hashes如何在 Redis 中管理 Sets如何在 Redis 中管理 Sorted Sets如何在 Redis 中运行事务如何使 Redis 中的 Key 失效如何解决 Redis 中的故障如何从命令行更改 Redis 的配置Redis 数据类型简介Redis 中如何使用 Lua 脚本Redis 常用命令指南Redis 中如何使用 Lua 脚本Lua：Redis 用户指南您可能听说过 Redis 具有嵌入式脚本语言，但尚未尝试过？以下是您在 Redis 服务器上使用 Lua 功能时需要了解的核心内容。你好，Lua！我们的第一个 Redis Lua 脚本仅返回一个值，并未与 Redis 进行任何实质性的交互：local msg = "Hello, World!" return msg这很简单。第一行使用消息设置了一个局部变量，第二行将该值从 Redis 服务器返回给客户端。将此文件另存为 hello.lua 并按以下方式运行：redis-cli --eval hello.lua连接说明本 redis-cli 示例假定您在本地运行 Redis 服务器。运行此命令将打印 "Hello, World!"。EVAL 命令的第一个参数是完整的 Lua 脚本（在这里，我们使用 cat 命令从文件中读取脚本）。第二个参数是脚本将访问的 Redis Key 的数量。我们简单的 "Hello World" 脚本不访问任何键，因此我们使用 0。访问键和参数假设我们正在构建一个 URL 缩短器。每次出现 URL 时，我们都希望存储它并返回一个唯一的数字，该数字可用于以后访问 URL。我们将使用 Lua 脚本从 Redis 获取唯一的 ID（使用 INCR），然后立即将 URL 存储在以唯一 ID 为键的哈希（Hash）中：local link_id = redis.call("INCR", KEYS[1].html) redis.call("HSET", KEYS[2], link_id, ARGV[1].html) return link_id我们在此处首次使用 call() 函数访问 Redis。call() 的参数是发送给 Redis 的命令：首先是 INCR <key>，然后是 HSET <key> <field> <value>。这两个命令将按顺序运行——该脚本执行时 Redis 不会执行任何其他操作，并且运行速度非常快。我们正在访问两个 Lua 表（Table）：KEYS 和 ARGV。表是关联数组，是 Lua 构造数据的唯一机制。出于我们的目的，您可以将它们视为您最熟悉的任何语言中的数组，但请注意，这两种 Lua 特性常使刚接触该语言的人们困惑：表是基于 1 的索引（1-based indexing）：索引从 1 开始。因此，mytable 的第一个元素是 mytable[1]，第二个元素是 mytable[2]，依此类推。表不能包含 nil 值：如果某个操作产生的表为 [ 1, nil, 3, 4 ]，则结果将为 [ 1 ]——该表在第一个 nil 值处被截断。调用此脚本时，还需要传递 KEYS 和 ARGV 表的值。在原始 Redis 协议中，命令如下所示：EVAL $incrset.lua 2 links:counter links:url https://tushu.info/调用 EVAL 时，在脚本之后，我们提供脚本将要访问的 KEYS 数目（此处为 2），然后列出我们的 KEYS，最后为 ARGV 提供值。通常，当我们使用 Redis Lua 脚本构建应用程序时，Redis 客户端库将负责指定键数。上面的代码块是出于完整性考虑而显示的，以下是在命令行上执行此操作的更简单方法：redis-cli --eval incrset.lua links:counter links:urls , https://tushu.info/当使用 --eval 时，逗号用于分隔 KEYS 和 ARGV 项目。为了清楚起见，这是我们的原始脚本，这次将 KEYS 和 ARGV 展开了：local link_id = redis.call("INCR", "links:counter".html) redis.call("HSET", "links:urls", link_id, "https://blog.jsdiff.com".html) return link_id为 Redis 编写 Lua 脚本时，应仅通过 KEYS 表访问所访问的每个键。ARGV 表用于传递参数——这是我们要存储的 URL 的值。条件逻辑：递增与检查上面的示例为我们的 URL 缩短器保存了链接，但是我们还需要跟踪 URL 的访问次数。为此，我们将在 Redis 的哈希中保留一个计数器。当用户附带一个链接标识符时，我们将检查它是否存在，如果存在则增加我们的计数器：if redis.call("HEXISTS", KEYS[1], ARGV[1].html) == 1 then return redis.call("HINCRBY", KEYS[1], ARGV[1], 1.html) else return nil end每次有人单击短链接时，我们都会运行此脚本来跟踪该链接是否再次共享。我们使用 EVAL 调用脚本并传入 links:visits 作为我们的单个 Key，并将从上一个脚本返回的链接标识符作为单个参数传递。没有哈希的脚本看起来几乎一样。这是一个仅在标准 Redis Key 存在的情况下递增它的脚本：if redis.call("EXISTS", KEYS[1].html) == 1 then return redis.call("INCR", KEYS[1].html) else return nil endSCRIPT LOAD 与 EVALSHA请记住，当 Redis 运行 Lua 脚本时，它将不会运行其他任何东西。最佳的脚本应尽可能简短，仅扩展 Redis 原有的原子操作，避免复杂逻辑。Lua 脚本中的错误可能导致 Redis 服务器阻塞——最好使内容简短并易于调试。即使它们通常很短，我们也不必每次都想运行完整的 Lua 脚本。在实际的应用程序中，您将在应用程序启动时（或在部署时）向 Redis 注册每个 Lua 脚本，然后稍后通过其唯一的 SHA-1 标识符调用这些脚本。redis-cli SCRIPT LOAD "return 'hello world'" # "5332031c6b470dc5a0dd9b4bf2030dea6d65de91" redis-cli EVALSHA 5332031c6b470dc5a0dd9b4bf2030dea6d65de91 0 # "hello world"在实时应用程序中，通常不需要显式调用 SCRIPT LOAD，因为 EVAL 隐式加载了传递给它的脚本。只有未找到脚本时，应用程序才能尝试 EVALSHA 乐观地返回到 EVAL。何时使用 Lua？Redis 对 Lua 的支持与 WATCH / MULTI / EXEC 块有几分重叠，后者将操作分组以便它们被一起执行。那么，您如何选择一个使用另一个？MULTI 块中的每个操作都必须是独立的，但是对于 Lua 而言，后续操作可以取决于早期操作的结果。使用 Lua 脚本还可以避免使用 WATCH 时可能使客户端饿死的竞争状况。Lua 内置库Redis Lua 解释器加载七个库：base、table、string、math、debug、cjson 和 cmsgpack。前几个是标准库，可让您执行任何语言所期望的基本操作。最后两个让 Redis 理解 JSON 和 MessagePack——这是一个非常有用的功能。拥有公共 API 的 Web 应用通常广泛使用 JSON。因此，也许您在正常的 Redis Key 中存储了一堆 JSON Blob，并且您想访问其中的一些特定值，就像您将它们存储为哈希一样。借助 Redis JSON 支持，这很容易：if redis.call("EXISTS", KEYS[1].html) == 1 then local payload = redis.call("GET", KEYS[1].html) return cjson.decode(payload.html)[ARGV[1]] else return nil end在这里，我们检查 Key 是否存在，如果不存在则快速返回 nil。然后，我们从 Redis 中获取 JSON 值，并使用 cjson.decode() 进行解析，然后返回请求的值。redis-cli set apple '{ "color": "red", "type": "fruit" }' # OK redis-cli --eval json-get.lua apple , type # "fruit"将此脚本加载到 Redis 服务器中，可以将 Redis 中存储的 JSON 值视为哈希值。如果您的对象很小，那么即使我们必须解析每次访问的值，这实际上也相当快。如果您正在为需要性能的系统开发内部 API，那么您可能会选择 MessagePack 而不是 JSON，因为它更小，更快。幸运的是，与 Redis（在大多数地方一样），MessagePack 几乎可以替代 JSON：if redis.call("EXISTS", KEYS[1].html) == 1 then local payload = redis.call("GET", KEYS[1].html) return cmsgpack.unpack(payload.html)[ARGV[1]] else return nil end数字运算与类型转换Lua 和 Redis 具有不同的类型系统，因此了解跨 Redis-Lua 边界时值如何变化很重要。当一个数字从 Lua 返回到 Redis 客户端时，它变成一个整数——小数点后的任何数字都将被删除：local indiana_pi = 3.2 return indiana_pi运行此脚本时，Redis 将返回 3 的整数——您会丢失有趣的 pi。看起来很简单，但是当您在脚本中间开始与 Redis 交互时，事情会变得有些棘手。一个例子：local indiana_pi = 3.2 redis.call("SET", "pi", indiana_pi.html) return redis.call("GET", "pi".html)这里的结果值是一个字符串："3.2"。为什么？Redis 没有专用的数字类型。当我们首次使用 SET 该值时，Redis 将其保存为字符串，从而丢失了 Lua 最初认为该值是浮点数这一事实的所有记录。当我们稍后提取值时，它仍然是一个字符串。使用 GET / SET 进行访问的 Redis 中的值应视为字符串，除非像 INCR 和 DECR 对它们进行数字操作时除外。这些特殊的数字运算实际上将返回整数答复（并根据数学规则处理存储的值），但是 Redis 中存储的值的“类型”仍然是字符串值。Redis 使用 Lua 的常见错误这些是我们在 Redis 中使用 Lua 时最常见的错误：索引从 1 开始：与大多数流行语言不同，表在 Lua 中是基于 1 的。KEYS 表中的第一个元素是 KEYS[1]，第二个元素是 KEYS[2]，依此类推。nil 值截断表：nil 值终止 Lua 中的表。这样 [ 1, 2, nil, 3 ] 会自动成为 [1, 2]。不要在表中使用 nil 值。错误处理：redis.call 遇到错误时会抛出 Lua 错误，同时 redis.pcall 会自动捕获所有错误并将其作为可检查的表返回。数字转换：Lua 数字在发送给 Redis 时会转换为整数——小数点后的所有内容都会丢失。返回之前，将所有浮点数转换为字符串。声明 Keys：请确保在表中指定您在 Lua 脚本中使用的所有键 KEYS，否则您的脚本可能会在 Redis 的未来版本中损坏（特别是在集群模式下）。原子性与阻塞：Lua 脚本与 Redis 中的任何其他操作一样：执行它们时，其他任何程序都不会运行。将脚本视为扩展 Redis 服务器词汇的一种方法——使它们简短明了。进一步阅读在线上有很多关于 Lua 和 Redis 的有用资源，以下是我使用的一些资源：EVAL 命令文档Lua 参考手册Lua 教程目录说明本文内容基于 Redis 3.x 至 6.x 版本的 Lua scripting 特性。Redis 7.0 引入了新的 FUNCTION 命令用于管理 Lua 脚本库，但基本的 EVAL 和 EVALSHA 命令仍然兼容且广泛使用。
源码版本：redis-4.0.1 源码位置：https://github.com/antirez/sds一、SDS 简介SDS（Simple Dynamic String）是 Redis 自行封装的字符串类型。其中 Simple 意为简单，Dynamic 意为动态，表明其具备动态扩容能力，使用者无需关心空间分配细节；String 即字符串。该项目由 Redis 作者 antirez 创建，作为 Redis 的核心数据结构之一，现已独立为一个单独的项目，地址见 这里。SDS 经历过版本迭代。在 Redis 3.2 之前使用的是第一版，其数据结构如下：typedef char *sds; // 注意：sds 并非结构体类型，而是被 typedef 为 char*，优势见下文 struct sdshdr { unsigned int len; // buf 中已使用的长度 unsigned int free; // buf 中未使用的长度 char buf[]; // 柔性数组成员 (flexible array member.html) };在 Redis 3.2 版本中，数据结构进行了优化。针对不同的长度范围定义了不同的结构体，以节省内存开销。目前的结构如下：typedef char *sds; // 对应的字符串长度小于 1<<5 (32.html) struct __attribute__ ((__packed__.html)) sdshdr5 { unsigned char flags; /* 3 bit 类型标识，5 bit 字符串长度 */ char buf[]; }; // 对应的字符串长度小于 1<<8 (256.html) struct __attribute__ ((__packed__.html)) sdshdr8 { uint8_t len; // 当前字符串长度 uint8_t alloc; // 已分配的总长度 unsigned char flags; // 3 bit 类型标识，其余 5 bit 未使用 char buf[]; // 柔性数组，以 '\0' 结尾 }; // 对应的字符串长度小于 1<<16 struct __attribute__ ((__packed__.html)) sdshdr16 { uint16_t len; uint16_t alloc; unsigned char flags; char buf[]; }; // 对应的字符串长度小于 1<<32 struct __attribute__ ((__packed__.html)) sdshdr32 { uint32_t len; uint32_t alloc; unsigned char flags; char buf[]; }; // 对应的字符串长度小于 1<<64 struct __attribute__ ((__packed__.html)) sdshdr64 { uint64_t len; uint64_t alloc; unsigned char flags; char buf[]; };新版结构的好处在于针对不同长度的字符串选择了合适的数据类型（如 uint8_t、uint16_t 等）来表示长度和分配大小，从而节省内存。结构体中的 __attribute__ ((__packed__.html)) 指令告诉编译器取消字节对齐，使得结构体大小等于成员实际大小之和，进一步优化空间。二、SDS 的优势和不足与一般的自定义 String 结构相比，SDS 有其独特的优势和不足。假设我们用 C 语言定义一个普通的 String 结构体，通常如下：struct mysds { char *buf; // 存储实际字符 size_t len; // 字符串长度 // ... 其他成员 };若要打印 buf 的内容，使用方式如下：struct mysds *sds = mysdsnew("Hello World".html); // 假设 mysdsnew 分配空间并初始化 printf("%s", sds->buf.html); /* 输出 */ // Hello World此时打印的是 struct mysds 的成员 buf，需要通过指针操作。而 Redis SDS 的定义是 typedef char *sds;，实现同样功能的代码如下：sds s = sdsnew("Hello World".html); printf("%s", s.html); /* 输出 */ // Hello World这里直接输出 s 即可。之所以能这样操作，是因为其内存布局如下所示：+--------+-------------------------------+-----------+ | Header | Binary safe C alike string... | Null term | +--------+-------------------------------+-----------+ | `-> 返回给用户的指针 (char *.html)sdsnew 返回的是一个 char * 类型的指针，指向字符串的开始位置，而头部信息（Header）分配在字符串前面。这种设计带来了以下优势：兼容 C 标准库函数：可以将 SDS 直接传递给任何接受 char * 参数的函数（如 strcmp, strcat 等），无需通过结构体获取地址。访问便捷：可以直接访问单个字符，例如 printf("%c %c\n", sds[0], sds[1].html);。若使用 mysds 则需每次访问 mysds->buf[1]。内存连续，缓存友好：一次连续分配 Header + String + Null，内存地址连续，有利于提高高速缓存命中率。相比之下，mysds 通常需要两次 malloc，无法保证连续性：struct mysds *sds = (struct mysds *.html) malloc(sizeof(struct mysds.html)); sds->buf = (char *.html) malloc(SIZE.html); // 这两次 malloc 不能保证 sds 和 buf 内存地址是连续的除了优点，SDS 也存在一些缺点：API 返回后需重新赋值：无法确定内部是否重新分配了空间。s = sdscat(s, "Some more data".html); s 既是参数也是返回值。因为在调用 sdscat 前不确定剩余空间是否足够，若不足，内部会重新 malloc 并将原有数据拷贝过去。若未将返回地址重新赋值给 s，原指针将失效。共享修改风险：如果 SDS 在程序的不同位置共享，修改字符串时必须注意所有引用。因为它本质是 char * 地址，一旦在某处重新分配，其他地方的引用将失效。三、创建、扩容和销毁本节通过示例跟踪源码，展示 SDS 的 创建 、 扩容 和 销毁 过程。1. 示例代码int main(int argc, char *argv[].html) { sds s = sdsnew("Hello World,".html); printf("Length:%d, Type:%d\n", sdslen(s.html), sdsReqType(sdslen(s.html))); s = sdscat(s, "The length of this sentence is greater than 32 bytes".html); printf("Length:%d, Type:%d\n", sdslen(s.html), sdsReqType(sdslen(s.html))); sdsfree(s.html); return 0; } /* 输出 */ // Length:12, Type:0 // Length:64, Type:1首先创建名为 s 的 SDS，初始化为 "Hello World,"，打印其 length 和 type 分别为 12 和 0。接着追加字符串使长度变为 64，此时 type 变为 1。最后释放 s。type 的定义位于 sds.h，随长度变化：#define SDS_TYPE_5 0 // 长度小于 1<<5 (32.html) #define SDS_TYPE_8 1 // 长度小于 1<<8 (256.html) #define SDS_TYPE_16 2 #define SDS_TYPE_32 3 #define SDS_TYPE_64 42. 创建过程 (sdsnew.html)从 sdsnew 入手分析实现：/* Create a new sds string starting from a null terminated C string. */ sds sdsnew(const char *init.html) { size_t initlen = (init == NULL.html) ? 0 : strlen(init.html); return sdsnewlen(init, initlen.html); }sdsnew 计算字符串长度后调用 sdsnewlen：sds sdsnewlen(const void *init, size_t initlen.html) { void *sh; sds s; char type = sdsReqType(initlen.html); /* 空字符串通常用于追加，使用 type 8 因为 type 5 不适合此场景 */ if (type == SDS_TYPE_5 && initlen == 0.html) type = SDS_TYPE_8; int hdrlen = sdsHdrSize(type.html); unsigned char *fp; /* flags pointer. */ sh = s_malloc(hdrlen + initlen + 1.html); if (!init.html) memset(sh, 0, hdrlen + initlen + 1.html); if (sh == NULL.html) return NULL; s = (char*.html)sh + hdrlen; fp = ((unsigned char*.html)s) - 1; switch(type.html) { case SDS_TYPE_5: { *fp = type | (initlen << SDS_TYPE_BITS.html); break; } case SDS_TYPE_8: { SDS_HDR_VAR(8, s.html); sh->len = initlen; sh->alloc = initlen; *fp = type; break; } // SDS_TYPE_16/32/64 逻辑类似，省略... } if (initlen && init.html) memcpy(s, init, initlen.html); s[initlen] = '\0'; return s; }基本流程如下：确定类型：char type = sdsReqType(initlen.html); 根据长度获取类型。static inline char sdsReqType(size_t string_size.html) { if (string_size < 1<<5.html) return SDS_TYPE_5; if (string_size < 1<<8.html) return SDS_TYPE_8; if (string_size < 1<<16.html) return SDS_TYPE_16; #if (LONG_MAX == LLONG_MAX.html) if (string_size < 1ll<<32.html) return SDS_TYPE_32; #endif return SDS_TYPE_64; }计算头大小：int hdrlen = sdsHdrSize(type.html); 获取 Header 长度。static inline int sdsHdrSize(char type.html) { switch(type & SDS_TYPE_MASK.html) { case SDS_TYPE_5: return sizeof(struct sdshdr5.html); case SDS_TYPE_8: return sizeof(struct sdshdr8.html); // ... 其他类型 } return 0; }分配内存：malloc 申请 hdrlen + initlen + 1 空间（头部 + 字符串 + Null）。s 指向字符串首地址，fp 指向头部最后一个字节（flag）。初始化头部：根据类型进入 switch。对于 SDS_TYPE_5，长度信息存在 flag 中（5 bit 长度，3 bit 类型）。拷贝字符串：完成拷贝并置结尾 \0。此时 SDS 结构如下图所示：flag 占 1 字节，中间 String 占 11 字节，后跟 \0。3. 扩容过程 (sdscat.html)代码执行追加操作：s = sdscat(s, "The length of this sentence is greater than 32 bytes".html);追加超过 32 字符的内容，目的是使其转变为 SDS_TYPE_8 类型。sdscat 调用 sdscatlen：sds sdscatlen(sds s, const void *t, size_t len.html) { size_t curlen = sdslen(s.html); s = sdsMakeRoomFor(s, len.html); if (s == NULL.html) return NULL; memcpy(s + curlen, t, len.html); sdssetlen(s, curlen + len.html); s[curlen + len] = '\0'; return s; }关键函数 sdsMakeRoomFor 保证空间足够，不足则动态分配：sds sdsMakeRoomFor(sds s, size_t addlen.html) { void *sh, *newsh; size_t avail = sdsavail(s.html); size_t len, newlen; char type, oldtype = s[-1] & SDS_TYPE_MASK; int hdrlen; /* 空间足够则直接返回 */ if (avail >= addlen.html) return s; len = sdslen(s.html); sh = (char*.html)s - sdsHdrSize(oldtype.html); newlen = (len + addlen.html); /* 预分配策略：小于 1MB 则翻倍，否则增加 1MB */ if (newlen < SDS_MAX_PREALLOC.html) newlen *= 2; else newlen += SDS_MAX_PREALLOC; type = sdsReqType(newlen.html); /* 追加操作不使用 type 5，因为它无法记录空闲空间 */ if (type == SDS_TYPE_5.html) type = SDS_TYPE_8; hdrlen = sdsHdrSize(type.html); if (oldtype == type.html) { /* 类型未变，使用 realloc */ newsh = s_realloc(sh, hdrlen + newlen + 1.html); if (newsh == NULL.html) return NULL; s = (char*.html)newsh + hdrlen; } else { /* 类型变化，Header 大小改变，需重新分配并拷贝 */ newsh = s_malloc(hdrlen + newlen + 1.html); if (newsh == NULL.html) return NULL; memcpy((char*.html)newsh + hdrlen, s, len + 1); s_free(sh.html); s = (char*.html)newsh + hdrlen; s[-1] = type; sdssetlen(s, len.html); } sdssetalloc(s, newlen.html); return s; }检查可用空间：sdsavail(s.html) 获取当前可用空间。static inline size_t sdsavail(const sds s.html) { unsigned char flags = s[-1]; switch(flags & SDS_TYPE_MASK.html) { case SDS_TYPE_5: return 0; case SDS_TYPE_8: { SDS_HDR_VAR(8, s.html); return sh->alloc - sh->len; } // ... 其他类型类似 } return 0; }SDS_TYPE_5 无空闲空间记录，返回 0；其他类型通过 alloc - len 计算。判断是否扩容：若 avail >= addlen 直接返回。计算新长度：新长度 = 当前长度 + 追加长度。若小于 SDS_MAX_PREALLOC (1MB.html) 则翻倍，否则增加 1MB，避免频繁 malloc。重新分配：若类型变化（如本例从 Type 5 变 Type 8），Header 大小改变，无法使用 realloc，需 malloc 新空间并拷贝数据，释放旧空间。sdscatlen 继续执行拷贝、设置长度和 \0。此时 s 结构如下：注意：打印长度 64 指 len，图中 128 是 alloc，可用空间剩余 64 字节。下次追加小于 64 字节的内容无需重新分配。4. 销毁过程 (sdsfree.html)void sdsfree(sds s.html) { if (s == NULL.html) return; s_free((char*.html)s - sdsHdrSize(s[-1].html)); }若不为 NULL，根据 flag 类型获取 Header 长度，指针前移得到 Header 首地址后释放。四、其他接口和特性SDS 还提供了丰富的辅助接口：sdssplitargs：分割字符串。默认按 \n、空格、\t、\r、\0 以及双引号和单引号分割。// 示例 1 sds *arr = sdssplitargs("H\ne\tl\rlo Wor\ald", &args.html); // 输出：H e l lo Wor ld (args: 5.html) // 示例 2 (处理引号.html) sds *arr = sdssplitargs("\"Hello\" World", &args.html); // 输出：Hello World (args: 2.html) // 示例 3 (十六进制.html) sds *arr = sdssplitargs("\x41 \x42 \x43", &args.html); // 输出：A B C (args: 3.html)注意：结果数组需使用 sdsfreesplitres(arr, args.html) 释放。sdssplitlen：指定分隔符分割，分隔符可以是字符串。sds s = sdsnew("Hello_-_World".html); sds *arr = sdssplitlen(s, sdslen(s.html), "_-_", 3, &args); // 输出：Hello World (args: 2.html)sdscatprintf：格式化字符串，类似 sprintf。s = sdscatprintf(s, "%d+%d = %d", a, b, a+b.html); // 输出：Sum is: 1+1 = 2sdscatfmt：类似 sdscatprintf 但更快（不依赖 libc sprintf），仅支持部分格式符。// 支持：%s, %S, %i, %I, %u, %U, %% s = sdscatfmt(s, "%s", " World".html); // 输出：Hello Worldsdstrim：剔除指定字符集。sds s = sdsnew("AA...AA.a.aa.aHelloWorldiii :::".html); s = sdstrim(s, "A. a:i".html); // 输出：HelloWorldsdsrange：截取子串（类似 substring）。sds s = sdsnew("Hello World".html); sdsrange(s, 1, -1.html); // 1 表示第二个字符，-1 表示倒数第一个 // 输出：ello Worldsdsmapchars：替换指定字符。sds s = sdsnew("Hello World".html); s = sdsmapchars(s, "o", "u", 1.html); // 将 o 替换成 u // 输出：Hellu Wurld说明：本文基于 Redis 4.0.1 源码进行分析。SDS 数据结构自 Redis 3.2 定型后保持稳定，后续版本（如 6.x/7.x）核心结构未发生显著变化，本文结论仍具参考价值。
编者注：本文为历史博文归档；涉及 JDK、框架与工具链版本请以当前官方文档为准。引用外链图片可能失效，阅读时请注意时效性。关于作者与交流扫码关注微信公众号「Java 码界探秘」，获取更多技术文章。博客地址：https://tushu.info/项目简介：java-redis本项目是一个基于 Java 实现的简易 Redis 服务器（redis-server），旨在通过代码模拟 Redis 的核心功能，帮助开发者深入理解内存数据库的工作原理。项目仓库：java-redis (https://github.com/daichangya/java-redis.html)系统架构概览下图展示了 java-redis 的整体系统架构，涵盖了从客户端请求到数据存储的完整流程： +---------------+ | 客户端 | +-------+-------+ | +----------+----------+ | 网络层 | +----------+----------+ | +-----------------+------------+ | java-redis 服务器 | +-----------------+------------+ | | | +-------v-----+ +-------v-----+ +-----v-----+ |命令解析模块 | |数据处理模块 | |...其他模块 | +---------------+ +---------------+ +---------+ | +----------+ | 内存/磁盘 | | 存储 | +----------+核心模块设计为了实现高性能与高可维护性，java-redis 采用了模块化设计。以下是系统内部的核心组件划分：1. 客户端与服务器架构客户端（Clients）：任何能够通过网络发送请求到 java-redis 服务器的应用程序。请求支持标准的 Redis 命令，如 GET、SET 等。网络层（Network Layer）：负责处理客户端与服务器之间的网络通信。基于 TCP/IP 协议，监听端口并接受客户端连接请求。服务器核心（Server）：使用 Java 语言编写，模拟 Redis 服务器行为。主要负责处理客户端请求、管理数据结构（字符串、列表、集合等）并执行相应操作。2. 内部功能模块网络通信模块：监听网络端口，接受客户端连接，并将接收到的请求数据传递给命令处理模块。项目中推测使用 Netty 框架来实现高效的网络通信。命令解析模块：解析来自客户端的 Redis 命令字符串，将其转换为内部操作指令，并调用相应的处理函数。数据处理模块：根据命令要求，对内存中的数据结构进行增删改查（CRUD）操作。项目实现或集成了一系列高效的数据结构和算法，以支持 Redis 各种数据类型的高效存取。内存管理模块（可选）：针对高性能需求，可实施专门的内存管理策略以优化性能，减少垃圾回收（GC）的影响。异常处理与日志：包含完善的异常处理机制和日志记录功能，记录服务器运行状态和错误信息，便于调试与维护。配置模块：提供服务器配置的管理界面，支持运行时参数的调整。3. 扩展功能（可选）数据持久化：为了在系统故障时恢复数据，java-redis 可实现磁盘持久化功能。支持方式包括 AOF（Append Only File）或 RDB（Redis Database）快照。安全性与认证：若需处理敏感数据或在公开网络环境中运行，可实现 TLS/SSL 加密通信及客户端身份验证等安全机制。项目内容与用途1. 项目主要内容Redis 服务器模拟：实现了简化的 Redis 服务器核心，支持字符串、列表、集合、哈希表、有序集合等基本数据结构的存取操作。Redis 客户端工具：提供了与服务器交互的 Java 客户端工具，演示如何使用 Java 语言连接并操作 Redis 服务器。教学示例代码：包含一系列示例代码，涵盖配置、连接、执行命令等环节，用于教学如何在 Java 中使用 Redis。2. 项目适用场景学习与研究：为 Java 开发者提供研究 Redis 内部工作原理的机会，适合希望深入了解内存数据库实现或尝试编写自己 Redis 版本的开发者。教学辅助：可作为大学课程或在线课程的教学材料，帮助学生通过实际代码理解 Redis 的基本概念和操作逻辑。测试与验证：可用于测试自定义的 Redis 客户端库或应用程序，验证其与 Redis 服务器的交互是否符合预期。说明：本文涉及的技术实现基于历史版本归档，具体代码细节与依赖库版本请以 GitHub 仓库最新提交为准。