Spring Boot 消息队列与异步处理实战

接着，在 application.properties 中配置连接信息：

# 服务器端口
server.port=8080
# RabbitMQ 连接信息
spring.rabbitmq.host=localhost
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest

2. 编写核心代码

我们需要创建三个主要部分：配置类、生产者、消费者。

RabbitMQ 配置类：定义交换机、队列和绑定关系。

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RabbitMQConfig {

    @Bean
    public DirectExchange productExchange() {
        return new DirectExchange("product-exchange");
    }

    @Bean
    public Queue productQueue() {
        return new Queue("product-queue");
    }

    @Bean
    public Binding productBinding(Queue productQueue, DirectExchange productExchange) {
        return BindingBuilder.bind(productQueue).to(productExchange).with("product-routing-key");
    }
}

消息生产者：负责发送消息。

import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Component
public class ProductMessageProducer {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    private static final String EXCHANGE_NAME = "product-exchange";
    private static final String ROUTING_KEY = "product-routing-key";

    public void sendProductMessage(String message) {
        rabbitTemplate.convertAndSend(EXCHANGE_NAME, ROUTING_KEY, message);
        System.out.println("消息已发送：" + message);
    }
}

消息消费者：监听队列并处理业务逻辑。

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
public class ProductMessageConsumer {

    @RabbitListener(queues = "product-queue")
    public void receiveProductMessage(String message) {
        System.out.println("消息已接收：" + message);
        processProductMessage(message);
    }

    private void processProductMessage(String message) {
        // 模拟处理消息的业务逻辑
        try {
            Thread.sleep(2000);
            System.out.println("消息处理完成：" + message);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

控制器接口：提供 HTTP 入口触发发送。

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;

@RestController
@RequestMapping("/api/products")
public class ProductController {

    @Autowired
    private ProductMessageProducer productMessageProducer;

    @PostMapping("/send")
    public String sendProductMessage(@RequestBody String message) {
        productMessageProducer.sendProductMessage(message);
        return "消息已发送";
    }
}

启动应用后，访问 http://localhost:8080/api/products/send 并传入消息体，即可看到控制台输出发送及接收日志。由于消费者中有 2 秒的模拟延迟，你会明显感受到非阻塞的效果。

实际应用场景

在实际业务中，这种模式的应用非常广泛。比如订单系统中的支付回调处理、用户注册后的邮件通知、或者大数据量的批量导入任务等。通过将耗时操作剥离到消息队列后端处理，前端接口可以瞬间返回成功状态，极大提升了用户体验。

总之，掌握 Spring Boot 与消息队列的集成，是解决高并发场景下系统瓶颈的关键技能之一。根据具体业务需求选择合适的 MQ 组件和异步策略，能让你的架构更加稳健。