问题

我有一个使用案例，需要在运行时消费从队列中接收到的消息。

这里我有一个配置类和一个监听器类。我已经为两个现有队列定义了一个消费者，并希望能够消费在运行时可能发现的新队列，这些新队列遵循相同的命名约定，即队列.animals.*。

另外，我还有另一个服务，将在名为 "newQueues" 的队列上发送新发现的队列名称。如果不需要的话，可以更改这种方法，并且可以摆脱发送消息到 "newQueues" 的服务。

以下是代码示例：

@Configuration
@EnableRabbit
public class RabbitConfiguration implements RabbitListenerConfigurer {

    public static final String queue1 = "queue.animals.cat";
    public static final String queue2 = "queue.animals.dog";

    @Autowired
    private ConnectionFactory connectionFactory;

    @Bean
    public AmqpAdmin amqpAdmin() {
        return new RabbitAdmin(connectionFactory);
    }

    @Bean
    public DirectRabbitListenerContainerFactory rabbitListenerContainerFactory() {
        DirectRabbitListenerContainerFactory factory = new DirectRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setMessageConverter(new Jackson2JsonMessageConverter());
        return factory;
    }

    @Bean
    public RabbitListenerEndpointRegistry listenerEndpointRegistry() {
        return new RabbitListenerEndpointRegistry();
    }

    @Override
    public void configureRabbitListeners(RabbitListenerEndpointRegistrar registrar) {
        registrar.setEndpointRegistry(listenerEndpointRegistry());
    }

    @Autowired
    private RabbitListenerEndpointRegistry listenerEndpointRegistry;

    @RabbitListener(id = "qEvent", queues = {"newQueues"})
    public void processQueueEvents(String newQueueName) {
        ((DirectMessageListenerContainer) this.listenerEndpointRegistry.getListenerContainer("animalQContainer"))
            .addQueueNames(newQueueName);

        System.out.println("Received a message with the new queue name: " + newQueueName);
    }

    @RabbitListener(id = "animalQContainer", queues = {queue1, queue2})
    public void processAnimals(Animal animalObj, @Header(AmqpHeaders.CONSUMER_QUEUE) String queue) {
        System.out.println("Received a message on queue: " + queue + "data: " + animalObj);
        //处理 animalObj
    }
}

我目前遇到以下异常：

Caused by: com.fasterxml.jackson.core.JsonParseException: Unrecognized token 'queue': was expecting (JSON String, Number, Array, Object or token 'null', 'true' or 'false')

我对RabbitMQ非常陌生，所以不确定是否所有组件都配置正确。感谢您的帮助。

@EnableRabbit
public class RabbitConfiguration implements RabbitListenerConfigurer {
public static final String queue1= "queue.animals.cat";
public static final String queue2= "queue.animals.dog";
@Autowired
private ConnectionFactory connectionFactory;
@Bean
public AmqpAdmin amqpAdmin() {
return new RabbitAdmin(connectionFactory);
}
@Bean
public DirectRabbitListenerContainerFactory rabbitListenerContainerFactory() {
DirectRabbitListenerContainerFactory factory = new DirectRabbitListenerContainerFactory();
factory.setConnectionFactory(connectionFactory);
factory.setMessageConverter(new Jackson2JsonMessageConverter());
return factory;
}
@Bean
public RabbitListenerEndpointRegistry listenerEndpointRegistry() {
return new RabbitListenerEndpointRegistry();
}
@Override
public void configureRabbitListeners(RabbitListenerEndpointRegistrar registrar) {
registrar.setEndpointRegistry(listenerEndpointRegistry());
}

@Autowired
private RabbitListenerEndpointRegistry listenerEndpointRegistry;
@RabbitListener(id = "qEvent", queues = {"newQueues"})
public void processQueueEvents(String newQueueName) {
((DirectMessageListenerContainer) this.listenerEndpointRegistry.getListenerContainer("animalQContainer"))
.addQueueNames(newQueueName);
System.out.println("Received a message with the new queue name: " + newQueueName);
}
@RabbitListener(id = "animalQContainer" , queues = { queue1, queue2 })
public void processAnimals(Animal animalObj, @Header(AmqpHeaders.CONSUMER_QUEUE) String queue) {
System.out.println("Received a message on queue: " + queue + "data: " + animalObj);
//process animalObj
}

Caused by: com.fasterxml.jackson.core.JsonParseException: Unrecognized token 'queue': was expecting (JSON String, Number, Array, Object or token 'null', 'true' or 'false')

答案1

得分: 0

你的问题有点混淆，标题是关于添加新队列，但之后你提到了JSON异常。听起来你可能收到了非JSON内容。

问题具体在哪里？

要添加新队列，可以使用以下代码：

((AbstractMessageListenerContainer) rabbitListenEndpointRegistry.getMessageListenerContainer("animalQContainer"))
    .addQueueNames("newQueue1", "newQueue2");

((AbsrtactMessageListenerContainer) rabbitListenEndpointRegistry.getMessageListenerContainer("animalQContainer"))
    .addQeueNames("newQueue1", "newQueue2");

答案2

得分: 0

Sure, here's the translation of the provided content:

由于在此处配置了Jackson2JsonMessageConverter，processQueueEvents方法无法解析字符串。创建了一个新的类并传入一个对象到processQueueEvent方法，以解决问题中提到的异常：

    public void processQueueEvents(NewQueue newQueueName) {
System.out.println("Received a message on a new queue: " + newQueueName);
String name = newQueueName.toString();
<details>
<summary>英文:</summary>
Since the Jackson2JsonMessageConverter is configured here, the processQueueEvents method can not parse a string. Created a new class and passed in an object to the processQueueEvent method to get past the exception mentioned in the question: 
```  @RabbitListener(id = &quot;qEvent&quot;, queues = {&quot;new_queues&quot;})
public void processQueueEvents(NewQueue newQueueName) {
System.out.println(&quot;Received a message on a new queue: &quot; + newQueueName);
String name = newQueueName.toString();
</details>