使用MongoDB与Jakarta EE实现批处理：从客户分群到可扩展工作流

批处理解决了企业中的一个常见需求：可靠地将业务规则应用于大型数据集，而不依赖于单个用户请求。诸如重新计算客户细分、迁移记录、核对交易、更新价格或处理数百万个文档等任务，需要的不仅仅是基本的循环。应用程序必须高效地控制内存、处理故障、最小化不必要的写入，并将长时间运行的进程与交互式用户体验分离。

MongoDB 和 Jakarta Batch 可以很好地协同工作来满足这些需求。在本文中，我们将开发一个客户细分工作流，允许管理员调整 Bronze、Silver、Gold 和 Platinum 客户的消费阈值，预览影响，并启动批处理作业以仅重新分类必要的记录。我们的重点超越批处理 API，着眼于设计一个清晰、可扩展的工作流，将业务决策与受控的数据处理连接起来。

本教程介绍 Batch 模式，并演示如何使用 MongoDB 实现它。

在本教程中，您将：

为电子商务应用程序建模客户细分系统。
使用 Java 建模并与 MongoDB 交互。
使用 Jakarta Faces Flow 和 Jakarta Batch 构建批处理工作流，以预览并应用客户重新分类规则。

您可以在 GitHub 仓库中找到本教程中展示的所有代码：

[email protected]:soujava/mongodb-jakarta-batch.git

先决条件

本教程需要：

Java 21。
Maven。
一个 MongoDB 集群。
- MongoDB Atlas（选项 1）

批处理将大量记录作为受控工作负载进行管理，而不是作为单个冗长的请求。工作负载不是在一个 Web 请求中处理所有数据，而是被划分为可重复的单元。典型的批处理管道读取项目、应用业务逻辑、将结果分组为块，并将这些块写入数据库。这种方法提供了用于处理检查点、故障、重试、过滤和可重启性的不同点，同时还控制内存使用和数据库写入。

在 Jakarta EE 中，Jakarta Batch 使用作业、步骤和面向块的执行来标准化此模型。一个作业由一个或多个步骤组成，而块步骤通常链接 ItemReader、ItemProcessor 和 ItemWriter。读取器提供数据，处理器应用业务规则，写入器持久化更新后的项目。在我们的示例中，Jakarta Faces 收集新的客户细分阈值，而 Jakarta Batch 管理与 HTTP 请求无关的重新分类 MongoDB 客户的长期运行任务。 [LOADING...]

本教程使用一个实际的电子商务场景，涵盖使用 Jakarta EE 和 MongoDB 进行批处理。我们将展示 Jakarta Batch 如何跨许多记录管理操作，同时保持简单、响应迅速的用户体验。具体来说，我们将构建一个客户仪表板，根据可配置的消费阈值将用户分为 Bronze、Silver、Gold 和 Platinum 类别。

从仪表板中，管理员可以查看客户分布、调整类别阈值、预览更改的影响并启动批处理过程。然后，Jakarta Batch 将从 MongoDB 读取数据、应用新规则，并仅更新类别发生变化的客户，从而重新分类客户。

步骤 1：生成项目

本教程的第一步是创建初始 Maven 结构。您可以使用 Eclipse Starter for Jakarta EE。有关此步骤的更多详细信息，请参阅上一教程。

我们将在 pom.xml 中包含 MongoDB 驱动程序和 PrimeFaces。您可以在此处看到最终结果：

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
                             https://maven.apache.org/xsd/maven-4.0.0.xsd">

    <modelVersion>4.0.0</modelVersion>

    <groupId>expert.os.demos</groupId>
    <artifactId>ecommerce-customer-segmentation</artifactId>
    <version>1.0.0-SNAPSHOT</version>
    <packaging>war</packaging>

    <name>E-Commerce Customer Segmentation</name>
    <description>
        E-commerce application that reclassifies customers into
        Bronze, Silver, Gold, and Platinum tiers based on configurable
        spending thresholds using Jakarta EE and MongoDB.
    </description>

    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>

        <maven.compiler.release>21</maven.compiler.release>

        <jakartaee-api.version>11.0.0</jakartaee-api.version>
        <jnosql.version>1.1.17</jnosql.version>
        <primefaces.version>16.0.0</primefaces.version>
        <junit.version>5.10.5</junit.version>
        <yasson.version>3.0.4</yasson.version>

        <compiler-plugin.version>3.15.0</compiler-plugin.version>
        <surefire-plugin.version>3.5.6</surefire-plugin.version>
        <war-plugin.version>3.5.1</war-plugin.version>

        <glassfish-plugin.version>8.0</glassfish-plugin.version>
        <glassfish.version>8.0.4</glassfish.version>
    </properties>

    <dependencies>

        <dependency>
            <groupId>jakarta.platform</groupId>
            <artifactId>jakarta.jakartaee-api</artifactId>
            <version>${jakartaee-api.version}</version>
            <scope>provided</scope>
        </dependency>

        <dependency>
            <groupId>org.eclipse.jnosql.databases</groupId>
            <artifactId>jnosql-mongodb</artifactId>
            <version>${jnosql.version}</version>
        </dependency>

        <dependency>
            <groupId>org.primefaces</groupId>
            <artifactId>primefaces</artifactId>
            <version>${primefaces.version}</version>
        </dependency>

        <dependency>
            <groupId>org.junit.jupiter</groupId>
            <artifactId>junit-jupiter</artifactId>
            <version>${junit.version}</version>
            <scope>test</scope>
        </dependency>

        <dependency>
            <groupId>org.eclipse</groupId>
            <artifactId>yasson</artifactId>
            <version>${yasson.version}</version>
            <scope>test</scope>
        </dependency>

    </dependencies>

    <build>
        <finalName>ecommerce-customer-segmentation</finalName>

        <plugins>

            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>${compiler-plugin.version}</version>
            </plugin>

            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-war-plugin</artifactId>
                <version>${war-plugin.version}</version>
            </plugin>

            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-surefire-plugin</artifactId>
                <version>${surefire-plugin.version}</version>
            </plugin>

            <plugin>
                <groupId>org.glassfish.embedded</groupId>
                <artifactId>embedded-glassfish-maven-plugin</artifactId>
                <version>${glassfish-plugin.version}</version>

                <configuration>
                    <app>target/ecommerce-customer-segmentation.war</app>
                    <glassfish.version>${glassfish.version}</glassfish.version>
                    <contextRoot>/</contextRoot>
                    <port>8080</port>
                </configuration>
            </plugin>

        </plugins>
    </build>

</project>

下一步是更新数据库中的初始配置，在 src/main/resources/META-INF/microprofile-config.properties 中，您将包含数据库配置和属性。

jnosql.document.database=ecommerce
jnosql.mongodb.url=mongodb+srv://admin:<db_password>@cluster0.gblhb3d.mongodb.net/?appName=devrel-article-java-jnosql
jnosql.mongodb.application.name=devrel-article-java-jnosql

数据库配置完成后，我们现在可以创建 Customer 实体。该实体将包含以下属性：id、name、totalSpent 和 tier。tier 属性将使用枚举。

@Entity
public class Customer {

    @Id
    private String id;

    @Column
    private String name;

    @Column
    private BigDecimal totalSpent;

    @Column
    private CustomerTier tier;
//builder
//getter and setter
}

public enum CustomerTier {
    BRONZE,
    SILVER,
    GOLD,
    PLATINUM
}

Java 和 MongoDB 使用 Jakarta Data Repository 进行通信，它提供了一个简单的集成接口。无需额外配置。

@Repository
public interface CustomerRepository extends BasicRepository<Customer, String> {
}
```## 步骤 2：创建批处理结构

本教程讲解 Jakarta Batch 的结构、执行模型以及各组件的职责。在启动作业之前，需要明确批处理过程所需的信息。在客户分层的示例中，这些信息指的是将客户划分为 Bronze、Silver、Gold 和 Platinum 等级别的消费阈值。

每个阈值由最低消费金额和对应的客户等级组成。分层策略将这些阈值组合在一起，并可以序列化为 JSON。这个序列化后的策略被用作作业参数，使得批处理作业能够在不修改作业实现的情况下依据不同的业务规则运行。

```java
public record SegmentationThreshold(
        BigDecimal minimumValue,
        CustomerTier tier) {
}

public record CustomerSegmentationPolicy(
        List<SegmentationThreshold> thresholds) {
}

定义策略之后，CustomerSegmentationService 充当批处理操作的入口。它的主要作用是接收选定的策略，将其序列化为作业参数，并指示 Jakarta Batch 运行时启动 customer-segmentation 作业。

$ java

package expert.os.demos.ecommerce.batch;

import jakarta.annotation.PostConstruct;
import jakarta.batch.operations.JobOperator;
import jakarta.batch.operations.NoSuchJobException;
import jakarta.batch.runtime.BatchRuntime;
import jakarta.batch.runtime.BatchStatus;
import jakarta.batch.runtime.JobExecution;
import jakarta.batch.runtime.JobInstance;
import jakarta.enterprise.context.ApplicationScoped;

import java.io.IOException;
import java.io.InputStream;
import java.io.UncheckedIOException;
import java.nio.charset.StandardCharsets;
import java.util.Comparator;
import java.util.List;
import java.util.Optional;
import java.util.Properties;

@ApplicationScoped
public class CustomerSegmentationService {

    public static final String JOB_NAME = "customer-segmentation";
    private static final String DEFAULT_THRESHOLDS = "/segmentation-thresholds.json";

    private volatile CustomerSegmentationPolicy currentPolicy;

    @PostConstruct
    void initialize() {
        currentPolicy = loadDefaultPolicy();
    }

    public CustomerSegmentationPolicy currentPolicy() {
        return currentPolicy;
    }

    public long start(CustomerSegmentationPolicy policy) {
        if (isRunning()) {
            throw new IllegalStateException("A customer segmentation batch is already running");
        }

        Properties parameters = new Properties();
        parameters.setProperty(CustomerSegmentationPolicy.JOB_PARAMETER, policy.toJson());

        long executionId = jobOperator().start(JOB_NAME, parameters);
        currentPolicy = policy;
        return executionId;
    }

    public Optional<BatchStatus> latestStatus() {
        JobOperator operator = jobOperator();

        try {
            List<JobInstance> instances = operator.getJobInstances(JOB_NAME, 0, 1);
            if (instances.isEmpty()) {
                return Optional.empty();
            }

            return operator.getJobExecutions(instances.getFirst()).stream()
                    .max(Comparator.comparing(JobExecution::getCreateTime))
                    .map(JobExecution::getBatchStatus);
        } catch (NoSuchJobException exception) {
            return Optional.empty();
        }
    }

    public boolean isRunning() {
        try {
            return !jobOperator().getRunningExecutions(JOB_NAME).isEmpty();
        } catch (NoSuchJobException exception) {
            return false;
        }
    }

    private CustomerSegmentationPolicy loadDefaultPolicy() {
        try (InputStream stream =
                     CustomerSegmentationService.class.getResourceAsStream(DEFAULT_THRESHOLDS)) {

            if (stream == null) {
                throw new IllegalStateException("Resource not found: " + DEFAULT_THRESHOLDS);
            }

            String json = new String(stream.readAllBytes(), StandardCharsets.UTF_8);
            return CustomerSegmentationPolicy.fromJson(json);
        } catch (IOException exception) {
            throw new UncheckedIOException(
                    "Unable to load default segmentation thresholds", exception);
        }
    }

    private JobOperator jobOperator() {
        return BatchRuntime.getJobOperator();
    }
}

作业启动后，分块处理中的第一个批处理组件是 ItemReader。它一次向批处理运行时提供一个条目。在本应用中，CustomerItemReader 从仓库中检索客户，按客户 ID 排序，并通过 readItem() 顺序提供这些客户。

该读取器还演示了检查点（checkpointing），这是 Jakarta Batch 的一个关键概念。checkpointInfo() 返回当前位置，使运行时能够在分块边界处跟踪进度，并在需要时恢复执行。在本演示中，读取器将所有客户加载到内存中。对于更大的数据集，可以在不改变 Reader -> Processor -> Writer 模型的前提下，采用分页或基于游标的访问方式进行调整。

$ java

package expert.os.demos.ecommerce.batch;

import expert.os.demos.ecommerce.Customer;
import expert.os.demos.ecommerce.CustomerRepository;
import jakarta.batch.api.chunk.AbstractItemReader;
import jakarta.enterprise.context.Dependent;
import jakarta.inject.Inject;
import jakarta.inject.Named;

import java.io.Serializable;
import java.util.Comparator;
import java.util.List;
import java.util.logging.Logger;
import java.util.stream.Stream;

@Named("customerItemReader")
@Dependent
public class CustomerItemReader extends AbstractItemReader {

    private static final Logger LOGGER = Logger.getLogger(CustomerItemReader.class.getName());

    @Inject
    private CustomerRepository customerRepository;

    private List<Customer> customers = List.of();
    private int nextIndex;

    @Override
    public void open(Serializable checkpoint) {
        try (Stream<Customer> customerStream = customerRepository.findAll()) {
            customers = customerStream
                    .sorted(Comparator.comparing(Customer::getId))
                    .toList();
        }

        nextIndex = checkpoint instanceof Integer index ? index : 0;

        if (nextIndex < 0 || nextIndex > customers.size()) {
            throw new IllegalArgumentException("Invalid customer reader checkpoint: " + nextIndex);
        }

        LOGGER.info(() -> "Customer reader opened: total=%d, nextIndex=%d"
                .formatted(customers.size(), nextIndex));
    }

    @Override
    public Customer readItem() {
        if (nextIndex >= customers.size()) {
            return null;
        }

        return customers.get(nextIndex++);
    }

    @Override
    public Serializable checkpointInfo() {
        return nextIndex;
    }
}

ItemProcessor 负责业务转换。Jakarta Batch 将读取器提供的每个客户发送给 CustomerTierProcessor，后者使用作业启动时提供的策略计算合适的等级。Jakarta Batch 通过 @BatchProperty 将策略作为序列化后的作业参数注入，并在处理器初始化时重新构造它。对于每个客户，处理器会比较计算出的等级与现有等级。如果等级未变，则返回 null 以过滤掉该条目。如果等级发生变化，则返回一个带有更新分类的新客户对象。

$ java

package expert.os.demos.ecommerce.batch;

import expert.os.demos.ecommerce.Customer;
import expert.os.demos.ecommerce.CustomerTier;
import jakarta.annotation.PostConstruct;
import jakarta.batch.api.BatchProperty;
import jakarta.batch.api.chunk.ItemProcessor;
import jakarta.enterprise.context.Dependent;
import jakarta.inject.Inject;
import jakarta.inject.Named;

import java.util.Objects;

@Named("customerTierProcessor")
@Dependent
public class CustomerTierProcessor implements ItemProcessor {

    @Inject
    @BatchProperty(name = CustomerSegmentationPolicy.JOB_PARAMETER)
    private String thresholdsJson;

    private CustomerSegmentationPolicy policy;

    public CustomerTierProcessor() {
    }

    CustomerTierProcessor(CustomerSegmentationPolicy policy) {
        this.policy = Objects.requireNonNull(policy);
    }

    @PostConstruct
    void initialize() {
        policy = CustomerSegmentationPolicy.fromJson(thresholdsJson);
    }

    @Override
    public Customer processItem(Object item) {
        if (!(item instanceof Customer customer)) {
            throw new IllegalArgumentException("Expected a Customer item");
        }

        CustomerTier calculatedTier = policy.tierFor(customer.getTotalSpent());
        if (calculatedTier == customer.getTier()) {
            return null;
        }

        return Customer.builder()
                .id(customer.getId())
                .name(customer.getName())
                .totalSpent(customer.getTotalSpent())
                .tier(calculatedTier)
                .build();
    }
}

最后一个组件是 ItemWriter。与处理单个条目的读取器和处理器不同，写入器接收当前分块中累积的所有条目。在本应用中，这些条目仅是分类发生变化的客户，因为处理器会过滤掉未变化的客户。

CustomerItemWriter 将持久化工作委托给 CustomerRepository.saveAll()，使变更后的客户能够一起写入。这将持久化关注点从处理器中分离出来，并维持了 Jakarta Batch 模型中预期的职责分离。

$ java

package expert.os.demos.ecommerce.batch;

import expert.os.demos.ecommerce.Customer;
import expert.os.demos.ecommerce.CustomerRepository;
import jakarta.batch.api.chunk.AbstractItemWriter;
import jakarta.enterprise.context.Dependent;
import jakarta.inject.Inject;
import jakarta.inject.Named;

import java.util.List;
import java.util.logging.Logger;

@Named("customerItemWriter")
@Dependent
public class CustomerItemWriter extends AbstractItemWriter {

    private static final Logger LOGGER = Logger.getLogger(CustomerItemWriter.class.getName());

    @Inject
    private CustomerRepository customerRepository;

    @Override
    public void writeItems(List<Object> items) {
        List<Customer> customers = items.stream()
                .map(this::toCustomer)
                .toList();

        customerRepository.saveAll(customers);
        LOGGER.info(() -> "Updated %d customer tiers".formatted(customers.size()));
    }

    private Customer toCustomer(Object item) {
        if (item instanceof Customer customer) {
            return customer;
        }

        throw new IllegalArgumentException("Expected a Customer item");
    }
}

最后一个元素是作业定义。Jakarta Batch 使用 JSL XML 描述批处理组件如何连接以及运行时应如何执行它们。在本示例中，customer-segmentation 作业包含单个步骤，以每 20 个条目为一个分块处理客户。读取器提供客户，处理器应用分层策略，写入器持久化变更后的记录。作业定义将阈值传递给处理器，让执行架构保持声明式，同时业务逻辑仍保留在 Java 中。

位于 resources/META-INF/batch-jobs/customer-segmentation.xml 的 XML 充当批处理管道的编排层。它不包含分层规则；相反，它连接各个组件并定义执行行为。这使得职责划分清晰：Java 类实现读取、处理和写入逻辑，而 JSL 文件定义这些组件如何参与作业。

$ xml

<?xml version="1.0" encoding="UTF-8"?>
<job id="customer-segmentation"
    xmlns="https://jakarta.ee/xml/ns/jakartaee"
    version="2.0">
    <step id="recalculate-customer-tiers">
        <chunk item-count="20">
            <reader ref="customerItemReader"/>
            <processor ref="customerTierProcessor">
                <properties>
                    <property name="thresholds" value="#{jobParameters['thresholds']}"/>
                </properties>
            </processor>
            <writer ref="customerItemWriter"/>
        </chunk>
    </step>
</job>

下图展示了批处理工作过程和处理流程：

CustomerSegmentationService 启动 customer-segmentation 作业，并提供序列化后的阈值作为作业参数。
Jakarta Batch 加载 JSL 定义并启动 recalculate-customer-tiers 步骤。
CustomerItemReader 检索客户并将其逐个提供给批处理运行时。
CustomerTierProcessor 处理每个客户，并根据配置的阈值确定合适的等级。
如果等级保持不变，处理器返回 null，将该客户从写入阶段排除。
如果等级发生变化，更新后的客户将被包含在当前分块中。
作业配置了 item-count="20"，因此 Jakarta Batch 会在调用写入器写入该分块之前，最多处理并累积 20 个变更条目。
CustomerItemWriter 在写入阶段将变更后的客户保存到 MongoDB。
Jakarta Batch 对每个分块重复读取、处理和写入循环，直到所有客户处理完毕。

[LOADING...]

步骤 3：创建 Bean

下一步是将批处理流程连接到 Web 层。首页使用 @ViewScoped 的 Bean，因为其状态仅在仪表板视图活跃时需要。DashboardBean 通过 DashboardService 初始化仪表板状态，将其暴露给页面，并提供 refresh() 方法，使 UI 能够获取最新的客户分布和批处理状态，而无需将这些状态存储在用户会话中。

$ java

package expert.os.demos.ecommerce.web;

import jakarta.annotation.PostConstruct;
import jakarta.faces.view.ViewScoped;
import jakarta.inject.Inject;
import jakarta.inject.Named;

import java.io.Serial;
import java.io.Serializable;

@Named
@ViewScoped
public class DashboardBean implements Serializable {

    @Serial
    private static final long serialVersionUID = 1L;

    @Inject
    private DashboardService dashboardService;

    private DashboardState state;

    @PostConstruct
    public void initialize() {
        state = dashboardService.initializeState();
    }

    public void refresh() {
        state = dashboardService.currentState();
    }

    public DashboardState getState() {
        return state;
    }
}

分层向导遵循不同的生命周期。Jakarta Faces 提供了 @FlowScoped，用于多步骤交互场景——状态需要在多个视图之间保持，但在用户离开流程后应被丢弃。在这里，CustomerSegmentationFlow 在管理员依次经过 Configure（配置）、Preview（预览）和 Review（审查） 时保留阈值配置和预览状态，并且仅在最终确认后才启动批处理。

$ java

package expert.os.demos.ecommerce.web;

import jakarta.annotation.PostConstruct;
import jakarta.faces.application.FacesMessage;
import jakarta.faces.context.FacesContext;
import jakarta.faces.flow.FlowScoped;
import jakarta.inject.Inject;
import jakarta.inject.Named;

import java.io.Serial;
import java.io.Serializable;
import java.util.Objects;

@Named
@FlowScoped("customer-segmentation")
public class CustomerSegmentationFlow implements Serializable {

    @Serial
    private static final long serialVersionUID = 1L;

    @Inject
    private CustomerSegmentationFlowService flowService;

    private CustomerSegmentationFlowState state;

    public CustomerSegmentationFlow() {
    }

    CustomerSegmentationFlow(CustomerSegmentationFlowService flowService) {
        this.flowService = Objects.requireNonNull(flowService);
    }

    @PostConstruct
    public void initialize() {
        state = flowService.initializeState();
    }

    public String preview() {
        try {
            flowService.preview(state);
            return "preview";
        } catch (IllegalArgumentException | IllegalStateException exception) {
            addMessage(
                    FacesMessage.SEVERITY_ERROR,
                    "Invalid thresholds",
                    exception.getMessage());
            return null;
        }
    }

    public String review() {
        return "review";
    }

    public String execute() {
        try {
            long executionId = flowService.start(state);

            addMessage(
                    FacesMessage.SEVERITY_INFO,
                    "Segmentation started",
                    "Batch execution %d started successfully".formatted(executionId),
                    true);

            return "home";
        } catch (IllegalArgumentException | IllegalStateException exception) {
            addMessage(
                    FacesMessage.SEVERITY_ERROR,
                    "Unable to start segmentation",
                    exception.getMessage());
            return null;
        }
    }

    public CustomerSegmentationFlowState getState() {
        return state;
    }

    private void addMessage(
            FacesMessage.Severity severity,
            String summary,
            String detail) {

        addMessage(severity, summary, detail, false);
    }

    private void addMessage(
            FacesMessage.Severity severity,
            String summary,
            String detail,
            boolean keepAfterRedirect) {

        FacesContext context = currentFacesContext();

        if (context != null) {
            if (keepAfterRedirect) {
                context.getExternalContext()
                        .getFlash()
                        .setKeepMessages(true);
            }

            context.addMessage(
                    null,
                    new FacesMessage(severity, summary, detail));
        }
    }

    FacesContext currentFacesContext() {
        return FacesContext.getCurrentInstance();
    }
}

该流程定义在 src/main/webapp/customer-segmentation/customer-segmentation-flow.xml 中。该文件指定了向导的导航方式：configure 是入口点，随后是 preview 和 review。返回首页节点用于退出流程并重定向到仪表板。

$ xml

<?xml version="1.0" encoding="UTF-8"?>
<faces-config xmlns="https://jakarta.ee/xml/ns/jakartaee"
              xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
              xsi:schemaLocation="https://jakarta.ee/xml/ns/jakartaee
                                  https://jakarta.ee/xml/ns/jakartaee/web-facesconfig_4_1.xsd"
              version="4.1">

    <flow-definition id="customer-segmentation">

        <start-node>configure</start-node>

        <view id="configure">
            <vdl-document>
                /customer-segmentation/configure.xhtml
            </vdl-document>
        </view>

        <view id="preview">
            <vdl-document>
                /customer-segmentation/preview.xhtml
            </vdl-document>
        </view>

        <view id="review">
            <vdl-document>
                /customer-segmentation/review.xhtml
            </vdl-document>
        </view>

        <flow-return id="home">
            <from-outcome>/index?faces-redirect=true</from-outcome>
        </flow-return>

    </flow-definition>

</faces-config>

关键概念在于两种生命周期之间的区别：

@ViewScoped 为当前页面保持仪表板状态。
@FlowScoped 跨多个页面保持向导状态。
批处理执行独立于这两种 UI 作用域运行。

这种分离确保了 Web 交互保持短生命周期，而批处理作业在流程返回首页后仍继续处理。## 第 4 步：启动应用程序并使用它

要运行应用，请在本地启动 MongoDB，或配置 MongoDB Atlas 连接。使用提供的 Maven 和 GlassFish 设置构建并启动 Jakarta EE 应用。应用启动后，在浏览器中打开仪表板。如果客户集合为空，示例数据会自动加载，使你能够立即测试客户细分流程。

在仪表板上，选择 开始分段，并更新每个客户层级的最低消费金额。继续进入预览步骤，比较当前分布和预测分布。查看更改后，确认启动 Jakarta Batch 作业。处理期间应用会返回仪表板；作业完成后，刷新后的仪表板会显示更新后的客户分布。

[LOADING...]

结论

批处理提供了一种有条理的方法，可以独立于单个请求或用户交互来管理大规模流程。通过将工作负载划分为更小、可管理的单元，并以可预测的方式处理它们，运行时可以协调执行、监控进度并处理恢复。这种方法对于重复性的企业任务尤其有效，例如重新分类、对账、迁移和批量更新。

分块是批处理中的一个核心概念。它不是一次性处理整个数据集，而是将工作划分为更小的单元，以便增量处理和提交。最佳块大小因数据量、处理成本、内存和数据库性能而异。批处理为平衡这些权衡提供了清晰的结构，同时保持简单、可控的业务工作流。

准备好探索 MongoDB Atlas 的优势了吗？现在就开始，试用 MongoDB Atlas。

访问源代码，该源代码用于本教程。

有任何问题吗？来 MongoDB 社区论坛与我们交流。

参考资料：

源代码

分类导航

先决条件

步骤 1：生成项目

步骤 3：创建 Bean

结论