springboot中使用rabbitt的详细方法

RabbitMQ的示例，涉及到Direct、Fanout、Topic和Headers交换机以及普通队列、延迟队列和死信队列

在pom.xml文件中添加以下依赖：

<dependency>
   <groupId>org.springframework.boot</groupId>
   <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

配置RabbitMQ连接信息,在application.properties文件中添加以下配置：

spring.rabbitmq.host=localhost
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest

创建消息队列

创建一个普通的Direct交换机和队列：

@Configuration
public class RabbitMQConfig {
   @Bean
   public Queue queue() {
       return new Queue("directQueue", false);
   }
   @Bean
   public DirectExchange directExchange() {
       return new DirectExchange("directExchange");
   }
   @Bean
   public Binding binding(Queue queue, DirectExchange directExchange) {
       return BindingBuilder.bind(queue).to(directExchange).with("directRoutingKey");
   }
}

创建一个Fanout交换机和队列：

@Configuration
public class RabbitMQConfig {
   @Bean
   public Queue queueA() {
       return new Queue("fanoutQueueA", false);
   }
   @Bean
   public Queue queueB() {
       return new Queue("fanoutQueueB", false);
   }
   @Bean
   public FanoutExchange fanoutExchange() {
       return new FanoutExchange("fanoutExchange");
   }
   @Bean
   public Binding bindingA(Queue queueA, FanoutExchange fanoutExchange) {
       return BindingBuilder.bind(queueA).to(fanoutExchange);
   }
   @Bean
   public Binding bindingB(Queue queueB, FanoutExchange fanoutExchange) {
       return BindingBuilder.bind(queueB).to(fanoutExchange);
   }
}

创建一个Topic交换机和队列：

@Configuration
public class RabbitMQConfig {
   @Bean
   public Queue queueA() {
       return new Queue("topicQueueA", false);
   }
   @Bean
   public Queue queueB() {
       return new Queue("topicQueueB", false);
   }
   @Bean
   public TopicExchange topicExchange() {
       return new TopicExchange("topicExchange");
   }
   @Bean
   public Binding bindingA(Queue queueA, TopicExchange topicExchange) {
       return BindingBuilder.bind(queueA).to(topicExchange).with("topic.key.*");
   }
   @Bean
   public Binding bindingB(Queue queueB, TopicExchange topicExchange) {
       return BindingBuilder.bind(queueB).to(topicExchange).with("topic.#");
   }
}

创建一个Headers交换机和队列：

@Configuration
public class RabbitMQConfig {
   @Bean
   public Queue queue() {
       return new Queue("headerQueue", false);
   }
   @Bean
   public HeadersExchange headersExchange() {
       return new HeadersExchange("headerExchange");
   }
   @Bean
   public Binding binding(Queue queue, HeadersExchange headersExchange) {
       Map<String, Object> headers = new HashMap<>();
       headers.put("header1", "value1");
       headers.put("header2", "value2");
       return BindingBuilder.bind(queue).to(headersExchange).whereAll(headers).match();
   }
}

创建一个延迟队列和死信队列：

@Configuration
public class RabbitMQConfig {
   @Bean
   public Queue delayQueue() {
       Map<String, Object> args = new HashMap<>();
       args.put("x-message-ttl", 10000); // 消息过期时间为10秒
       args.put("x-dead-letter-exchange", "deadLetterExchange");
       args.put("x-dead-letter-routing-key", "deadLetterRoutingKey");
       return new Queue("delayQueue", false, false, false, args);
   }
   @Bean
   public Queue deadLetterQueue() {
       return new Queue("deadLetterQueue", false);
   }
   @Bean
   public DirectExchange directExchange() {
       return new DirectExchange("directExchange");
   }
   @Bean
   public DirectExchange deadLetterExchange() {
       return new DirectExchange("deadLetterExchange");
   }
   @Bean
   public Binding binding(Queue delayQueue, DirectExchange directExchange) {
       return BindingBuilder.bind(delayQueue).to(directExchange).with("delayRoutingKey");
   }
   @Bean
   public Binding deadLetterBinding(Queue deadLetterQueue, DirectExchange deadLetterExchange) {
       return BindingBuilder.bind(deadLetterQueue).to(deadLetterExchange).with("deadLetterRoutingKey");
   }
}

发送和接收消息

@Service
public class RabbitMQService {
   @Autowired
   private RabbitTemplate rabbitTemplate;
   public void sendMessage(String message) {
       rabbitTemplate.convertAndSend("directExchange", "directRoutingKey", message);
   }
   @RabbitListener(queues = "directQueue")
   public void receiveDirectMessage(String message) {
       System.out.println("Received message from directQueue: " + message);
   }
   public void sendFanoutMessage(String message) {
       rabbitTemplate.convertAndSend("fanoutExchange", "", message);
   }
   @RabbitListener(queues = "fanoutQueueA")
   public void receiveFanoutMessageA(String message) {
       System.out.println("Received message from fanoutQueueA: " + message);
   }
   @RabbitListener(queues = "fanoutQueueB")
   public void receiveFanoutMessageB(String message) {
       System.out.println("Received message from fanoutQueueB: " + message);
   }
   public void sendTopicMessage(String message) {
       rabbitTemplate.convertAndSend("topicExchange", "topic.key.message", message);
   }
   @RabbitListener(queues = "topicQueueA")
   public void receiveTopicMessageA(String message) {
       System.out.println("Received message from topicQueueA: " + message);
   }
   @RabbitListener(queues = "topicQueueB")
   public void receiveTopicMessageB(String message) {
       System.out.println("Received message from topicQueueB: " + message);
   }
   public void sendHeaderMessage(String message) {
       MessageProperties messageProperties = new MessageProperties();
       messageProperties.setHeader("header1", "value1");
       messageProperties.setHeader("header2", "value2");
       Message msg = MessageBuilder.withBody(message.getBytes()).andProperties(messageProperties).build();
       rabbitTemplate.send("headerExchange", "", msg);
   }
   @RabbitListener(queues = "headerQueue")
   public void receiveHeaderMessage(Message message) {
       System.out.println("Received message from headerQueue: " + new String(message.getBody()));
   }
   public void sendDelayMessage(String message) {
       rabbitTemplate.convertAndSend("directExchange", "delayRoutingKey", message);
   }
   @RabbitListener(queues = "deadLetterQueue")
   public void receiveDeadLetterMessage(String message) {
       System.out.println("Received message from deadLetterQueue: " + message);
   }
}

消息确认模式

在RabbitMQ中，消息确认模式有两种：自动确认和手动确认。自动确认是指当消息被成功接收时，RabbitMQ会自动确认消息。手动确认是指当消费者成功处理消息后，显式地向RabbitMQ发送确认消息。
1 使用手动确认模式：

@Configuration
public class RabbitMQConfig {
   @Bean
   public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory) {
       SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
       factory.setConnectionFactory(connectionFactory);
       factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
       return factory;
   }
}

@Service
public class RabbitMQService {
   @RabbitListener(queues = "directQueue")
   public void receiveDirectMessage(Message message, Channel channel) throws IOException {
       try {
           System.out.println("Received message from directQueue: " + new String(message.getBody()));
           channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
       } catch (Exception e) {
           channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
       }
   }
}

在上述代码中，我们使用了MANUAL（手动确认）模式，并在消息处理成功后使用channel.basicAck()方法显式地确认消息。如果发生任何异常，我们使用channel.basicNack()方法拒绝消息并重新加入队列。

希望这个示例能够帮助您了解如何在Spring Boot中使用RabbitMQ，并使用不同的交换机和队列类型以及消息确认模式。

来源：https://blog.csdn.net/u010480873/article/details/130797346