消息中间件RabbitMQ的使用

        
            org.springframework.boot
            spring-boot-starter-amqp
        

package com.atguigu.gulimall.order.config;

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.Exchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.HashMap;




@Configuration
public class MyRabbitMQConfig {

    

    @Bean
    public Queue orderDelayQueue() {
        
        HashMap arguments = new HashMap<>();
        arguments.put("x-dead-letter-exchange", "order-event-exchange");
        arguments.put("x-dead-letter-routing-key", "order.release.order");
        arguments.put("x-message-ttl", 60000); // 消息过期时间 1分钟
        Queue queue = new Queue("order.delay.queue", true, false, false, arguments);

        return queue;
    }

    
    @Bean
    public Queue orderReleaseQueue() {

        Queue queue = new Queue("order.release.order.queue", true, false, false);

        return queue;
    }

    
    @Bean
    public Exchange orderEventExchange() {
        
        return new TopicExchange("order-event-exchange", true, false);

    }


    @Bean
    public Binding orderCreateBinding() {
        
        return new Binding("order.delay.queue",
                Binding.DestinationType.QUEUE,
                "order-event-exchange",
                "order.create.order",
                null);
    }

    @Bean
    public Binding orderReleaseBinding() {

        return new Binding("order.release.order.queue",
                Binding.DestinationType.QUEUE,
                "order-event-exchange",
                "order.release.order",
                null);
    }

    
    @Bean
    public Binding orderReleaseOtherBinding() {

        return new Binding("stock.release.stock.queue",
                Binding.DestinationType.QUEUE,
                "order-event-exchange",
                "order.release.other.#",
                null);
    }


    
    @Bean
    public Queue orderSecKillOrrderQueue() {
        Queue queue = new Queue("order.seckill.order.queue", true, false, false);
        return queue;
    }

    @Bean
    public Binding orderSecKillOrrderQueueBinding() {
        //String destination, DestinationType destinationType, String exchange, String routingKey,
        // 			Map arguments
        Binding binding = new Binding(
                "order.seckill.order.queue",
                Binding.DestinationType.QUEUE,
                "order-event-exchange",
                "order.seckill.order",
                null);

        return binding;
    }


}

  //TODO 确保每个消息发送成功，给每个消息做好日志记录，(给数据库保存每一个详细信息)保存每个消息的详细信息
     rabbitTemplate.convertAndSend("order-event-exchange", "order.release.other", orderTo);

package com.atguigu.gulimall.order.listener;

import com.rabbitmq.client.Channel;
import com.atguigu.gulimall.order.entity.OrderEntity;
import com.atguigu.gulimall.order.service.OrderService;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.io.IOException;



@RabbitListener(queues = "order.release.order.queue")
@Service
public class OrderCloseListener {

    @Autowired
    private OrderService orderService;

    @RabbitHandler
    public void listener(OrderEntity orderEntity, Channel channel, Message message) throws IOException {
        System.out.println("收到过期的订单信息，准备关闭订单" + orderEntity.getOrderSn());
        try {
            orderService.closeOrder(orderEntity);
            channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
        } catch (Exception e) {
            channel.basicReject(message.getMessageProperties().getDeliveryTag(),true);
        }

    }

}

总结：1. 引入依赖

           2. 创建交换机、队列、并将交换机与队列之间根据路由键（路由规则）进行绑定。

           3. 生产者发送消息：指定发送给哪个交换机、路由规则、携带的消息

           4. 消费者：指定监听的队列，实时监听这个队列消息，有就开始消费。并手动ack机制，保证消息确实已经被消费，或如果消息消费失败重新消费，防止消息的丢失。

           5.  消息丢失主要在三个过程中： （1）发送者发送给交换机的时候；解决方法confirmCallback 回调机制；（2） 交换机发送给消息队列的时候；解决方法returnCallback回调机制。（3）消费者端导致消息的丢失；解决方法手动ack机制；当然也有其他宕机等重大事故，这些一般就是主从机制。

## 开启发送端消息抵达Broker确认
#spring.rabbitmq.publisher-confirms=true
## 开启发送端消息抵达Queue确认
#spring.rabbitmq.publisher-returns=true
## 只要消息抵达Queue，就会异步发送优先回调returnfirm
#spring.rabbitmq.template.mandatory=true
## 手动ack消息，不使用默认的消费端确认
#spring.rabbitmq.listener.simple.acknowledge-mode=manual

package com.atguigu.gulimall.seckill.config;

import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;



@Configuration
public class MyRabbitMQConfig {

    // @Resource
    // private RabbitTemplate rabbitTemplate;

    @Bean
    public MessageConverter messageConverter() {
        return new Jackson2JsonMessageConverter();
    }



}