如何实现RabbitMQ生产者向RestController返回响应？

要实现消费者将保存后的对象返回给REST接口，核心是使用RabbitMQ的请求-响应（RPC）模式，让生产者发送消息后等待消费者的回复，再将回复作为HTTP响应返回。以下是具体实现步骤：

一、生产者（REST应用）改造 #

1. 配置RabbitTemplate与消息转换器 #

确保消息能以JSON格式序列化/反序列化，避免类型转换问题：

@Configuration
public class RabbitProducerConfig {
    @Bean
    public MessageConverter jsonMessageConverter() {
        return new Jackson2JsonMessageConverter();
    }

    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory, MessageConverter messageConverter) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate.setMessageConverter(messageConverter);
        // 设置回复超时时间（单位：毫秒），防止请求长时间挂起
        rabbitTemplate.setReplyTimeout(5000);
        return rabbitTemplate;
    }
}

2. 修改REST接口，发送消息并等待回复 #

使用RabbitTemplate.convertSendAndReceive()方法发送消息，该方法会阻塞等待消费者的响应：

@RestController
public class EntityController {
    private final RabbitTemplate rabbitTemplate;

    // 构造注入RabbitTemplate
    public EntityController(RabbitTemplate rabbitTemplate) {
        this.rabbitTemplate = rabbitTemplate;
    }

    @GetMapping("/")
    public ResponseEntity<YourEntity> createEntity() {
        // 构造创建实体的请求参数（可根据业务自定义DTO）
        CreateEntityRequest request = new CreateEntityRequest("测试字段1", "测试字段2");

        // 发送消息到example-queue，并等待回复
        YourEntity savedEntity = (YourEntity) rabbitTemplate.convertSendAndReceive(
            "",          // 使用默认交换机
            "example-queue", // 目标队列
            request
        );

        // 处理超时或无响应的情况
        if (savedEntity == null) {
            return ResponseEntity.status(HttpStatus.REQUEST_TIMEOUT).build();
        }

        return ResponseEntity.ok(savedEntity);
    }
}

二、消费者（第二个应用）改造 #

1. 配置消息转换器 #

同样配置JSON消息转换器，保证和生产者序列化规则一致：

@Configuration
public class RabbitConsumerConfig {
    @Bean
    public MessageConverter jsonMessageConverter() {
        return new Jackson2JsonMessageConverter();
    }
}

2. 修改RabbitListener方法，直接返回保存后的实体 #

Spring AMQP会自动将方法返回值作为回复消息，发送回生产者的临时回复队列：

@Component
public class EntityConsumer {
    private final YourEntityRepository entityRepository;

    // 构造注入Repository
    public EntityConsumer(YourEntityRepository entityRepository) {
        this.entityRepository = entityRepository;
    }

    @RabbitListener(queues = "example-queue")
    public YourEntity handleCreateEntity(CreateEntityRequest request) {
        // 将请求参数转换为实体对象
        YourEntity entity = new YourEntity();
        entity.setField1(request.getField1());
        entity.setField2(request.getField2());

        // 保存到数据库并返回
        return entityRepository.save(entity);
    }
}

三、关键注意事项 #

• 确保YourEntity和CreateEntityRequest类在两个应用中结构完全一致（可提取为共享依赖包避免重复代码）。
• 若需要自定义回复队列，可在消费者方法上添加@SendTo("指定回复队列名")，并在生产者中配置监听该队列，但默认临时队列更灵活，无需额外配置队列。
• 根据业务场景调整replyTimeout时间，避免超时过短导致正常请求失败，或过长导致请求阻塞。

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