使用Spring Integration的JdbcChannelMessageStore遇异常及数据丢失问题

配置代码 #

<int:transaction-synchronization-factory id="syncFactory">
    <int:after-commit expression="@store.removeFromIdCache(headers.id)" />
    <int:after-rollback expression="@store.removeFromIdCache(headers.id)"/>
</int:transaction-synchronization-factory>
 
<task:executor id="pool" pool-size="10"  queue-capacity="10" rejection-policy="CALLER_RUNS" />

<int:channel id="objectToJsonChannelTMF639">
    <int:queue message-store="store"/>
    <!--<int:interceptors>
        <bean class="ca.bell.bmf.customer.service.MyChannelInterceptor" />
    </int:interceptors>-->
</int:channel>       

<bean id="store" class="org.springframework.integration.jdbc.store.JdbcChannelMessageStore">
    <property name="dataSource" ref="dataSource"/>
    <property name="channelMessageStoreQueryProvider" ref="queryProvider"/>
    <property name="region" value="CONCURRENT_POLL"/>
    <property name="usingIdCache" value="true"/>
</bean>

<int:bridge input-channel="objectToJsonChannelTMF639">
    <int:poller fixed-delay="500" receive-timeout="500"
        max-messages-per-poll="1" task-executor="pool">
        <int:transactional propagation="REQUIRED" synchronization-factory="syncFactory"
        isolation="READ_COMMITTED" transaction-manager="transactionManager" />
    </int:poller>
</int:bridge>

错误日志 #

[2023-08-10 00:08:38,909] ERROR org.springframework.integration.handler.LoggingHandler:250 - ErrorMessage [payload=org.springframework.messaging.MessageDeliveryException: Dispatcher has no subscribers for channel 'application.jdbcMessageStoreChannel'.; nested exception is org.springframework.integration.MessageDispatchingException: Dispatcher has no subscribers, failedMessage=GenericMessage [payload=CPMPayload [cpmBody=[{"op":"replace","jsonPath":"$.resourceCharacteristic[?(@.name=='AStatus')].value","value":"returned"},{"op":"replace","jsonPath":"$.resourceCharacteristic[?(@.name=='FStatus')].value","value":"FCredited"},{"op":"replace","jsonPath":"$.resourceCharacteristic[?(@.name=='tmpCPMData_orderId')].value","value":"89CJ00"}], domain=logicalResource, transactionID=44234, externalid=89CJ00, specName=VideoCpe_LD, externalId=COM-BC743HE1C-1-1, sourceSystem=3PL-Process, transnSeqId=44234], headers={id=bb7a6bac-7767-5467-490a-c326e780e0ab, timestamp=1691606317062}], failedMessage=GenericMessage [payload=CPMPayload [cpmBody=[{"op":"replace","jsonPath":"$.resourceCharacteristic[?(@.name=='AStatus')].value","value":"returned"},{"op":"replace","jsonPath":"$.resourceCharacteristic[?(@.name=='FStatus')].value","value":"FCredited"},{"op":"replace","jsonPath":"$.resourceCharacteristic[?(@.name=='tmpCPMData_orderId')].value","value":"89CJ00"}], domain=logicalResource, transactionID=44234, externalid=89CJ00, specName=VideoCpe_LD, externalId=COM-BC743HE1C-1-1, sourceSystem=3PL-Process, transnSeqId=44234], headers={id=bb7a6bac-7767-5467-490a-c326e780e0ab, timestamp=1691606317062}], headers={id=5458abcc-d97f-5b0f-9278-5525d2663fa8, timestamp=1691606318909}] for original GenericMessage [payload=CPMPayload [cpmBody=[{"op":"replace","jsonPath":"$.resourceCharacteristic[?(@.name=='AStatus')].value","value":"returned"},{"op":"replace","jsonPath":"$.resourceCharacteristic[?(@.name=='FStatus')].value","value":"FCredited"},{"op":"replace","jsonPath":"$.resourceCharacteristic[?(@.name=='tmpCPMData_orderId')].value","value":"89CJ00"}], domain=logicalResource, transactionID=44234, externalid=89CJ00, specName=VideoCpe_LD, externalId=COM-BC743HE1C-1-1, sourceSystem=3PL-Process, transnSeqId=44234], headers={id=bb7a6bac-7767-5467-490a-c326e780e0ab, timestamp=1691606317062}]

遇到的问题 #

• 运行代码时抛出Dispatcher has no subscribers for channel 'application.jdbcMessageStoreChannel'异常
• 插入数据库的数据刷新后消失，事务提交操作未执行

可能原因及解决方法 #

针对Dispatcher无订阅者异常 #

原因

• application.jdbcMessageStoreChannel通道未配置对应的消费者端点，消息无法被处理
• 通道ID拼写错误，或端点配置时关联的通道名称不匹配
• 若该通道为JdbcChannelMessageStore内部使用，可能存在Spring Integration版本兼容问题，或事务管理器未正确关联导致内部通道初始化异常

解决方法

• 检查application.jdbcMessageStoreChannel的定义，确保该通道绑定了合法的订阅者（如service-activator、bridge、router等端点）
• 核对所有涉及该通道的配置，确保通道名称拼写一致
• 验证Spring Integration版本与JdbcChannelMessageStore的兼容性，同时确认transactionManager配置正确且与消息存储使用的数据源关联

针对数据丢失、事务未提交问题 #

原因

1. 事务配置失效：transactionManager未正确关联JdbcChannelMessageStore使用的数据源，或poller的事务传播行为设置不当
2. ID缓存与数据库不一致：usingIdCache=true时，事务同步的removeFromIdCache操作可能在事务提交前执行，导致缓存清理后数据库事务回滚，或缓存状态与数据库不一致
3. 数据源自动提交冲突：数据源开启了自动提交，与Spring事务管理逻辑冲突，导致事务未被正确管理
4. 表结构异常：queryProvider对应的数据库表结构不完整（如缺少主键、索引），导致插入操作实际未成功写入数据库
5. 多线程竞争：CONCURRENT_POLL模式下，多线程消费可能导致事务提交逻辑异常

解决方法

• 确认transactionManager关联的数据源与JdbcChannelMessageStore的dataSource为同一实例，保证事务上下文一致
• 检查事务同步工厂的表达式@store.removeFromIdCache(headers.id)，确保store是正确的bean引用；CONCURRENT_POLL模式下，需确保事务提交/回滚后再执行缓存清理，避免状态不一致
• 关闭数据源的自动提交配置，交由Spring事务管理器统一控制事务生命周期
• 验证queryProvider对应的数据库表结构是否符合JdbcChannelMessageStore的要求（如是否创建了INT_CHANNEL_MESSAGE表及必要索引）
• 开启事务日志，查看transactionManager的事务提交/回滚记录，定位事务未提交的具体环节
• 尝试调整poller的事务隔离级别，或暂时移除isolation="READ_COMMITTED"配置，使用默认隔离级别测试

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