springboot中的消息队列和用法
消息队列简介
消息队列是一种异步通信机制，允许应用程序通过发送和接收消息进行解耦。在Spring Boot中，常用的消息队列实现包括RabbitMQ、Apache Kafka和ActiveMQ。
集成RabbitMQ
RabbitMQ是一个开源消息代理，支持多种消息协议。在Spring Boot中集成RabbitMQ需要添加依赖：
<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId> </dependency>配置RabbitMQ连接信息：
spring.rabbitmq.host=localhost spring.rabbitmq.port=5672 spring.rabbitmq.username=guest spring.rabbitmq.password=guest创建消息生产者：
@Service public class MessageProducer { @Autowired private RabbitTemplate rabbitTemplate; public void sendMessage(String exchange, String routingKey, String message) { rabbitTemplate.convertAndSend(exchange, routingKey, message); } }创建消息消费者：
@Component public class MessageConsumer { @RabbitListener(queues = "queue.name") public void receiveMessage(String message) { System.out.println("Received message: " + message); } }集成Apache Kafka
Kafka是一个分布式流处理平台，适合处理高吞吐量数据。添加Kafka依赖：
<dependency> <groupId>org.springframework.kafka</groupId> <artifactId>spring-kafka</artifactId> </dependency>配置Kafka连接信息：
spring.kafka.bootstrap-servers=localhost:9092 spring.kafka.consumer.group-id=group_id spring.kafka.consumer.auto-offset-reset=earliest创建Kafka生产者：
@Service public class KafkaProducer { @Autowired private KafkaTemplate<String, String> kafkaTemplate; public void sendMessage(String topic, String message) { kafkaTemplate.send(topic, message); } }创建Kafka消费者：
@Component public class KafkaConsumer { @KafkaListener(topics = "topic.name", groupId = "group_id") public void listen(String message) { System.out.println("Received message: " + message); } }消息队列使用场景
消息队列适用于异步处理、应用解耦、流量削峰等场景。例如在电商系统中，订单创建后通过消息队列通知库存系统减库存，而不是直接调用库存服务。
消息确认机制
为了保证消息可靠传递，Spring Boot支持消息确认机制。在RabbitMQ中配置生产者确认：
spring.rabbitmq.publisher-confirms=true spring.rabbitmq.publisher-returns=true在Kafka中配置手动提交偏移量：
spring.kafka.consumer.enable-auto-commit=false消息序列化
默认情况下，Spring Boot使用SimpleMessageConverter进行消息序列化。可以自定义消息转换器：
@Bean public MessageConverter jsonMessageConverter() { return new Jackson2JsonMessageConverter(); }错误处理
实现ErrorHandler接口可以处理消息消费过程中的异常：
@Bean public ErrorHandler errorHandler() { return (e, record) -> { // 自定义错误处理逻辑 }; }性能优化
对于高吞吐量场景，可以调整消费者并发数量：
spring.kafka.listener.concurrency=3RabbitMQ中可以预取消息数量：
spring.rabbitmq.listener.simple.prefetch=10消息队列实现票务系统
// 配置RabbitMQ连接 @Configuration public class RabbitMQConfig { @Value("${spring.rabbitmq.host}") private String host; @Value("${spring.rabbitmq.port}") private int port; @Bean public ConnectionFactory connectionFactory() { CachingConnectionFactory connectionFactory = new CachingConnectionFactory(); connectionFactory.setHost(host); connectionFactory.setPort(port); return connectionFactory; } @Bean public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) { return new RabbitTemplate(connectionFactory); } }// 定义票务消息DTO @Data @AllArgsConstructor @NoArgsConstructor public class TicketMessage { private String ticketId; private String eventName; private String userId; private LocalDateTime purchaseTime; private int quantity; }// 创建消息生产者服务 @Service public class TicketProducer { private static final String TICKET_QUEUE = "ticket.queue"; @Autowired private RabbitTemplate rabbitTemplate; public void sendTicketPurchase(TicketMessage ticketMessage) { rabbitTemplate.convertAndSend(TICKET_QUEUE, ticketMessage); System.out.println("Sent ticket purchase message: " + ticketMessage); } }// 创建消息消费者服务 @Service public class TicketConsumer { private static final String TICKET_QUEUE = "ticket.queue"; @RabbitListener(queues = TICKET_QUEUE) public void receiveTicketPurchase(TicketMessage ticketMessage) { System.out.println("Received ticket purchase message: " + ticketMessage); // 处理票务逻辑：更新库存 processTicket(ticketMessage); } private void processTicket(TicketMessage ticket) { // 实现票务处理逻辑 减库存，插数据 } }// 创建队列配置 @Configuration public class TicketQueueConfig { @Bean public Queue ticketQueue() { return new Queue("ticket.queue", true); } }异常处理
// 添加异常处理器 @Component public class RabbitMQErrorHandler implements RabbitListenerErrorHandler { @Override public Object handleError(Message amqpMessage, org.springframework.messaging.Message<?> message, ListenerExecutionFailedException exception) { System.err.println("Error processing message: " + exception.getMessage()); // 可添加重试或死信队列逻辑 return null; } }测试代码
// 测试消息发送 @SpringBootTest public class TicketSystemTest { @Autowired private TicketProducer ticketProducer; @Test public void testSendTicketMessage() { TicketMessage message = new TicketMessage( "TICKET-12345", "Concert", "USER-1001", LocalDateTime.now(), 2 ); ticketProducer.sendTicketPurchase(message); } }