Apache Camel：轮询后访问自定义Route出现类型转换异常问题

问题原因 #

你遇到的类转换异常是因为Camel在启动时会将你自定义的MyRoute包装成内部的DefaultRoute实例，调用getRoute("dataRoute")返回的是这个封装后的对象，而非你的MyRoute原实例，因此强转失败。

可行解决方案 #

不要直接通过Route实例获取lastId，改用全局状态存储的方式，让PollStrategy能安全拿到数据。

方案一：利用CamelContext属性存储状态 #

在路由的处理逻辑中，将lastId存入CamelContext的全局属性，然后在PollStrategy中读取：

1. 修改路由中的process逻辑：

private Long startId = 0L;
private Long lastId = 0L;

from("jpa://Data?consumer.namedQuery=findByIdGreaterThan&consumer.parameter=#{startId}")
    .routeId("dataRoute")
    .process(ex -> {
        Data data = ex.getIn().getBody(Data.class);
        lastId = data.getId();
        // 将lastId存入CamelContext全局属性
        ex.getContext().getProperties().put("dataRouteLastId", lastId);
        NewData newData = (NewData) convertData(data);
        ex.getMessage().setBody(newData);
    })
    .to("jpa://NewData")

2. 在自定义PollStrategy的commit方法中读取：

@Override
public void commit(Consumer consumer, Endpoint endpoint, int polledMessages) {
    CamelContext context = endpoint.getCamelContext();
    Long lastId = (Long) context.getProperties().get("dataRouteLastId");
    if (lastId != null && lastId > startId) {
        startId = lastId;
        // 执行startId的持久化操作（比如写入数据库）
        log.debug("New highest ID saved: {}", startId);
    }
}

方案二：自定义状态管理Bean（推荐） #

用一个单例Bean统一管理路由的状态，避免直接操作CamelContext属性，更适合复杂场景：

1. 定义状态管理Bean：

@Component
public class DataRouteState {
    private volatile Long startId = 0L;
    private volatile Long lastId = 0L;

    public void updateLastId(Long id) {
        if (id > lastId) {
            this.lastId = id;
        }
    }

    public void syncStartId() {
        this.startId = this.lastId;
    }

    public Long getStartId() {
        return startId;
    }

    public Long getLastId() {
        return lastId;
    }
}

2. 在路由中注入Bean并更新状态：

@Autowired
private DataRouteState routeState;

from("jpa://Data?consumer.namedQuery=findByIdGreaterThan&consumer.parameter=#{routeState.startId}")
    .routeId("dataRoute")
    .process(ex -> {
        Data data = ex.getIn().getBody(Data.class);
        routeState.updateLastId(data.getId());
        NewData newData = (NewData) convertData(data);
        ex.getMessage().setBody(newData);
    })
    .to("jpa://NewData")

3. 在自定义PollStrategy中注入Bean并完成startId持久化：

@Autowired
private DataRouteState routeState;

@Override
public void commit(Consumer consumer, Endpoint endpoint, int polledMessages) {
    Long lastId = routeState.getLastId();
    Long startId = routeState.getStartId();
    if (lastId > startId) {
        routeState.syncStartId();
        // 这里执行startId的持久化逻辑，比如写入数据库
        log.debug("Updated persisted startId to: {}", routeState.getStartId());
    }
}

额外注意点 #

• 原路由中的onCompletion是每个消息处理完成后触发，而PollStrategy的commit是整个轮询批次完成后触发，后者更符合你“轮询结束后统一更新startId”的需求。
• 如果是分布式场景，建议将startId持久化到数据库或配置中心，避免应用重启后状态丢失。

内容的提问来源于stack exchange，提问作者Noel93