项目中的问题

1.搜索与商品服务的问题

商品服务修改了商品的上架状态,商品就可以被搜索到.采用消息通知,商品服务修改完商品上架状态,发送消息给搜索服务,搜索服务消费消息,进行商品数据ES保存.下架也是一样.

2.订单服务取消订单问题

延迟队里

保存订单之后开始计时,时间到了,取消未支付的订单.

rabbitMQ的延时消息.

3.分布式事务问题

之前有一天专门讲过分布式事务,都是概念.其中有一个消息的最终数据一致性.

建议每个人回去把分布式事务的课件或者课程看一遍.

场景:

支付----------订单----------库存就是分布式事务场景.

4.秒杀的时候

使用rabbitMQ进行消息通知、进行用户的排队。

消息队列解决什么问题

消息队列都解决了什么问题？

1、异步

2、并行

3、解耦

4.排队削峰削去流量的峰值

消息队列工具 RabbitMQ

1 、常见MQ产品

- ActiveMQ：基于JMS（java协议）、消息类型比较少，2种一对一，一对多的

- RabbitMQ：基于AMQP协议，erlang语言开发，稳定性好，消息类型多

- RocketMQ：基于JMS，阿里巴巴产品，目前交由Apache基金会，使用也不少。

- Kafka：分布式消息系统，高吞吐量，消息准确性不好

2 、RabbitMQ基础概念

Broker：简单来说就是消息队列服务器实体

Exchange：消息交换机，它指定消息按什么规则，路由到哪个队列

Queue：消息队列载体，每个消息都会被投入到一个或多个队列

Binding：绑定，它的作用就是把 exchange和 queue按照路由规则绑定起来

Routing Key：路由关键字， exchange根据这个关键字进行消息投递

vhost：虚拟主机，一个 broker里可以开设多个 vhost，用作不同用户的权限分离

producer：消息生产者，就是投递消息的程序

consumer：消息消费者，就是接受消息的程序

channel：消息通道，在客户端的每个连接里，可建立多个 channel，每个 channel代表一个会话任务

3、消息模型

RabbitMQ提供了6种消息模型，但是第6种其实是RPC，并不是MQ，因此不予学习。那么也就剩下5种。

但是其实3、4、5这三种都属于订阅模型，只不过进行路由的方式不同。

基本消息模型：生产者–>队列–>消费者

work消息模型：生产者–>队列–>多个消费者共同消费

订阅模型-Fanout：广播模式，将消息交给所有绑定到交换机的队列，每个消费者都会收到同一条消息

订阅模型-Direct：定向，把消息交给符合指定 rotingKey 的队列

订阅模型-Topic 主题模式：通配符，把消息交给符合routing pattern（路由模式）的队列

消息不丢失

消息的准确性保证！！！！

消息的不丢失，在MQ角度考虑，一般有三种途径：

生产者不丢数据

生产者生产一个消息，准确的投递到交换机和队列中。

解决：rabbitMQ提供了消息的发送确认，交换机应答和队列应答。

消息生产出来向交互机中投递，如投递成功，会给true回执，失败了，会给false的回执。

交换机向队列路由消息的时候，如果消息没到达队列中，会触发队列应答，会执行对应应答的方法。

rabbitmq:

    host: 192.168.200.128

    port: 5672

    username: guest

    password: guest

    publisher-confirm-type: correlated #开启交换机应答

    publisher-returns: true   #队列应答

    listener:

      simple:

        acknowledge-mode: manual #默认情况下消息消费者是自动确认消息的，如果要手动确认消息则需要修改确认模式为manual

        prefetch: 1 # 消费者每次从队列获取的消息数量。此属性当不设置时为：轮询分发，设置为1为：公平分发

prefetch: 1 消费者每次从队列中消费消息的数量。

轮询分发：多个消费者的情况下，一人一次，不管消费方上次的消息有没有消费完，只要轮谁了就给谁。

公平分发：多个消费者的情况下，一人一次，如果轮到这个消费方了，但是它上一个消息还没有消费完，这个消息就给别人。只有上一个消息消费完了，才能消费下一个消息。

2.MQ服务器不丢数据

消息队列数据存在内存中，MQ挂了重启，内存容易释放，消息就没了。

提供了持久化，可以对交换机、队列、消息进行持久化。

3.消费者不丢数据

消费者在消费消息的过程中，可能会出现问题，消息还得存在。

手动签收消息，一旦消息消费过程中出问题了，可以拒绝签收，把消息转发到另外一个队列或者把消费异常的消息做记录，等到问题解决了，再重新投递。

保证消息不丢失有两种实现方式：

开启事务模式
消息确认模式

说明：开启事务会大幅降低消息发送及接收效率，使用的相对较少。

在投递消息时开启事务支持，如果消息投递失败，则回滚事务，但是，很少有人这么干，因为这是同步操作，一条消息发送之后会使发送端阻塞，以等待RabbitMQ-Server的回应，之后才能继续发送下一条消息，生产者生产消息的吞吐量和性能都会大大降低。

因此我们生产环境一般都采取消息确认模式。

1、消息持久化

如果希望RabbitMQ重启之后消息不丢失，那么需要对以下3种实体均配置持久化

Exchange

声明exchange时设置持久化（durable = true）并且不自动删除(autoDelete = false)

Queue

声明queue时设置持久化（durable = true）并且不自动删除(autoDelete = false)

message

发送消息时通过设置deliveryMode=2持久化消息

说明：

@Queue: 当所有消费客户端连接断开后，是否自动删除队列

true：删除 false：不删除

@Exchange：当所有绑定队列都不在使用时，是否自动删除交换器

true：删除 false：不删除

2、发送确认

有时，业务处理成功，消息也发了，但是我们并不知道消息是否成功到达了rabbitmq，如果由于网络等原因导致业务成功而消息发送失败，那么发送方将出现不一致的问题，此时可以使用rabbitmq的发送确认功能，即要求rabbitmq显式告知我们消息是否已成功发送。

3、手动消费确认

有时，消息被正确投递到消费方，但是消费方处理失败，那么便会出现消费方的不一致问题。比如订单已创建的消息发送到用户积分子系统中用于增加用户积分，但是积分消费方处理却失败了，用户就会问：我购买了东西为什么积分并没有增加呢？

要解决这个问题，需要引入消费方确认，即只有消息被成功处理之后才告知rabbitmq以ack，否则告知rabbitmq以nack

商品搜索上下架

1、service-product发送消息

我在商品上架与商品添加时发送消息

商品上架

实现类

@Override

@Transactional

  public void onSale(Long skuId) {

    // 更改销售状态

    SkuInfo skuInfoUp = new SkuInfo();

    skuInfoUp.setId(skuId);

    skuInfoUp.setIsSale(1);

    skuInfoMapper.updateById(skuInfoUp);

    //商品上架

    rabbitService.sendMessage(MqConst.EXCHANGE_DIRECT_GOODS, MqConst.ROUTING_GOODS_UPPER, skuId);

}

商品下架

实现类

@Override

@Transactional

  public void cancelSale(Long skuId) {

    // 更改销售状态

    SkuInfo skuInfoUp = new SkuInfo();

    skuInfoUp.setId(skuId);

    skuInfoUp.setIsSale(0);

    skuInfoMapper.updateById(skuInfoUp);

    //商品下架

    rabbitService.sendMessage(MqConst.EXCHANGE_DIRECT_GOODS, MqConst.ROUTING_GOODS_LOWER, skuId);

}

2、service-list消费消息

package com.atguigu.gmall.list.receiver;

  @Component

  public class ListReceiver {

    @Autowired

    private SearchService searchService;

    /**

     * 商品上架

     * @param skuId

     * @throws IOException

     */

    @RabbitListener(bindings = @QueueBinding(

            value = @Queue(value = MqConst.QUEUE_GOODS_UPPER, durable = "true"),

            exchange = @Exchange(value = MqConst.EXCHANGE_DIRECT_GOODS, type = ExchangeTypes.DIRECT, durable = "true"),

            key = {MqConst.ROUTING_GOODS_UPPER}

    ))

    public void upperGoods(Long skuId, Message message, Channel channel) throws IOException {

        if (null != skuId) {

            searchService.upperGoods(skuId);

        }

        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);

    }

    /**

     * 商品下架

        * @param skuId

     */

    @RabbitListener(bindings = @QueueBinding(

            value = @Queue(value = MqConst.QUEUE_GOODS_LOWER, durable = "true"),

            exchange = @Exchange(value = MqConst.EXCHANGE_DIRECT_GOODS, type = ExchangeTypes.DIRECT, durable = "true"),

            key = {MqConst.ROUTING_GOODS_LOWER}

    ))

    public void lowerGoods(Long skuId, Message message, Channel channel) throws IOException {

        if (null != skuId) {

            searchService.lowerGoods(skuId);

        }

        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);

    }

}

延迟队列关闭过期订单

延迟消息有两种实现方案：

1.基于死信队列

使用消息的存活时间，给队列设置参数，时间到了转发到另外一个队列，消费者真正消费的是另一个队列。

2.集成延迟插件

延迟插件中有个小型的库，延迟交换机把消息消息先存在小型的库中，时间到了再转发到队列。

区别：

死信的延迟消息：要求消息的时间一致，如果不一致，后面的消息出不来。

延迟插件的：没有这个要求。消息的时间可以不一致。

1 、基于死信实现延迟消息

使用RabbitMQ来实现延迟消息必须先了解RabbitMQ的两个概念：消息的TTL和死信Exchange，通过这两者的组合来实现延迟队列

1.1、消息的TTL（Time To Live）

消息的TTL就是消息的存活时间。RabbitMQ可以对队列和消息分别设置TTL。对队列设置就是队列没有消费者连着的保留时间，也可以对每一个单独的消息做单独的设置。超过了这个时间，我们认为这个消息就死了，称之为死信。

我们创建一个队列queue.temp，在Arguments 中添加x-message-ttl 为5000 （单位是毫秒），那所在压在这个队列的消息在5秒后会消失。

1.2、死信交换器 Dead Letter Exchanges

一个消息在满足如下条件下，会进死信路由，记住这里是路由而不是队列，一个路由可以对应很多队列。

（1）一个消息被Consumer拒收了，并且reject方法的参数里requeue是false。也就是说不会被再次放在队列里，被其他消费者使用。

（2）上面的消息的TTL到了，消息过期了。

（3）队列的长度限制满了。排在前面的消息会被丢弃或者扔到死信路由上。

Dead Letter Exchange其实就是一种普通的exchange，和创建其他exchange没有两样。只是在某一个设置了Dead Letter Exchange的队列中有消息过期了，会自动触发消息的转发，发送到Dead Letter Exchange中去。

2 、基于延迟插件实现延迟消息

2.1、插件安装

1. 首先我们将刚下载下来的rabbitmq_delayed_message_exchange-3.8.0.ez文件上传到RabbitMQ所在服务器，下载地址：https://www.rabbitmq.com/community-plugins.html

2. 切换到插件所在目录，执行 docker cp rabbitmq_delayed_message_exchange-3.8.0.ez rabbitmq:/plugins 命令，将刚插件拷贝到容器内plugins目录下

3. 执行 docker exec -it rabbitmq /bin/bash 命令进入到容器内部，并 cd plugins 进入plugins目录

4. 执行 ls -l|grep delay 命令查看插件是否copy成功

5. 在容器内plugins目录下，执行 rabbitmq-plugins enable rabbitmq_delayed_message_exchange 命令启用插件

6. exit命令退出RabbitMQ容器内部，然后执行 docker restart rabbitmq 命令重启RabbitMQ容器

2.2、代码实现

配置队列

package com.atguigu.gmall.mq.config;

  

  

  @Configuration

  public class DelayedMqConfig {

  

    public static final String exchange_delay = "exchange.delay";

    public static final String routing_delay = "routing.delay";

    public static final String queue_delay_1 = "queue.delay.1";

  

    /**

     * 队列不要在RabbitListener上面做绑定，否则不会成功，如队列2，必须在此绑定

     *

     * @return

     */

  

    @Bean

    public Queue delayQeue1() {

        // 第一个参数是创建的queue的名字，第二个参数是是否支持持久化

        return new Queue(queue_delay_1, true);

    }

  

    @Bean

    public CustomExchange delayExchange() {

        Map<String, Object> args = new HashMap<String, Object>();

        args.put("x-delayed-type", "direct");

        return new CustomExchange(exchange_delay, "x-delayed-message", true, false, args);

    }

  

    @Bean

    public Binding delayBbinding1() {

        return BindingBuilder.bind(delayQeue1()).to(delayExchange()).with(routing_delay).noargs();

    }

  

}

发送消息

@GetMapping("sendDelay")

  public Result sendDelay() {

   SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

   this.rabbitTemplate.convertAndSend(DelayedMqConfig.exchange_delay, DelayedMqConfig.routing_delay, sdf.format(new Date()), new MessagePostProcessor() {

      @Override

      public Message postProcessMessage(Message message) throws AmqpException {

         message.getMessageProperties().setDelay(10 * 1000);

         System.out.println(sdf.format(new Date()) + " Delay sent.");

         return message;

      }

   });

   return Result.ok();

}

接收消息

package com.atguigu.gmall.mq.receiver;

  

  

  @Component

@Configuration

  public class DelayReceiver {

  

    @RabbitListener(queues = DelayedMqConfig.queue_delay_1)

    public void get(String msg) {

        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

        System.out.println("Receive queue_delay_1: " + sdf.format(new Date()) + " Delay rece." + msg);

    }

  

}

3 、基于延迟插件实现取消订单

rabbit-util模块延迟接口封装

RabbitService
/**

 * 发送延迟消息

 * @param exchange 交换机

 * @param routingKey 路由键

 * @param message 消息

 * @param delayTime 单位：秒

 */

  public boolean sendDelayMessage(String exchange, String routingKey, Object message, int delayTime) {

    GmallCorrelationData correlationData = new GmallCorrelationData();

    String correlationId = UUID.randomUUID().toString();

    correlationData.setId(correlationId);

    correlationData.setMessage(message);

    correlationData.setExchange(exchange);

    correlationData.setRoutingKey(routingKey);

    correlationData.setDelay(true);

    correlationData.setDelayTime(delayTime);

  

    redisTemplate.opsForValue().set(correlationId, JSON.toJSONString(correlationData), OBJECT_TIMEOUT, TimeUnit.MINUTES);

    this.rabbitTemplate.convertAndSend(exchange, routingKey, message, new MessagePostProcessor() {

        @Override

        public Message postProcessMessage(Message message) throws AmqpException {

            message.getMessageProperties().setDelay(delayTime*1000);

            return message;

        }

    },correlationData);

    return true;

}

3.1、发送消息

创建订单时，发送延迟消息

修改保存订单方法

@Override

@Transactional

  public Long saveOrderInfo(OrderInfo orderInfo) {

    // orderInfo

    // 总金额，订单状态，用户Id，第三方交易编号，创建时间，过期时间，进程状态

    orderInfo.sumTotalAmount();

    orderInfo.setOrderStatus(OrderStatus.UNPAID.name());

    String outTradeNo = "ATGUIGU" + System.currentTimeMillis() + "" + new Random().nextInt(1000);

    orderInfo.setOutTradeNo(outTradeNo);

    orderInfo.setCreateTime(new Date());

    // 定义为1天

    Calendar calendar = Calendar.getInstance();

    calendar.add(Calendar.DATE, 1);

    orderInfo.setExpireTime(calendar.getTime());

  

    orderInfo.setProcessStatus(ProcessStatus.UNPAID.name());

    orderInfoMapper.insert(orderInfo);

  

    StringBuffer tradeBody = new StringBuffer();

    // 保存订单明细

    List<OrderDetail> orderDetailList = orderInfo.getOrderDetailList();

    for (OrderDetail orderDetail : orderDetailList) {

        orderDetail.setId(null);

        orderDetail.setOrderId(orderInfo.getId());

        orderDetailMapper.insert(orderDetail);

  

        tradeBody.append(orderDetail.getSkuName()).append(" ");

    }

  

    //更新支付描述

    orderInfo.setTradeBody(tradeBody.toString());

    orderInfoMapper.updateById(orderInfo);

    //发送延迟队列，如果到过期时间了未支付，取消订单

 rabbitService.sendDelayMessage(MqConst.EXCHANGE_DIRECT_ORDER_CANCEL, MqConst.ROUTING_ORDER_CANCEL, orderInfo.getId(), MqConst.DELAY_TIME);

    // 返回

    return orderInfo.getId();

}

3.2、接收消息

package com.atguigu.gmall.order.receiver;

  @Component

  public class OrderReceiver {

    @Autowired

    private OrderService orderService;

    /**

     * 取消订单消费者

       * 延迟队列，不能再这里做交换机与队列绑定
* 需要在配置类里去绑定死信交换机
     * @param orderId
     * @throws IOException
     */
    @RabbitListener(queues = MqConst.QUEUE_ORDER_CANCEL)
    public void orderCancel(Long orderId, Message message, Channel channel) throws IOException {
        if (null != orderId) {
            //防止重复消费
            OrderInfo orderInfo = orderService.getById(orderId);
            if (null != orderInfo && orderInfo.getOrderStatus().equals(ProcessStatus.UNPAID.getOrderStatus().name())) {
                orderService.execExpiredOrder(orderId);
            }
        }
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    }

}

取消订单业务，取消订单要关闭支付交易

@Override

  public void execExpiredOrder(Long orderId) {

    // orderInfo

    updateOrderStatus(orderId, ProcessStatus.CLOSED);

    // paymentInfo

    //paymentFeignClient.closePayment(orderId);

    //发送取消交易的消息

    rabbitService.sendMessage(MqConst.EXCHANGE_DIRECT_PAYMENT_CLOSE, MqConst.ROUTING_PAYMENT_CLOSE, orderId);

}

关闭交易消息消费者

package com.atguigu.gmall.payment.receiver;

@Component

public class PaymentReceiver {

    @Autowired

    private PaymentService paymentService;

    /**

     * 取消交易

     * @param orderId

     * @throws IOException

     */

    @RabbitListener(bindings = @QueueBinding(

            value = @Queue(value = MqConst.QUEUE_PAYMENT_CLOSE, durable = "true"),

            exchange = @Exchange(value = MqConst.EXCHANGE_DIRECT_PAYMENT_CLOSE),

            key = {MqConst.ROUTING_PAYMENT_CLOSE}

    ))

    public void closePayment(Long orderId) throws IOException {

        if (null != orderId) {

            paymentService.closePayment(orderId);

        }

    }

}

更改支付日志表，状态为关闭交易

@Override

public void closePayment(Long orderId) {

    QueryWrapper<PaymentInfo> queryWrapper = new QueryWrapper<>();

    queryWrapper.eq("order_id", orderId);

    PaymentInfo paymentInfoUp = new PaymentInfo();

    paymentInfoUp.setPaymentStatus(PaymentStatus.ClOSED.name());

    paymentInfoMapper.update(paymentInfoUp, queryWrapper);

    //关闭交易

     alipayService.closePay(orderId);
}
支付宝支付AlipayServiceImpl实现类
/***

 * 关闭交易

 * @param orderId

 * @return

 */

@Override

public Boolean closePay(Long orderId) {

    OrderInfo orderInfo = orderFeignClient.getOrderInfo(orderId);



    //AlipayClient alipayClient = new DefaultAlipayClient("https://openapi.alipay.com/gateway.do","app_id","your private_key","json","GBK","alipay_public_key","RSA2");

    AlipayTradeCloseRequest request = new AlipayTradeCloseRequest();

    HashMap<String, Object> map = new HashMap<>();

    map.put("trade_no", "");

    map.put("out_trade_no", orderInfo.getOutTradeNo());

    map.put("operator_id", "YX01");



    request.setBizContent(JSON.toJSONString(map));

    AlipayTradeCloseResponse response = null;

    try {

        response = alipayClient.execute(request);

    } catch (AlipayApiException e) {

        e.printStackTrace();

    }

    if(response.isSuccess()){

        log.info("调用成功");

        return true;

    }

    return false;

}

项目中分布式事务的业务场景

RabbitMQ常见问题

1、使用RabbitMQ有什么好处？

1）.解耦：系统A在代码中直接调用系统B和系统C的代码，如果将来D系统接入，系统A还需要修改代码，过于麻烦！

2）.异步：将消息写入消息队列，非必要的业务逻辑以异步的方式运行，加快响应速度

3）.削峰：并发量大的时候，所有的请求直接怼到数据库，造成数据库连接异常

2、消息顺序问题

场景：比如支付操作，支付成功之后，会发送修改订单状态和扣减库存的消息，如果这两个消息同时发送，就不能保证完全按照顺序消费，有可能是先减库存了，后更改订单状态。

解决方案：同步执行，当一个消息执行完之后，再发布下一个消息。

3、如何保证RabbitMQ消息的可靠传输？

消息不可靠的原因是因为消息丢失

生产者丢失消息:

RabbitMQ提供transaction事务和confirm模式来确保生产者不丢消息；

Transaction事务机制就是说：发送消息前，开启事务（channel.txSelect()）,然后发送消息，如果发送过程中出现什么异常，事务就会回滚（channel.txRollback()）,如果发送成功则提交事务（channel.txCommit()），然而，这种方式有个缺点：吞吐量下降。

confirm模式用的居多：一旦channel进入confirm模式，所有在该信道上发布的消息都将会被指派一个唯一的ID（从1开始），一旦消息被投递到所有匹配的队列之后；

rabbitMQ就会发送一个ACK给生产者（包含消息的唯一ID），这就使得生产者知道消息已经正确到达目的队列了；

如果rabbitMQ没能处理该消息，则会发送一个Nack消息给你，可以进行重试操作。

消息列表丢失消息:

可以消息持久化, 即使rabbitMQ挂了，重启后也能恢复数据

消费者丢失消息:

消费者丢数据一般是因为采用了自动确认消息模式，消费者在收到消息之后，处理消息之前，会自动回复RabbitMQ已收到消息；如果这时处理消息失败，就会丢失该消息；改为手动确认消息即可！