spring amqp rabbitmq MessageListener不工作

Question

我想用springmq使用spring amqp，下面是我的配置。

<rabbit:connection-factory id="rabbitConnectionFactory" 
    port="${rabbitmq.port}" host="${rabbitmq.host}" /> 

<rabbit:admin connection-factory="rabbitConnectionFactory" /> 

<rabbit:queue name="${rabbitmq.import.queue}" /> 

<rabbit:template id="importAmqpTemplate" 
    connection-factory="rabbitConnectionFactory" queue="${rabbitmq.import.queue}" /> 

<beans:bean id="importExchangeMessageListener" 
    class="com.stockopedia.batch.foundation.ImportMessageListener" /> 

<rabbit:listener-container 
    connection-factory="rabbitConnectionFactory" concurrency="5"> 
    <rabbit:listener queues="${rabbitmq.import.queue}" ref="importMessageListener" /> 
</rabbit:listener-container> 

这是简单的消息监听器类，

import org.springframework.amqp.core.Message; 
import org.springframework.amqp.core.MessageListener; 

public class ImportMessageListener implements MessageListener { 

    @Override 
    public void onMessage(Message message) { 
     System.out.println("consumer output: " + message); 
    } 

} 

这是制片人（这是春天一批itemWriter），

public class ImportItemWriter<T> implements ItemWriter<T> { 

    private AmqpTemplate template; 

    public AmqpTemplate getTemplate() { 
     return template; 
    } 

    public void setTemplate(AmqpTemplate template) { 
     this.template = template; 
    } 

    public void write(List<? extends T> items) throws Exception { 
     for (T item : items) { 
      Object reply = template.convertSendAndReceive(item.toString()); 
      System.out.println("producer output: " + reply); 
     } 
    } 

} 

当我跑我的春天批处理作业，ImportItemWriter。写入被调用。但ImportMessageListener.onMessage不起作用。它不打印消息。我得到下面的所有项目输出在控制台上

producer output: null 
producer output: null 
producer output: null 
producer output: null 
producer output: null 
producer output: null 
producer output: null 

Answer 1

你的消费不发送的结果...

@Override 
public void onMessage(Message message) { 
    System.out.println("consumer output: " + message); 
} 

将其更改为一个简单的POJO;容器的MessageListenerAdapter将负责您的转换，并发送结果。

@Override 
public String handleMessage(String message) { 
    System.out.println("consumer output: " + message); 
    return "result"; 
} 

编辑：

您还没有建立到队列中的任何交换或路由。如果您想使用默认的交换/路由，使用...

convertSendAndReceive("", queueName, item.toString()); 

EDIT2：

或者，将routingKey在模板上的队列名称，然后你可以用更简单的方法。

...sendAndReceive()方法用于请求/回复场景，因此需要阻止。要异步执行此操作，您必须使用...send()方法之一，并将自己的SimpleListenerContainer连接起来以接收答复;你将不得不做你自己的关联。使用

public void convertAndSend(Object message, MessagePostProcessor postProcessor) 

，并在您的邮件后处理器，设置replyTo和correlationId头...

message.getMessageProperties().setReplyTo("foo"); 
message.getMessageProperties().setCorrelationId("bar"); 

或者建立自己的Message对象（例如by using the MessageBuilder），并使用send方法...

template.send(MessageBuilder.withBody("foo".getBytes()) 
      .setReplyTo("bar") 
      .setCorrelationId("baz".getBytes()) 
      .build()); 

每个请求都需要一个唯一的correlationId这样你就可以响应相关。