求助：如何用Spring Integration实现RabbitMQ官方RPC示例？

Got it, let's break down how to implement the RabbitMQ RPC pattern using Spring Integration, aligned with the Spring AMQP official RPC example, specifically for your scenario where the client uses convertSendAndReceive and the server is built with Spring Integration. I'll walk you through step-by-step configurations and key details to avoid common pitfalls.

The server needs to listen for RPC requests, process them, and send back responses with proper correlation to match the client's original request.

1.1 Core Dependencies #

First, ensure you have these dependencies in your project:

• Spring Integration AMQP
• Spring AMQP
• Spring Boot Starter (if using Spring Boot for auto-configuration)

1.2 Configuration #

Here's a Java-based configuration for the server:

@Configuration
@EnableIntegration
public class RabbitRpcServerConfig {

    @Value("${rabbitmq.rpc.request.queue}")
    private String rpcRequestQueue;

    @Autowired
    private ConnectionFactory connectionFactory;

    // Define the request queue (matches where the client sends requests)
    @Bean
    public Queue rpcRequestQueue() {
        return new Queue(rpcRequestQueue, false);
    }

    // AMQP Inbound Channel Adapter - listens for requests on the queue
    @Bean
    public AmqpInboundChannelAdapter amqpInboundChannelAdapter(MessageChannel rpcRequestChannel) {
        AmqpInboundChannelAdapter adapter = new AmqpInboundChannelAdapter(new SimpleMessageListenerContainer(connectionFactory));
        adapter.setQueueNames(rpcRequestQueue);
        adapter.setOutputChannel(rpcRequestChannel);
        // Critical: Preserve RPC-related headers (replyTo, correlationId)
        adapter.setHeaderMapper(new DefaultAmqpHeaderMapper());
        return adapter;
    }

    // Request channel for the Integration flow
    @Bean
    public MessageChannel rpcRequestChannel() {
        return new DirectChannel();
    }

    // Service Activator - your business logic to process requests
    @ServiceActivator(inputChannel = "rpcRequestChannel")
    public String handleRpcRequest(String request) {
        // Replace with your actual business processing
        return "Processed response for request: " + request;
    }

    // AMQP Outbound Gateway - sends responses back to the client's reply queue
    @Bean
    public AmqpOutboundGateway amqpOutboundGateway() {
        AmqpOutboundGateway gateway = new AmqpOutboundGateway(connectionFactory);
        // Use the replyTo header from the incoming request to route responses
        gateway.setRoutingKeyExpression(new SpelExpressionParser().parseExpression("headers['amqp_replyTo']"));
        // Propagate the correlation ID to match client requests
        gateway.setCorrelationKeyExpression(new SpelExpressionParser().parseExpression("headers['amqp_correlationId']"));
        return gateway;
    }

    // Integration Flow: Connect request processing to response sending
    @Bean
    public IntegrationFlow rpcResponseFlow() {
        return IntegrationFlow.from("rpcRequestChannel")
                .handle("rabbitRpcServerConfig", "handleRpcRequest")
                .handle(amqpOutboundGateway())
                .get();
    }
}

Key Server-Side Notes: #

• Correlation ID Preservation: The AmqpOutboundGateway uses the incoming amqp_correlationId header for the response. This is how the client matches the response to its original request.
• Dynamic Reply Queue: The server doesn't need a fixed reply queue—it uses the amqp_replyTo header from the request, which Spring AMQP's convertSendAndReceive automatically sets to a temporary queue.
• Header Propagation: DefaultAmqpHeaderMapper ensures critical RPC headers aren't lost between AMQP messages and Spring Integration messages.

The client uses RabbitTemplate's convertSendAndReceive method, which handles temporary reply queue setup, correlation, and response matching automatically.

2.1 Configuration #

@Configuration
public class RabbitRpcClientConfig {

    @Value("${rabbitmq.rpc.request.queue}")
    private String rpcRequestQueue;

    @Autowired
    private ConnectionFactory connectionFactory;

    @Bean
    public RabbitTemplate rabbitTemplate() {
        RabbitTemplate template = new RabbitTemplate(connectionFactory);
        // Set a timeout to avoid hanging indefinitely
        template.setReplyTimeout(5000);
        return template;
    }

    @Bean
    public Queue rpcRequestQueue() {
        return new Queue(rpcRequestQueue, false);
    }
}

2.2 Client Usage #

Inject the RabbitTemplate and call the RPC method:

@Service
public class RpcClientService {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Value("${rabbitmq.rpc.request.queue}")
    private String rpcRequestQueue;

    public String sendRpcRequest(String message) {
        // Send request and wait for the correlated response
        Object response = rabbitTemplate.convertSendAndReceive(rpcRequestQueue, message);
        return response != null ? response.toString() : null;
    }
}

If you hit bottlenecks earlier, these are the most likely issues:

• Missing Correlation: Double-check that the server preserves the amqp_correlationId header. If you use a custom header mapper, ensure it doesn't drop this or the amqp_replyTo header.
• Message Conversion Mismatch: Ensure both client and server use compatible message converters. For custom objects, configure Jackson2JsonMessageConverter on both sides.
• Timeout Mismatch: If the client gets timeout errors, adjust replyTimeout on the RabbitTemplate to match your server's processing latency.
• Queue Permissions: Verify that the server has permission to write to the temporary reply queue created by the client.

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