1

我有一条看起来像这样的骆驼路线：

<routeContext id="rollover-route" xmlns="http://camel.apache.org/schema/spring">
    <route id="pickUpEntries">
        <from uri="jpa://cameltest.model.CamelTest?consumer.namedQuery=unconsumedEntries&amp;?consumeDelete=false&amp;persistenceUnit=persistenceProvider&amp;consumeLockEntity=true"/>
        <split executorServiceRef="rolloverProfile">
            <simple>${body}</simple>
            <setProperty propertyName="taskId">
                <simple>${body.taskId}</simple>
            </setProperty>
            <setProperty propertyName="providerId">
                <simple>${body.providerId}</simple>
            </setProperty>
            <setProperty propertyName="intakeId">
                <simple>${body.intakeId}</simple>
            </setProperty>
            <setProperty propertyName="consumed">
                <simple>${body.consumed}</simple>
            </setProperty>
            <to uri="jpa://cameltest.model.CamelTest?persistenceUnit=persistenceProvider"/>
            <process ref="JPAToMCS"/> 
            <process ref="MCSRequest"/> 
            <process ref="MCSToJPA"/> 
            <to uri="jpa://cameltest.model.CamelTest?persistenceUnit=persistenceProvider"/>
        </split>
    </route>
</routeContext>

CamelTest JPA 实体是：

package cameltest.model;

import java.io.Serializable;

import javax.persistence.Entity;
import javax.persistence.Id;
import javax.persistence.NamedQuery;
import javax.persistence.Table;

import org.apache.camel.component.jpa.Consumed;
import org.springframework.transaction.annotation.Transactional;

@Entity
@Table(name="CamelTestTable")
@NamedQuery(name = "unconsumedEntries", query = "select c from CamelTest c where c.consumed = 0")
public class CamelTest implements Serializable {


    private static final long serialVersionUID = 5608716486854758950L;
    @Id
    private Integer taskId;
    private Integer providerId;
    private Integer intakeId;
    private Integer consumed;
    private Integer complete;


    public CamelTest()
    {}

    public CamelTest(Integer taskId, Integer providerId, Integer intakeId,
            Integer consumed, Integer complete) {
        super();
        this.taskId = taskId;
        this.providerId = providerId;
        this.intakeId = intakeId;
        this.consumed = consumed;
        this.complete = complete;
    }

    public Integer getComplete() {
        return complete;
    }

    public void setComplete(Integer complete) {
        this.complete = complete;
    }

    public Integer getConsumed() {
        return consumed;
    }

    public void setConsumed(Integer consumed) {
        this.consumed = consumed;
    }

    public Integer getTaskId() {
        return taskId;
    }

    public void setTaskId(Integer taskId) {
        this.taskId = taskId;
    }

    public Integer getProviderId() {
        return providerId;
    }

    public void setProviderId(Integer providerId) {
        this.providerId = providerId;
    }

    public Integer getIntakeId() {
        return intakeId;
    }

    public void setIntakeId(Integer intakeId) {
        this.intakeId = intakeId;
    }

    @Override
    public String toString() {
        return "CamelTest [taskId=" + taskId + ", providerId=" + providerId
                + ", intakeId=" + intakeId + "]";
    }

    @Consumed
    public void markAsConsumed()
    {
        setConsumed(getConsumed() + 1);
    }

}

如何强制 Camel 路由在结束前提交消费列的状态？中间处理器需要很多时间，我希望它在完成（并标记）之前已被提交到数据库中。

4

1 回答 1

1

我发现这样做的一种方法是将 JPA 对象路由到一个单独的线程以将其持久化，而另一个线程进行处理：

        <route id="pickUpEntries">
        <from
            uri="jpa:/batchprocessor.model.BatchProcess?consumer.namedQuery=unconsumedEntries&amp;?consumeDelete=false&amp;persistenceUnit=persistenceProvider&amp;consumeLockEntity=true" />
        <process ref="Consume" />
        <multicast parallelProcessing="true">
            <to
                uri="jpa://batchprocessor.model.BatchProcess?persistenceUnit=persistenceProvider" />
            <to uri="direct:process" />
        </multicast>
    </route>

不完全令人满意，但在这种情况下可以完成工作。我还必须删除@Consumed。令人困惑的是，它在路线完成后处理，而不是在处理项目时处理。

于 2013-10-03T15:17:20.993 回答