使用Camel与CachingConnectionFactory配置ActiveMQ持久化客户端ID报错求助

使用CachingConnectionFactory为Camel搭建ActiveMQ上下文，通过Camel连接ActiveMQ时抛出如下错误：

setClientID call not supported on proxy for shared Connection. Set the 'clientId' property on the SingleConnectionFactory instead.

相关代码 #

ActiveMQComponent amqComponent = new ActiveMQComponent();
ActiveMQConnectionFactory cf = new ActiveMQConnectionFactory();
cf.setBrokerURL("tcp://" + amqServer + ":" + amqPort + "?jms.watchTopicAdvisories=false");
            
UserCredentialsConnectionFactoryAdapter uca = new UserCredentialsConnectionFactoryAdapter();
uca.setUsername(amqUser);
uca.setPassword(amqPassword);
uca.setTargetConnectionFactory(cf);
            
CachingConnectionFactory ccf = new CachingConnectionFactory(uca);
ccf.setClientId(amqClientID);
            
amqComponent.setConnectionFactory(ccf);
amqComponent.setMaxConcurrentConsumers(1);

context.addComponent("activemq", amqComponent);

...

from("activemq:topic:" + amqFeedTopic + "?clientId=" + id + "&durableSubscriptionName=" + id + "-sub")
.id(id)
.bean(messageHandlerClass, "process")
.to("kafka:" + kafkaTopic + "?brokers=" + kafkaBootstrapServers);

问题原因 #

1. CachingConnectionFactory依赖共享连接代理工作，代理对象不支持直接调用setClientID方法，错误提示已明确建议在SingleConnectionFactory上配置clientId。
2. 代码同时在Camel端点URL中配置了clientId参数，这会与连接工厂的配置冲突——持久订阅的clientId需统一在连接工厂层面设置，不能在端点重复配置。

修复方案 #

步骤1：调整连接工厂配置

将CachingConnectionFactory包装在SingleConnectionFactory中，在SingleConnectionFactory上设置clientId，规避代理调用限制：

ActiveMQComponent amqComponent = new ActiveMQComponent();
ActiveMQConnectionFactory cf = new ActiveMQConnectionFactory();
cf.setBrokerURL("tcp://" + amqServer + ":" + amqPort + "?jms.watchTopicAdvisories=false");
            
UserCredentialsConnectionFactoryAdapter uca = new UserCredentialsConnectionFactoryAdapter();
uca.setUsername(amqUser);
uca.setPassword(amqPassword);
uca.setTargetConnectionFactory(cf);

// 通过SingleConnectionFactory设置clientId
SingleConnectionFactory scf = new SingleConnectionFactory(uca);
scf.setClientId(amqClientID);
            
// 将SingleConnectionFactory传入CachingConnectionFactory
CachingConnectionFactory ccf = new CachingConnectionFactory(scf);
            
amqComponent.setConnectionFactory(ccf);
amqComponent.setMaxConcurrentConsumers(1);

context.addComponent("activemq", amqComponent);

步骤2：移除端点URL中的clientId参数

持久订阅的clientId已在连接工厂配置，端点仅保留durableSubscriptionName即可：

from("activemq:topic:" + amqFeedTopic + "?durableSubscriptionName=" + id + "-sub")
.id(id)
.bean(messageHandlerClass, "process")
.to("kafka:" + kafkaTopic + "?brokers=" + kafkaBootstrapServers);

额外说明 #

• 持久订阅的clientId必须全局唯一，多消费者实例需保证各自的clientId不重复。
• 当前setMaxConcurrentConsumers(1)的配置与持久订阅的单连接模式匹配，无需调整。

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