带有回调的 RabbitTemplate 异步消息答案

【问题标题】：RabbitTemplate async message with callback带有回调的 RabbitTemplate 异步消息
【发布时间】：2016-10-24 09:35:38
【问题描述】：

我正在努力处理带有回调消息的异步 RabbitMQ 消息。

这是我制作人的一部分：

@Autowired(required = false)
@Qualifier("rabbitTemplate")
private RabbitTemplate queueTemplate;

@Override
public Response createIncident(String incident) {
    LOGGER.info("Sending incident into queue");
    queueTemplate.convertAndSend((Object)incident, new MessagePostProcessor() {

        @Override
        public Message postProcessMessage(Message message) throws AmqpException {
            message.getMessageProperties().setReplyTo("replyQueue");
            return message;
        }
    });
    Message message = queueTemplate.receive();
...

Spring rabbitMQ 上下文

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

<rabbit:queue name="rabbitQueue" />
<rabbit:queue name="replyQueue" />

<rabbit:topic-exchange name="rabbitExchange">
    <rabbit:bindings>
        <rabbit:binding queue="rabbitQueue" pattern="camel" />
        <rabbit:binding queue="replyQueue" pattern="camel" />
    </rabbit:bindings>
</rabbit:topic-exchange>


<rabbit:template id="rabbitTemplate" 
    connection-factory="rabbitConnFactory" exchange="rabbitExchange" routing-key="camel"
    reply-address="replyQueue" reply-queue="replyQueue">
    <rabbit:reply-listener/>
</rabbit:template>

<rabbit:listener-container
    connection-factory="rabbitConnFactory" concurrency="10">
    <rabbit:listener ref="rabbitMessageListener" method="createIncident"
        queue-names="rabbitQueue" />
</rabbit:listener-container>

<bean id="rabbitMessageListener" class="com.my.listener.QueueListener" />

以及应该接收和响应的监听器

@Autowired(required = false)
@Qualifier("rabbitTemplate")
private RabbitTemplate queueTemplate;

public void createIncident() {
    queueTemplate.receiveAndReply("replyQueue", new ReceiveAndReplyMessageCallback() {

        @Override
        public Message handle(Message message) {
            String incident = new String(message.getBody());
            LOGGER.debug("Queue incoming message: " + incident);
            String result = "" + messageHandler.createIncident(incident);
            return new Message(result.getBytes(), new MessageProperties());
        }

    });

它不想在侦听器中进入此句柄消息。以及应用容器显示消息

2016-06-22 14:15:30,457 错误 [AbstractFaultChainInitiatorObserver:115] 错误期间发生错误处理，放弃！ org.apache.cxf.interceptor.Fault：没有“队列” 指定的。检查 RabbitTemplate 的配置。在 org.apache.cxf.service.invoker.AbstractInvoker.createFault(AbstractInvoker.java:170) 在 org.apache.cxf.service.invoker.AbstractInvoker.invoke(AbstractInvoker.java:136) 在 org.apache.cxf.jaxrs.JAXRSInvoker.invoke(JAXRSInvoker.java:204) 在 org.apache.cxf.jaxrs.JAXRSInvoker.invoke(JAXRSInvoker.java:101) 在 org.apache.cxf.interceptor.ServiceInvokerInterceptor$1.run(ServiceInvokerInterceptor.java:58) 在 org.apache.cxf.interceptor.ServiceInvokerInterceptor.handleMessage(ServiceInvokerInterceptor.java:94) 在 org.apache.cxf.phase.PhaseInterceptorChain.doIntercept(PhaseInterceptorChain.java:272) 在 org.apache.cxf.transport.ChainInitiationObserver.onMessage(ChainInitiationObserver.java:121) 在 org.apache.cxf.transport.http.AbstractHTTPDestination.invoke(AbstractHTTPDestination.java:249) 在 org.apache.cxf.transport.servlet.ServletController.invokeDestination(ServletController.java:248) 在 org.apache.cxf.transport.servlet.ServletController.invoke(ServletController.java:222) 在 org.apache.cxf.transport.servlet.ServletController.invoke(ServletController.java:153) 在 org.apache.cxf.transport.servlet.CXFNonSpringServlet.invoke(CXFNonSpringServlet.java:171) 在 org.apache.cxf.transport.servlet.AbstractHTTPServlet.handleRequest(AbstractHTTPServlet.java:289) 在 org.apache.cxf.transport.servlet.AbstractHTTPServlet.doPost(AbstractHTTPServlet.java:209) 在 javax.servlet.http.HttpServlet.service(HttpServlet.java:650) 在 org.apache.cxf.transport.servlet.AbstractHTTPServlet.service(AbstractHTTPServlet.java:265) 在 org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:303) 在 org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:208) 在 org.apache.tomcat.websocket.server.WsFilter.doFilter(WsFilter.java:52) 在 org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:241) 在 org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:208) 在 com.mobilitymedia.connected.http.core.LoggerFilter.doFilter(LoggerFilter.java:41) 在 org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:241) 在 org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:208) 在 org.springframework.security.web.FilterChainProxy$VirtualFilterChain.doFilter(FilterChainProxy.java:330) 在 org.springframework.security.web.access.intercept.FilterSecurityInterceptor.invoke(FilterSecurityInterceptor.java:118) 在 org.springframework.security.web.access.intercept.FilterSecurityInterceptor.doFilter(FilterSecurityInterceptor.java:84) 在 org.springframework.security.web.FilterChainProxy$VirtualFilterChain.doFilter(FilterChainProxy.java:342) 在 org.springframework.security.web.access.ExceptionTranslationFilter.doFilter(ExceptionTranslationFilter.java:113) 在 org.springframework.security.web.FilterChainProxy$VirtualFilterChain.doFilter(FilterChainProxy.java:342) 在 org.springframework.security.web.session.SessionManagementFilter.doFilter(SessionManagementFilter.java:103) 在 org.springframework.security.web.FilterChainProxy$VirtualFilterChain.doFilter(FilterChainProxy.java:342) 在 org.springframework.security.web.authentication.AnonymousAuthenticationFilter.doFilter(AnonymousAuthenticationFilter.java:113) 在 org.springframework.security.web.FilterChainProxy$VirtualFilterChain.doFilter(FilterChainProxy.java:342) 在 org.springframework.security.web.servletapi.SecurityContextHolderAwareRequestFilter.doFilter(SecurityContextHolderAwareRequestFilter.java:154) 在 org.springframework.security.web.FilterChainProxy$VirtualFilterChain.doFilter(FilterChainProxy.java:342) 在 org.springframework.security.web.savedrequest.RequestCacheAwareFilter.doFilter(RequestCacheAwareFilter.java:45) 在 org.springframework.security.web.FilterChainProxy$VirtualFilterChain.doFilter(FilterChainProxy.java:342) 在 org.springframework.security.web.authentication.preauth.AbstractPreAuthenticatedProcessingFilter.doFilter(AbstractPreAuthenticatedProcessingFilter.java:94) 在 org.springframework.security.web.FilterChainProxy$VirtualFilterChain.doFilter(FilterChainProxy.java:342) 在 org.springframework.security.web.context.request.async.WebAsyncManagerIntegrationFilter.doFilterInternal(WebAsyncManagerIntegrationFilter.java:50) 在 org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:107) 在 org.springframework.security.web.FilterChainProxy$VirtualFilterChain.doFilter(FilterChainProxy.java:342) 在 org.springframework.security.web.context.SecurityContextPersistenceFilter.doFilter(SecurityContextPersistenceFilter.java:87) 在 org.springframework.security.web.FilterChainProxy$VirtualFilterChain.doFilter(FilterChainProxy.java:342) 在 org.springframework.security.web.FilterChainProxy.doFilterInternal(FilterChainProxy.java:192) 在 org.springframework.security.web.FilterChainProxy.doFilter(FilterChainProxy.java:160) 在 org.springframework.web.filter.DelegatingFilterProxy.invokeDelegate(DelegatingFilterProxy.java:344) 在 org.springframework.web.filter.DelegatingFilterProxy.doFilter(DelegatingFilterProxy.java:261) 在 org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:241) 在 org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:208) 在 org.apache.catalina.core.StandardWrapperValve.invoke(StandardWrapperValve.java:220) 在 org.apache.catalina.core.StandardContextValve.invoke(StandardContextValve.java:122) 在 org.apache.catalina.authenticator.AuthenticatorBase.invoke(AuthenticatorBase.java:505) 在 org.apache.catalina.core.StandardHostValve.invoke(StandardHostValve.java:169) 在 org.apache.catalina.valves.ErrorReportValve.invoke(ErrorReportValve.java:103) 在 org.apache.catalina.valves.AccessLogValve.invoke(AccessLogValve.java:956) 在 org.apache.catalina.core.StandardEngineValve.invoke(StandardEngineValve.java:116) 在 org.apache.catalina.connector.CoyoteAdapter.service（CoyoteAdapter.java:436）在 org.apache.coyote.http11.AbstractHttp11Processor.process(AbstractHttp11Processor.java:1078) 在 org.apache.coyote.AbstractProtocol$AbstractConnectionHandler.process(AbstractProtocol.java:625) 在 org.apache.tomcat.util.net.JIoEndpoint$SocketProcessor.run(JIoEndpoint.java:316) 在 java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) 在 java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) 在 org.apache.tomcat.util.threads.TaskThread$WrappingRunnable.run(TaskThread.java:61) 在 java.lang.Thread.run(Thread.java:745) 引起： org.springframework.amqp.AmqpIllegalStateException：没有“队列” 指定的。检查 RabbitTemplate 的配置。在 org.springframework.amqp.rabbit.core.RabbitTemplate.getRequiredQueue(RabbitTemplate.java:1514) 在 org.springframework.amqp.rabbit.core.RabbitTemplate.receive(RabbitTemplate.java:802) 在 com.myService.IncidentServiceDefaultImpl.createIncident(IncidentServiceDefaultImpl.java:36) 在 sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) 在 sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) 在 sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) 在 java.lang.reflect.Method.invoke(Method.java:498) 在 org.apache.cxf.service.invoker.AbstractInvoker.performInvocation(AbstractInvoker.java:188) 在 org.apache.cxf.service.invoker.AbstractInvoker.invoke(AbstractInvoker.java:104) ... 65 更多

【问题讨论】：

发布完整的堆栈跟踪。
我也遇到了同样的问题，你解决了吗？

标签： java spring rabbitmq spring-amqp

【解决方案1】：

你应该使用：

/**
 * The name of the default queue to receive messages from when none is specified explicitly.
 *
 * @param queue the default queue name to use for receive
 */
public void setQueue(String queue) {
    this.queue = queue;
}

而不是reply-address。

并重新考虑你的

queueTemplate.receiveAndReply("replyQueue",

在我看来，听众应该看看绑定到routing-key="camel" 的一些不同队列。

虽然你仍然对<binding> 感到困惑。

另外，如果您真的可以在消费者部分使用ListenerContainer 而不是receiveAndReply，那会更好。

【讨论】：