JMS 消息侦听器调用程序无法强制转换为 org.apache.qpid.client.AMQDestination

Question

我有一个如下配置文件，

@Slf4j
@Configuration
@EnableJms
public class MyQpidConfig {
    @Bean("myQpidConn")
    public AMQConnectionFactory amqConnectionFactory() throws URLSyntaxException {
        AMQConnectionFactory amqConnectionFactory = new AMQConnectionFactory("URL");
        return amqConnectionFactory;
    }

    @Bean("myJmsContainer")
    public DefaultMessageListenerContainer defaultMessageListenerContainer(@Qualifier("myQpidAdaptor") MessageListenerAdapter messageListenerAdapter ,@Qualifier("myQpidConn")AMQConnectionFactory connectionFactory) {
        DefaultMessageListenerContainer messageListenerContainer = new DefaultMessageListenerContainer();
        messageListenerContainer.setConnectionFactory(connectionFactory);
         messageListenerContainer.setDestinationName(destinationName);
        messageListenerContainer.setMessageListener(messageListenerAdapter);
        messageListenerContainer.setPubSubDomain(false);
        messageListenerContainer.setSessionTransacted(true);
        messageListenerContainer.setDurableSubscriptionName("MYSUBSCRIBER");
        messageListenerContainer.setSubscriptionDurable(true);
        messageListenerContainer.setAutoStartup(true);
        messageListenerContainer.setDestination((Queue) () -> "MYQUEUE");
        return messageListenerContainer;
    }

    @Bean("myQpidAdaptor")
    public MessageListenerAdapter messageListenerAdapter() {
        Preconditions.checkNotNull(qpidMessageListner, "Consumer listener not set");
        MessageListenerAdapter messageListenerAdapter = new MessageListenerAdapter(new MyQpidMessageListner());
        messageListenerAdapter.setDefaultListenerMethod("onMessage");
        messageListenerAdapter.setDefaultResponseQueueName("MYQUEUE");
        return messageListenerAdapter;
    }
   
}

destinationName 等于

  ADDR:MYSUBSCRIBER:MYQUEUE; {"create":"receiver",
    "node":{"durable":true,"x-declare":
    {"exclusive":true,"auto-delete":false,
    "x-bindings":[{"exchange":"amq.topic","key":"MYQUEUE"}]}}}

我的 Listener 类如下，

@Slf4j
@Component
public class MyQpidMessageListner implements  MessageListener{
    @Override
    public void onMessage(Message jmsBytesMessage) {
       log.info("Control Inside");
       //to do
    }
}

作为异步消费者，当我当时运行我的应用程序时遇到与目的地相关的异常，

DefaultMessageListenerContainer - 
Setup of JMS message listener invoker failed for destination 'com.my.MyQpidConfig$$Lambda$637/429515253@3aa1975' - trying to recover. Cause: com.my.MyQpidConfig$$Lambda$637/429515253 cannot be cast to org.apache.qpid.client.AMQDestination  

如果它与 AMQDestination 相关，那么我如何在 java 代码中创建 AMQDestination。 我没有为这个 QPID 问题获得任何资源。 任何建议的帮助都必须感谢。

Answer 1

由于某种原因，您将 lambda 传递给 org.springframework.jms.listener.DefaultMessageListenerContainer.setDestination(Destination)，即：

messageListenerContainer.setDestination((Queue) () -> "MYQUEUE");

此方法需要 javax.jms.Destination 实现（例如 org.apache.qpid.client.AMQDestination），而不是 lambda。

您已经在设置目的地名称，即：

messageListenerContainer.setDestinationName(destinationName);

所以不清楚您为什么还要尝试设置目的地。这是配置同一事物的两种方式。

我建议您只需删除尝试设置目的地的代码并保留设置目的地名称的代码。