基于ScyllaDB与Clean Architecture构建CQRS系统：从Jib容器化后端到React Native客户端的实现

后端架构

文章字数: 3.8k

阅读时长: 17 分

一个系统需要每分钟处理数百万次设备状态的写入请求，同时为数千个移动客户端提供亚秒级的仪表盘查询响应。在这样的需求下，传统的CRUD架构会不可避免地在主数据存储上引发写密集型操作与读密集型操作之间的竞争。这种竞争最终会演变成性能瓶颈，无论如何优化单一数据模型，都无法同时满足两种极端且对立的需求。这是架构选型的起点，也是我们放弃传统模型的直接原因。

架构决策：单一模型优化 vs. CQRS

在项目初期，我们评估了两种核心架构方案，以应对这种高吞吐、低延迟的混合负载场景。

方案A：基于ScyllaDB的单一高度优化模型

这个方案主张使用单一数据模型，但利用ScyllaDB的特性进行深度优化。例如，设计一个“大宽表”，将所有可能被查询的数据冗余存储。

优点:
1. 概念简单: 维护单一数据源，应用层逻辑相对直接，没有数据同步的复杂性。
2. 开发效率: 初期开发速度快，对于熟悉传统数据建模的团队来说，上手门槛较低。
缺点:
1. 模型僵化: 单一模型必须同时服务于写入和读取。一个为快速写入而设计的Schema（例如，按设备ID和时间戳分区，以实现顺序写入）在执行聚合查询（例如，查询特定区域内所有活跃设备）时会非常低效。
2. 查询性能妥协: 为了支持多维度的查询，需要创建多个二级索引或物化视图。在ScyllaDB这种写优化的数据库中，过多的索引会显著增加写操作的开销和延迟，这与我们的高吞t吐写入要求背道而驰。
3. 扩展性受限: 当新的查询需求出现时，很可能需要修改核心表的结构或增加更多索引，每次变更都伴随着数据迁移和性能回归测试的风险。

方案B：命令查询职责分离 (CQRS)

该方案彻底将读、写操作在模型层面进行分离。

命令端 (Command Side): 负责处理状态变更请求（如UpdateDeviceLocation）。其数据模型只为快速、原子化的写入服务。这个模型可以非常简单，甚至是追加式的事件日志。
查询端 (Query Side): 负责处理数据查询请求（如GetDevicesInRegion）。其数据模型是为特定查询场景预先优化的“物化视图”或“投影”。一个命令可能会触发多个不同查询模型的更新。
优点:
1. 独立的优化路径: 写入模型可以专注于写入性能，无需考虑任何读取需求。读取模型可以高度非规范化，为特定的UI界面或API端点量身定制，从而实现极致的查询速度。
2. 技术栈灵活性: 读写两端甚至可以使用不同的存储技术。例如，写入端使用ScyllaDB，读取端根据需求可以使用Elasticsearch进行全文搜索，或使用Redis进行热点数据缓存。
3. 清晰的扩展性: 增加新的查询需求，只需要创建一个新的读取模型和相应的更新逻辑，完全不影响现有的写入逻辑和其他查询逻辑。
缺点:
1. 架构复杂性增加: 需要引入一种机制来确保写入模型的变化能最终同步到读取模型。这通常意味着引入事件总线、消息队列或至少是进程内的事件分发机制。
2. 最终一致性: 读取模型的数据相对于写入模型存在一个微小的延迟窗口。这要求业务和产品层面必须接受这种数据并非100%实时。
3. 开发成本: 需要设计和维护多个数据模型，代码量和认知负荷更高。

最终决策

对于我们面临的场景，读写负载的矛盾是核心问题。方案A的妥协会在系统规模扩大后迅速演变成性能瓶颈。因此，我们选择方案B：CQRS。其带来的性能和扩展性优势，足以证明其增加的复杂性是值得的。我们接受最终一致性，因为对于设备监控仪表盘而言，秒级的延迟是完全可以接受的。

系统实现：遵循整洁架构 (Clean Architecture)

为了控制CQRS带来的复杂性，我们采用整洁架构作为代码组织的指导原则。这确保了业务逻辑与基础设施（数据库、框架）的解耦，使得系统更易于测试和维护。

graph TD
    subgraph "React Native Client"
        A[UI Components] --> B(API Service Layer)
    end

    subgraph "Java Backend"
        B -- HTTP/WebSocket --> C{Application Layer: Controllers}

        subgraph "Clean Architecture Layers"
            C -- Invokes --> D(Use Cases / Interactors)
            D -- Depends on --> E(Domain Entities / Events)
            D -- Uses Port --> F(Repositories / Event Bus Interfaces)
        end

        subgraph "Infrastructure Layer"
            G[ScyllaDB Command Repository] -- Implements --> F
            H[ScyllaDB Query Repository] -- Implements --> F
            I[In-Process Event Bus] -- Implements --> F
            C -- Uses --> I
        end
    end

    subgraph "Data Stores (ScyllaDB)"
        J[Write Model: device_events]
        K[Read Model: current_device_locations]
        G --> J
        H --> K
        I -- Triggers Update --> K
    end

项目结构如下 (Maven)：

fleet-monitoring/
├── pom.xml
├── core/                    # 核心领域模型和用例，无任何框架依赖
│   ├── src/main/java/com/fleet/core/
│   │   ├── command/         # 命令对象
│   │   ├── query/           # 查询对象
│   │   ├── domain/          # 实体和值对象 (e.g., Device)
│   │   ├── event/           # 领域事件 (e.g., DeviceLocationUpdated)
│   │   └── usecase/         # 用例/交互器 (Command & Query Handlers)
├── application/             # 应用层，编排和入口
│   ├── src/main/java/com/fleet/application/
│   │   ├── controller/      # REST API Controllers
│   │   └── bus/             # 命令和查询总线的实现
├── infrastructure/          # 基础设施，具体实现
│   ├── src/main/java/com/fleet/infrastructure/
│   │   ├── scylladb/        # ScyllaDB的Repository实现
│   │   ├── event/           # 进程内事件总线实现
│   │   └── config/          # Spring Boot配置
└── main-app/                # Spring Boot启动模块

命令路径的实现

命令路径的目标是快速、可靠地接收状态变更并持久化。

1. ScyllaDB 写入模型设计

我们设计一个事件溯源风格的表，只追加，不更新。这对于ScyllaDB来说极其高效。

-- Keyspace
CREATE KEYSPACE IF NOT EXISTS fleet_monitoring
WITH REPLICATION = { 'class' : 'NetworkTopologyStrategy', 'datacenter1' : 3 };

-- Write Model Table for device location events
CREATE TABLE IF NOT EXISTS fleet_monitoring.device_location_events (
    device_id UUID,
    event_time TIMESTAMP,
    latitude DOUBLE,
    longitude DOUBLE,
    speed DOUBLE,
    -- Idempotency key
    command_id UUID,
    PRIMARY KEY (device_id, event_time)
) WITH CLUSTERING ORDER BY (event_time DESC);

分区键 device_id: 所有来自同一设备的数据都落在同一个分区，便于查询单个设备的历史轨迹。
聚类键 event_time DESC: 事件按时间倒序排列，获取最新位置非常快。
command_id: 用于实现接口的幂等性，防止网络重试导致重复数据。

2. Java 实现 (核心代码)

首先是命令对象和领域事件，它们是纯粹的POJO，位于core模块。

// In: core/src/main/java/com/fleet/core/command/UpdateDeviceLocationCommand.java
public record UpdateDeviceLocationCommand(
    UUID commandId, // For idempotency
    UUID deviceId,
    double latitude,
    double longitude,
    double speed,
    Instant timestamp
) {}

// In: core/src/main/java/com/fleet/core/event/DeviceLocationUpdated.java
public record DeviceLocationUpdated(
    UUID deviceId,
    double latitude,
    double longitude,
    double speed,
    Instant timestamp
) {}

命令处理器负责业务逻辑验证和持久化。

// In: core/src/main/java/com/fleet/core/usecase/UpdateDeviceLocationUseCase.java
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class UpdateDeviceLocationUseCase {
    private static final Logger log = LoggerFactory.getLogger(UpdateDeviceLocationUseCase.class);

    private final DeviceEventRepository deviceEventRepository;
    private final EventPublisher eventPublisher;

    public UpdateDeviceLocationUseCase(DeviceEventRepository repository, EventPublisher publisher) {
        this.deviceEventRepository = repository;
        this.eventPublisher = publisher;
    }

    public void handle(UpdateDeviceLocationCommand command) {
        // A common error is to place business validation in controllers.
        // It belongs here, in the use case layer.
        if (command.latitude() < -90 || command.latitude() > 90 ||
            command.longitude() < -180 || command.longitude() > 180) {
            log.warn("Invalid coordinates for device {}: lat={}, lon={}", command.deviceId(), command.latitude(), command.longitude());
            throw new IllegalArgumentException("Invalid geographic coordinates.");
        }

        // The repository interface is defined in core, implementation in infrastructure
        // This makes the core logic independent of ScyllaDB.
        try {
            // The repository implementation should handle idempotency using commandId
            deviceEventRepository.save(command);
        } catch (Exception e) {
            log.error("Failed to persist location update for device {}", command.deviceId(), e);
            // In a real project, this would involve more sophisticated error handling,
            // possibly queuing the command for retry.
            throw new RuntimeException("Persistence failed", e);
        }

        // Publish an event for the query side to consume
        DeviceLocationUpdated event = new DeviceLocationUpdated(
            command.deviceId(),
            command.latitude(),
            command.longitude(),
            command.speed(),
            command.timestamp()
        );
        eventPublisher.publish(event);
    }
}

ScyllaDB的Repository实现在infrastructure模块。

// In: infrastructure/src/main/java/com/fleet/infrastructure/scylladb/ScyllaDeviceEventRepository.java
import com.datastax.oss.driver.api.core.CqlSession;
import com.datastax.oss.driver.api.core.cql.BoundStatement;
import com.datastax.oss.driver.api.core.cql.PreparedStatement;
import org.springframework.stereotype.Repository;

@Repository
public class ScyllaDeviceEventRepository implements DeviceEventRepository {
    private final CqlSession session;
    private final PreparedStatement insertStatement;

    public ScyllaDeviceEventRepository(CqlSession cqlSession) {
        this.session = cqlSession;
        // Preparing statements is crucial for performance in ScyllaDB/Cassandra
        this.insertStatement = session.prepare(
            "INSERT INTO fleet_monitoring.device_location_events " +
            "(device_id, event_time, latitude, longitude, speed, command_id) " +
            "VALUES (?, ?, ?, ?, ?, ?) IF NOT EXISTS" // Using IF NOT EXISTS for idempotency
        );
    }

    @Override
    public void save(UpdateDeviceLocationCommand command) {
        BoundStatement bound = insertStatement.bind(
            command.deviceId(),
            command.timestamp(),
            command.latitude(),
            command.longitude(),
            command.speed(),
            command.commandId()
        );
        // The IF NOT EXISTS clause makes the write a lightweight transaction,
        // which has a performance cost. An alternative for very high throughput
        // is to handle deduplication in a separate batch process or accept rare duplicates.
        session.execute(bound);
    }
}

查询路径的实现

查询路径的目标是极速响应。数据模型是为特定查询场景而生的。

1. ScyllaDB 读取模型设计

假设我们的React Native客户端有一个地图视图，需要查询某个地理区域内的所有设备的最新位置。

-- Read Model Table optimized for spatial queries
-- In a real-world scenario, a geospatial index or bucketing would be used.
-- For simplicity, we'll use a 'region' string as a shard key.
CREATE TABLE IF NOT EXISTS fleet_monitoring.current_device_locations_by_region (
    region TEXT, -- e.g., "us-west-1a", calculated from lat/lon
    device_id UUID,
    last_seen TIMESTAMP,
    latitude DOUBLE,
    longitude DOUBLE,
    speed DOUBLE,
    PRIMARY KEY (region, device_id)
);

分区键 region: 这是一个分片键。所有在同一区域的设备都在一个分区中，这使得SELECT * FROM ... WHERE region = ?的操作非常高效，因为它只命中单个节点（或副本集）。
device_id 作为聚类键，确保了每个区域内设备记录的唯一性。
这个表只存储每个设备的最新状态，通过UPSERT（INSERT语句在ScyllaDB中兼具UPDATE功能）来维护。

2. 事件监听与读取模型更新

当命令端发布DeviceLocationUpdated事件后，一个监听器会负责更新读取模型。

// In: infrastructure/src/main/java/com/fleet/infrastructure/event/ReadModelUpdater.java
import org.springframework.context.event.EventListener;
import org.springframework.stereotype.Component;

@Component
public class ReadModelUpdater {
    private final CurrentDeviceLocationRepository locationRepository;

    public ReadModelUpdater(CurrentDeviceLocationRepository repo) {
        this.locationRepository = repo;
    }

    @EventListener
    public void handleDeviceLocationUpdate(DeviceLocationUpdated event) {
        // This is a simplification. A real geo-hashing library
        // (like H3 or S2) would be used to calculate the region.
        String region = calculateRegion(event.latitude(), event.longitude());

        CurrentDeviceLocation location = new CurrentDeviceLocation(
            region,
            event.deviceId(),
            event.timestamp(),
            event.latitude(),
            event.longitude(),
            event.speed()
        );

        // This call is an UPSERT, updating the device's latest known location
        // in the read-optimized table.
        locationRepository.save(location);
    }

    private String calculateRegion(double latitude, double longitude) {
        // Dummy implementation for demonstration
        if (latitude > 0) {
            return "northern-hemisphere";
        } else {
            return "southern-hemisphere";
        }
    }
}

3. Java 查询实现

查询处理器直接从读取模型中获取数据，逻辑非常简单。

// In: core/src/main/java/com/fleet/core/usecase/GetDevicesInRegionUseCase.java
public class GetDevicesInRegionUseCase {

    private final CurrentDeviceLocationRepository repository;

    public GetDevicesInRegionUseCase(CurrentDeviceLocationRepository repository) {
        this.repository = repository;
    }

    public List<DeviceLocationDto> handle(GetDevicesInRegionQuery query) {
        // The repository fetches data directly from the read-optimized table.
        // The logic is trivial, which is the entire point of a pre-calculated read model.
        return repository.findByRegion(query.region()).stream()
            .map(this::toDto)
            .collect(Collectors.toList());
    }

    private DeviceLocationDto toDto(CurrentDeviceLocation entity) {
        // Map from persistence model to a data transfer object for the API
        return new DeviceLocationDto(...);
    }
}

使用 Jib 进行容器化

传统的Dockerfile方法需要一个Docker守护进程，并且构建过程可能很慢。Jib是一个Maven/Gradle插件，可以直接将Java应用构建成OCI兼容的容器镜像，无需Docker，并且构建速度更快。这是一个典型的资深工程师会做出的务实选择，旨在优化CI/CD流水线。

在main-app模块的pom.xml中配置Jib插件：

<plugin>
    <groupId>com.google.cloud.tools</groupId>
    <artifactId>jib-maven-plugin</artifactId>
    <version>3.4.0</version>
    <configuration>
        <from>
            <!-- A production-grade base image -->
            <image>eclipse-temurin:17-jre-focal</image>
        </from>
        <to>
            <!-- Target image repository -->
            <image>docker.io/your-repo/fleet-monitoring-service</image>
            <tags>
                <tag>${project.version}</tag>
                <tag>latest</tag>
            </tags>
        </to>
        <container>
            <!-- Production JVM options are critical -->
            <jvmFlags>
                <jvmFlag>-server</jvmFlag>
                <jvmFlag>-Xms1g</jvmFlag>
                <jvmFlag>-Xmx1g</jvmFlag>
                <jvmFlag>-XX:+UseG1GC</jvmFlag>
                <jvmFlag>-Djava.security.egd=file:/dev/./urandom</jvmFlag>
            </jvmFlags>
            <mainClass>com.fleet.main.FleetMonitoringApplication</mainClass>
        </container>
    </configuration>
</plugin>

只需运行mvn compile jib:build，Jib就会自动将应用分层打包并推送到指定的镜像仓库。这里的关键在于，它将依赖、资源和类文件分到不同的层，当代码变更时，只需重新构建和推送最小的类文件层，CI构建速度得到极大提升。

React Native 客户端集成

客户端与后端的交互严格遵循命令和查询的分离。

发送命令: 用户在地图上移动一个设备图标后，客户端会发送一个HTTP POST请求。
获取数据: 仪表盘界面会定期轮询或通过WebSocket/SSE连接来获取指定区域的设备数据。

一个简化的React Native服务层代码片段：

// services/api.js
import axios from 'axios';
import { v4 as uuidv4 } from 'uuid'; // For generating command IDs

const API_BASE_URL = 'https://api.yourdomain.com';

// A command function: fire-and-forget, expecting a 2xx response
export const updateDeviceLocation = async (deviceId, { latitude, longitude, speed }) => {
  const command = {
    commandId: uuidv4(), // Client generates idempotency key
    deviceId,
    latitude,
    longitude,
    speed,
    timestamp: new Date().toISOString(),
  };

  try {
    // Commands are typically sent via POST, PUT, or DELETE
    await axios.post(`${API_BASE_URL}/devices/${deviceId}/location`, command);
    console.log('Location update command sent successfully.');
  } catch (error) {
    // In a real app, this should include user-facing error handling
    console.error('Failed to send location update command:', error);
    throw error;
  }
};

// A query function: fetches data from a read-optimized endpoint
export const getDevicesInRegion = async (region) => {
  try {
    // Queries are always done via GET
    const response = await axios.get(`${API_BASE_URL}/devices/region/${region}`);
    // The response data is a DTO tailored for this view
    return response.data;
  } catch (error) {
    console.error(`Failed to fetch devices for region ${region}:`, error);
    return []; // Return a stable value on failure
  }
};

在React组件中使用这些服务：

// components/MapDashboard.js
import React, { useState, useEffect } from 'react';
import { getDevicesInRegion } from '../services/api';

const MapDashboard = ({ currentRegion }) => {
  const [devices, setDevices] = useState([]);
  const [isLoading, setIsLoading] = useState(true);

  useEffect(() => {
    const fetchData = async () => {
      setIsLoading(true);
      const deviceData = await getDevicesInRegion(currentRegion);
      setDevices(deviceData);
      setIsLoading(false);
    };

    // Poll for updates every 5 seconds
    const intervalId = setInterval(fetchData, 5000);
    fetchData(); // Initial fetch

    return () => clearInterval(intervalId); // Cleanup on unmount
  }, [currentRegion]);

  if (isLoading) {
    return <p>Loading map data...</p>;
  }

  // Render devices on a map component...
  return (
    <div>
      {/* ... Map rendering logic ... */}
    </div>
  );
};

架构的局限性与未来路径

当前这套架构并非银弹。它的主要局限性在于最终一致性。命令端写入成功后，查询端的数据更新存在一个延迟窗口（在本例中是事件处理的时间），这个延迟可能从几毫秒到几秒不等。对于需要强一致性读写的场景，例如金融交易，此架构并不适用。

其次，运维复杂性提高了。我们现在需要监控写入模型的健康状况、读取模型的健康状况以及两者之间的数据同步流程。任何一环出错都可能导致数据不一致。

当前的事件分发机制是进程内的，这意味着后端服务是单点。一个可行的优化路径是引入一个真正的消息队列（如Apache Kafka或Pulsar）来替代进程内事件总线。这将使命令服务和更新读取模型的服务可以独立部署和扩展，进一步提高系统的解耦程度和韧性。但同时，这也会引入一个新的基础设施依赖，增加了整体的维护成本。这些都是在架构演进中需要持续权衡的。

Jib React Native ScyllaDB Clean Architecture CQRS

构建实时软件供应链漏洞情报平台的架构权衡与实现

2023-10-27 架构设计

Material-UI WebSockets 依赖扫描 Pulsar Elasticsearch

构建基于Packer与Lua的ArangoDB图谱推荐引擎实时事件注入层

2023-10-27 数据工程与存储

Packer ArangoDB Lua 推荐系统