将分布式数据库实现到您的 Java 应用程序中

引言

i.o.cluster，也称为interference open cluster，是一个简单的Java框架，它允许您在Java应用程序中启动一个分布式数据库服务，使用类JPA的接口和注解进行结构映射和数据操作。该软件继承了interference project的名称，其机制在该项目中得到了发展。

i.o.cluster是一个开源的纯Java软件。

i.o.cluster服务的基本单元是节点——它可以是独立的运行服务，也可以是在某个Java应用程序中运行的服务。

每个i.o.cluster节点都有自己的持久化存储，并且可以被视为一个本地数据库，具有以下基本功能：

• 它使用简单的对象（POJOs）进行操作。
• 它使用基本的JPA注解（@Table、@Column、@Transient、@Index、@GeneratedValue）将对象直接映射到持久化存储。
• 它支持事务。
• 它支持具有读已提交隔离级别的SQL查询。
• 它使用持久化索引来快速访问数据并提高SQL JOIN的性能。
• 它允许灵活地管理内存中的数据，以确保节点在任何存储大小/可用内存的比例下都能稳定运行。

每个节点都包含几个确保其运行的机制：

• 核心算法（支持结构化持久化存储、索引、自定义序列化、堆管理、本地和分布式同步过程）。
• SQL和CEP处理器。
• 事件传输，用于节点之间以及节点与客户端应用程序之间的消息交换。

i.o.cluster实现了最简单的数据管理模型，该模型基于Session对象的几个标准类JPA方法：

• persist() - 将对象存入存储。
• find() - 按唯一标识符查找对象。
• execute() - 执行SQL查询。
• commit() - 提交事务。
• rollback() - 回滚事务。

此外，i.o.cluster软件还包含一个远程客户端，它可以通过内部事件传输功能远程连接到任何集群节点，并执行标准的类JPA命令（persist、find、execute、commit、rollback）。

 

 

分布式持久化模型

Interference open cluster是一个去中心化的系统。这意味着集群不使用任何协调节点；相反，每个节点都遵循一组正式的行为规则，这些规则在一定的交互框架内保证数据的完整性和可用性。

在这些规则的框架内，i.o.cluster的所有节点都是等效的。系统中没有主从节点之分——用户表中的更改可以从任何节点进行，并且所有更改都会复制到所有节点，无论更改是在哪个节点上进行的。

就事务而言，在本地用户会话中运行commit会自动确保更改的数据在集群的所有节点上可见。

要将一个节点包含到集群中，您必须在cluster.nodes配置参数中指定所有集群节点（不包括当前节点）的完整列表。

集群节点的最小数量是2，最大数量是64。

配置完成后，我们可以按任意顺序启动所有配置好的节点作为集群。所有节点将使用特定的消息（事件）来提供节点间数据一致性和水平扩展查询。

集群节点的最小数量是2，最大数量是64。

配置完成后，我们可以按任意顺序启动所有配置好的节点作为集群。所有节点将使用特定的消息（事件）来提供节点间数据一致性和水平扩展查询。

分发规则

Interference open cluster的概念基于一个简单的基本要求，该要求可以字面表达如下：我们必须允许从任何节点在集群级别插入和修改数据，并且我们必须允许从任何节点检索数据，尽可能多地利用整个集群的计算资源。此外，我们接受所有集群节点都必须健康且已通电的条件；如果任何节点暂时关闭，它将不会接收数据，直到其存储与其他节点同步。在实践中，在没有更改的时刻，这意味着集群节点上的存储具有相同的副本。为了防止集群中的更改冲突，使用了几种锁定模式：

• 表级别（节点上的会话锁定整个表）。
• 帧级别（节点上的会话锁定一个帧）。
• 不允许非所有者节点进行更改

这里需要更详细地解释：某个节点上的所有数据插入都执行到在该节点上分配的一个帧中，而该节点又是该帧的所有者。这样做是为了在多个节点同时向一个表插入数据时，避免在复制过程中发生冲突。之后，这种区别使我们能够判断是否需要在集群级别请求更改帧数据的权限。此外，它允许我们实现一种模式，即禁止在非所有者节点上对帧进行更改。当集群中的一个或多个其他节点不可用时（我们无法确定节点是否宕机或网络连接是否存在问题），集群节点就会使用此模式。

所以，让我们再重复一遍：

• 所有集群节点都应等效。
• 对任何节点上的所有更改会立即映射到其他节点。
• 数据插入执行到本地存储结构中，然后更改会复制到其他节点。
• 如果无法复制（节点不可用或连接中断），则会为此节点创建一个持久化更改队列。
• 任何数据帧的所有者是在该节点上分配该帧的节点。
• 对节点自己的数据帧的数据更改会立即执行，否则，将对集群级别的数据帧执行分布式锁定。
• 如果集群发生故障（某些节点离线或连接中断），则不允许进行任何数据更改，或者只允许对节点自己的数据帧进行更改。
• 集群使用为实体生成唯一标识符（@DistributedId 注解），以便标识符在集群内是唯一的，而不仅仅是在同一个节点内。
• 集群不使用任何额外的唯一性检查，这需要集群级别的锁定。

SQL水平扩展查询

在任何集群节点上调用的所有SQL查询都将自动分发到集群节点进行并行处理。节点会根据任务的体积（查询表的体积是否足够大等）来做出此决定。

如果在处理请求期间节点不可用（网络故障、服务停止），分发给该节点的任务将自动重新调度到另一个可用节点。

复杂事件处理概念

因此，我们必须允许从任何节点在集群级别插入和修改数据，并且我们必须允许从任何节点检索数据，尽可能多地利用整个集群的计算资源。

Interference open cluster的下一个概念是，任何表同时也是一个队列，特别是，使用SELECT STREAM子句，我们可以按照它们被添加的顺序检索记录。通常，在集群级别，session.persist()操作可以被视为发布一个持久化事件。根据我们的基本分发规则，我们将此事件发送到所有节点。

interference open cluster目前不支持标准的DMLUPDATE和DELETE操作。取而代之的是，对于批量表处理（包括可选的WHERE子句），我们实现了PROCESS和PROCESS STREAM子句，它们允许我们处理来自EventProcessor接口实现之一的选定记录。一方面，这种方法允许我们获得与使用UPDATE和DELETE相似的结果；另一方面，它极大地扩展了自定义处理记录的可能性，允许进行完整的事件处理。为了公平起见，需要注意的是，您可以使用标准的SELECT和SELECT STREAM，并使用一些自定义代码来处理结果集，从而获得类似的结果，但PROCESS和PROCESS STREAM是在集群的核心级别实现的，这大大提高了性能。其次，这些语句在集群级别启动，并提供了分布式事件处理的现成实现。

为了创建自定义的EventProcessor实现，我们需要实现两个方法：

boolean process(Object event)

在此方法中，应实现自定义事件处理，如果处理成功，则返回true。

boolean delete()

如果此方法返回true，则记录将在处理成功后（process方法返回true）从表中删除。

接下来，我们可以使用以下查询：

PROCESS fully_qualified_class_name alias

WITHIN fully_qualified_event_processor_class_name

[WHERE condition1 AND/OR condition2 … ]

[ORDER BY alias.column_name … ]

例如，它可能看起来像这样：

String sql = "process su.interference.entity.SomeEvent d within su.interference.processor.SomeEventProcessor where d.eventType = 1";

ResultSet rs = s.execute(sql);

 

PROCESS语句允许以批量模式处理一个特定表中的记录，目前该查询不支持与任何其他表的JOIN。PROCESS语句是一个分布式操作，并在集群的所有节点上执行处理，在此期间，为了防止其他节点或会话启动任何其他PROCESS语句，它会在查询运行时在集群级别锁定该表。

此处理在事务中执行，因此，执行后，我们需要显式地应用commit或rollback。

在线复杂事件处理

除了标准的批量处理外，interference open cluster还支持在线复杂事件处理，使用SQL语句中的SELECT STREAM（以及用于使用EventProcessor实现的PROCESS STREAM）子句。

SELECT STREAM查询支持三种CEP模式：

• 事件按原样处理，无需任何聚合。
• 使用任何分组函数（滚动窗口）按列值对事件进行聚合。
• 一些窗口会对每个新记录进行事件聚合（滑动窗口）。

流式查询与普通查询的基本区别如下：

• execute()方法返回一个StreamQueue对象。
• 请求异步执行，直到调用StreamQueue.stop()方法或直到应用程序终止。
• StreamQueue.poll()方法返回表中先前插入的所有记录，并根据WHERE条件（如果存在），并且继续返回新添加的记录。
• 每个StreamQueue.poll()方法始终返回会话中最后一个已轮询位置之后的下一条记录，因此，如果SQL请求停止并在同一会话中再次调用，数据将从最后一个固定位置继续检索；而在另一个会话中，数据将从表的开头检索。
• 与普通查询不同，流式查询不支持事务，并且始终返回实际插入的行，而不考虑会话中是否使用了commit()方法（脏读）。

更多信息

• GitHub：https://github.com/interference-project/interference
• 更多信息：http://io.digital