Khepri 日常运维

即使元数据存储不存储消息，其行为也会影响使用它的 RabbitMQ 的日常行为和技术操作，至少在使用此集群的应用程序希望对队列或流等资源进行身份验证、声明或删除时是如此。

如集群注意事项中所述，Khepri 是一个基于 Raft 的系统，与任何基于 Raft 的系统一样，必须有一定数量的集群成员在线且可用，元数据存储才能接受更新（写入/删除或集群成员资格更改）。

一致性模型和可见性保证

在模式修改（例如队列或流声明或绑定声明）方面，Khepri 与 Mnesia 有一个重要的区别。这些更改在许多工作负载中不会引起注意，但可能会影响某些工作负载，特别是某些集成测试。

示例场景

考虑两种场景，A 和 B。

只有一个客户端。客户端执行以下步骤

1. 它声明一个队列 Q。
2. 它将 Q 绑定到交换机 X。
3. 它将消息 M 发布到交换机 X。
4. 它期望消息被路由到队列 Q。
5. 它消费消息。

场景 A

在这种情况下，行为上应该没有明显的差异。客户端的期望将得到满足。

有两个客户端，One 和 Two，分别连接到节点 R1 和 R3，并使用相同的虚拟主机。节点 R2 没有客户端连接。

1. 客户端 One 声明一个队列 Q。
2. 它将 Q 绑定到交换机 X。
3. 它收到队列和绑定声明确认。
4. 它通知客户端 Two 或客户端 Two 隐式地发现它已完成上述步骤（例如，在集成测试中）。
5. 客户端 Two 将消息 M 发布到 X。
6. 客户端 One 和 Two 期望消息被路由到 Q。

场景 B

在这种情况下，在步骤三中，Mnesia 将在**所有**集群节点都提交更新后返回。但是，Khepri 将在包括处理客户端 One 操作的节点在内的**大多数**节点返回后返回。

这可能包括节点 R1 和 R2，但不包括节点 R3，这意味着在上述示例中，连接到节点 R3 的客户端 Two 发布的消息 M**不保证被路由**。

一旦所有模式更改传播到节点 R3，客户端 Two 随后在节点 R3 上的发布**将保证**被路由。

这是基于 Raft 的系统的权衡，它假设由大多数节点接受的写入可以被视为成功。

解决方法

为了满足场景 B 中客户端 Two 的期望，Khepri 可以执行绑定时的**一致性**（涉及大多数副本）查询，但这会对某些协议（例如 MQTT）和交换机/目标类型（任何类似于 AMQP 0-9-1 中的主题交换机的内容）的吞吐量产生**重大**影响。

依赖于多个连接并依赖于共享拓扑的应用程序有多种应对策略。

如果应用程序使用两个或多个连接到不同的节点，它可以在启动时声明其拓扑，然后在继续其他操作之前注入短暂的暂停（1-2 秒）。

依赖于动态拓扑的应用程序可以切换到使用“静态”交换机和绑定集。

不需要使用共享拓扑的应用程序组件可以分别配置其自己的队列/流/绑定。

使用多个连接到不同节点的测试套件可以选择只使用一个连接或连接到同一个节点，或者注入暂停，或者等待指示拓扑就位的特定条件。

客户端资源在集群少数方声明

定义消息如何路由的拓扑存储在元数据存储中。无论活动元数据存储后端是什么，声明资源的方式都保持不变。

但是，在使用 Mnesia 时，即使集群的大多数节点已关闭或无法访问，也可能声明一个队列，而使用 Khepri 时，相同的操作将超时。这样，客户端可以对问题做出适当的反应，而不是希望 Mnesia 之上实现的各种网络分区恢复策略能够顺利解决问题。

以下是在仅有 5 个节点中的 1 个节点正在运行的集群中，PerfTest 工具尝试声明其所需的交换机和队列的示例

./scripts/PerfTest

# => id: test-161339-979, starting consumer #0
# => id: test-161339-979, starting consumer #0, channel #0
# =>  Main thread caught exception: java.io.IOException
# =>  16:14:10.638 [com.rabbitmq.perf.PerfTest.main()] ERROR com.rabbitmq.perf.PerfTest - Main thread caught exception
# =>  (...)

同时，RabbitMQ 节点记录了以下消息

[error] <0.1373.0> Error on AMQP connection <0.1373.0> (127.0.0.1:55165 -> 127.0.0.1:5672 - perf-test-consumer-0, vhost: '/', user: 'guest', state: running), channel 1:
[error] <0.1373.0>  operation exchange.declare caused a connection exception internal_error: "failed to declare exchange 'direct' in vhost '/' because the operation timed out"