谷粒商城项目篇12_分布式高级篇_购物车功能、消息队列RabbitMQ

标签：12 return String 队列 RabbitMQ 购物车消息 import public

读多写少，在线、游客购物车都考虑redis来存储
使用hash数据类型来存储商品信息，方便购物车商品的增删改查、选中等状态
游客购物车的实现原理是浏览器客户端cookie里面存储user-key，标识用户身份，第一次使用购物车都会分配一个user-key，然后存储redis
登录购物车的实现原理是Session是否存在判断用户是否登录，然后存储redis

1.vo的编写

package henu.soft.xiaosi.cart.vo;

import java.math.BigDecimal;
import java.util.List;

public class CartVo {

    /**
     * 购物车子项信息
     */
    List<CartItemVo> items;

    /**
     * 商品数量
     */
    private Integer countNum;

    /**
     * 商品类型数量
     */
    private Integer countType;

    /**
     * 商品总价
     */
    private BigDecimal totalAmount;

    /**
     * 减免价格
     */
    private BigDecimal reduce = new BigDecimal("0.00");

    public List<CartItemVo> getItems() {
        return items;
    }

    public void setItems(List<CartItemVo> items) {
        this.items = items;
    }

    public Integer getCountNum() {
        int count=0;
        if (items != null && items.size() > 0) {
            for (CartItemVo item : items) {
                count += item.getCount();
            }
        }
        return count;
    }

    public void setCountNum(Integer countNum) {
        this.countNum = countNum;
    }

    public Integer getCountType() {
        int count=0;
        if (items != null && items.size() > 0) {
            for (CartItemVo item : items) {
                count += 1;
            }
        }
        return count;
    }

    public void setCountType(Integer countType) {
        this.countType = countType;
    }

    public BigDecimal getTotalAmount() {
        BigDecimal total = new BigDecimal(0);
        if (items != null && items.size() > 0) {
            for (CartItemVo item : items) {
                total.add(item.getTotalPrice());
            }
        }
        total.subtract(reduce);
        return total;
    }

    public void setTotalAmount(BigDecimal totalAmount) {
        this.totalAmount = totalAmount;
    }

    public BigDecimal getReduce() {
        return reduce;
    }

    public void setReduce(BigDecimal reduce) {
        this.reduce = reduce;
    }
}

package henu.soft.xiaosi.cart.vo;

import java.math.BigDecimal;
import java.util.List;

public class CartItemVo {

    private Long skuId;

    private Boolean check = true;

    private String title;

    private String image;

    /**
     * 商品套餐属性
     */
    private List<String> skuAttrValues;

    private BigDecimal price;

    private Integer count;

    private BigDecimal totalPrice;

    public Long getSkuId() {
        return skuId;
    }

    public void setSkuId(Long skuId) {
        this.skuId = skuId;
    }

    public Boolean getCheck() {
        return check;
    }

    public void setCheck(Boolean check) {
        this.check = check;
    }

    public String getTitle() {
        return title;
    }

    public void setTitle(String title) {
        this.title = title;
    }

    public String getImage() {
        return image;
    }

    public void setImage(String image) {
        this.image = image;
    }

    public List<String> getSkuAttrValues() {
        return skuAttrValues;
    }

    public void setSkuAttrValues(List<String> skuAttrValues) {
        this.skuAttrValues = skuAttrValues;
    }

    public BigDecimal getPrice() {
        return price;
    }

    public void setPrice(BigDecimal price) {
        this.price = price;
    }

    public Integer getCount() {
        return count;
    }

    public void setCount(Integer count) {
        this.count = count;
    }

    /**
     * 当前购物车项总价等于单价x数量
     * @return
     */
    public BigDecimal getTotalPrice() {
        return price.multiply(new BigDecimal(count));
    }

    public void setTotalPrice(BigDecimal totalPrice) {
        this.totalPrice = totalPrice;
    }
}


package henu.soft.xiaosi.cart.vo;

import lombok.Data;

import java.math.BigDecimal;

@Data
public class SkuInfoVo {

    private Long skuId;
    /**
     * spuId
     */
    private Long spuId;
    /**
     * sku名称
     */
    private String skuName;
    /**
     * sku介绍描述
     */
    private String skuDesc;
    /**
     * 所属分类id
     */
    private Long catalogId;
    /**
     * 品牌id
     */
    private Long brandId;
    /**
     * 默认图片
     */
    private String skuDefaultImg;
    /**
     * 标题
     */
    private String skuTitle;
    /**
     * 副标题
     */
    private String skuSubtitle;
    /**
     * 价格
     */
    private BigDecimal price;
    /**
     * 销量
     */
    private Long saleCount;

}

2.编写interceptor绑定user-key

在执行目标方法前，判断用户的登录状态，封装绑定user-key传递给controller目标请求，然后设置cookie

已经登录：此时session存在，将userId作为user-key
没有登录：第一次访问帮忙创建一个user-key作为当前用户标识

用户状态封装

package henu.soft.xiaosi.cart.to;

import lombok.Data;

@Data
public class UserInfoTo {

    private Long userId;

    private String userKey;

    /**
     * 浏览器是否已有user-key
     */
    private Boolean tempUser = false;

}

配置SpringMVC自定义拦截器

package henu.soft.xiaosi.cart.config;

import henu.soft.xiaosi.cart.interceptor.CartInterceptor;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.servlet.config.annotation.InterceptorRegistry;
import org.springframework.web.servlet.config.annotation.WebMvcConfigurer;

@Configuration
public class MyWebConfig implements WebMvcConfigurer {

    /**
     * 添加自定义拦截器
     * @param registry
     */
    @Override
    public void addInterceptors(InterceptorRegistry registry) {
        registry.addInterceptor(new CartInterceptor()).addPathPatterns("/**");
    }
}

谷粒商城项目篇12_分布式高级篇_购物车功能、消息队列RabbitMQ_List

1.线程共享数据

// 线程共享变量public static ThreadLocal<UserInfoTo> threadLocal = new ThreadLocal<>();

因为tomcat的一个线程会从拦截器、controller、service、dao从始到终的执行，想要各个线程之间能够数据共享

就要使用ThreadLocal，类型为Map<Thread,Object> threadLocal，在controller之前设置用户状态信息UserInfoTo

package henu.soft.xiaosi.cart.interceptor;


import com.alibaba.cloud.commons.lang.StringUtils;
import henu.soft.common.constant.AuthServerConstant;
import henu.soft.common.constant.CartConstant;
import henu.soft.common.to.MemberResponseTo;
import henu.soft.xiaosi.cart.to.UserInfoTo;

import org.springframework.web.servlet.HandlerInterceptor;
import org.springframework.web.servlet.ModelAndView;

import javax.servlet.http.Cookie;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import javax.servlet.http.HttpSession;
import java.util.UUID;

public class CartInterceptor implements HandlerInterceptor {

	// 线程共享变量
    public static ThreadLocal<UserInfoTo> threadLocal = new ThreadLocal<>();

    @Override
    public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) throws Exception {
        HttpSession session = request.getSession();
        MemberResponseTo memberResponseTo = (MemberResponseTo) session.getAttribute(AuthServerConstant.LOGIN_USER);
        UserInfoTo userInfoTo = new UserInfoTo();
        //1 用户已经登录，设置userId
        if (memberResponseTo != null) {
            userInfoTo.setUserId(memberResponseTo.getId());
        }

        Cookie[] cookies = request.getCookies();
        if (cookies != null && cookies.length > 0) {
            for (Cookie cookie : cookies) {
                //2 如果浏览器中已经有user-Key，则直接设置
                if (cookie.getName().equals(CartConstant.TEMP_USER_COOKIE_NAME)) {
                    userInfoTo.setUserKey(cookie.getValue());
                    userInfoTo.setTempUser(true);
                }
            }
        }

        //3 如果浏览器没有user-key，我们通过uuid生成user-key
        if (StringUtils.isEmpty(userInfoTo.getUserKey())) {
            String uuid = UUID.randomUUID().toString();
            userInfoTo.setUserKey(uuid);
        }

        threadLocal.set(userInfoTo);
        return true;
    }

	// 绑定user-key
    @Override
    public void postHandle(HttpServletRequest request, HttpServletResponse response, Object handler, ModelAndView modelAndView) throws Exception {
        UserInfoTo userInfoTo = threadLocal.get();
        //如果浏览器中没有user-key，我们为其生成
        if (!userInfoTo.getTempUser()) {
            Cookie cookie = new Cookie(CartConstant.TEMP_USER_COOKIE_NAME, userInfoTo.getUserKey());
            cookie.setDomain("gulishop.cn");
            cookie.setMaxAge(CartConstant.TEMP_USER_COOKIE_TIMEOUT);
            response.addCookie(cookie);
        }
    }
}

2.购物车商品的增加

使用自定义线程池 + CompletableFuture.runAsync( ( )->{ } )来执行

@Override
    public CartItemVo addCartItem(Long skuId, Integer num) {
		// 判断是登录状态还是不是登录状态
        BoundHashOperations<String, Object, Object> ops = getCartItemOps();
        // 判断当前商品是否已经存在购物车
        String cartJson = (String) ops.get(skuId.toString());
        // 1 已经存在购物车，将数据取出并添加商品数量
        if (!StringUtils.isEmpty(cartJson)) {
            //1.1 将json转为对象并将count+
            CartItemVo cartItemVo = JSON.parseObject(cartJson, CartItemVo.class);
            cartItemVo.setCount(cartItemVo.getCount() + num);
            //1.2 将更新后的对象转为json并存入redis
            String jsonString = JSON.toJSONString(cartItemVo);
            ops.put(skuId.toString(), jsonString);
            return cartItemVo;
        } else {

			// 商品新增
            CartItemVo cartItemVo = new CartItemVo();
            // 2 未存在购物车，则添加新商品
            CompletableFuture<Void> future1 = CompletableFuture.runAsync(() -> {
                //2.1 远程查询sku基本信息
                R info = productFeignService.info(skuId);
                SkuInfoVo skuInfo = info.getData(new TypeReference<SkuInfoVo>() {
                });

                cartItemVo.setCheck(true);
                cartItemVo.setCount(num);
                cartItemVo.setImage(skuInfo.getSkuDefaultImg());
                cartItemVo.setPrice(skuInfo.getPrice());
                cartItemVo.setSkuId(skuId);
                cartItemVo.setTitle(skuInfo.getSkuTitle());
            }, executor);

            //2.2 远程查询sku属性组合信息
            CompletableFuture<Void> future2 = CompletableFuture.runAsync(() -> {
                List<String> attrValuesAsString = productFeignService.getSkuSaleAttrValuesAsString(skuId);
                cartItemVo.setSkuAttrValues(attrValuesAsString);
            }, executor);

            try {
                CompletableFuture.allOf(future1, future2).get();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (ExecutionException e) {
                e.printStackTrace();
            }
            //2.3 将该属性封装并存入redis,登录用户使用userId为key,否则使用user-key
            String toJSONString = JSON.toJSONString(cartItemVo);
            ops.put(skuId.toString(), toJSONString);
            return cartItemVo;
        }
    }

// 判断是登录状态还是不是登录状态
BoundHashOperations<String, Object, Object> ops = getCartItemOps();

private BoundHashOperations<String, Object, Object> getCartItemOps() {
        //1判断是否已经登录
        UserInfoTo userInfoTo = CartInterceptor.threadLocal.get();
        //1.1 登录使用userId操作redis
        if (!StringUtils.isEmpty(userInfoTo.getUserId())) {
            return redisTemplate.boundHashOps(CartConstant.CART_PREFIX + userInfoTo.getUserId());
        } else {
            //1.2 未登录使用user-key操作redis
            return redisTemplate.boundHashOps(CartConstant.CART_PREFIX + userInfoTo.getUserKey());
        }
    }

谷粒商城项目篇12_分布式高级篇_购物车功能、消息队列RabbitMQ_List_02

谷粒商城项目篇12_分布式高级篇_购物车功能、消息队列RabbitMQ_java_03

3.添加完成重定向避免刷新页面重复添加商品

若是将添加、跳转的逻辑都放在addCartItem一个controller的话，刷新页面会重复提交cartService.addCartItem(skuId, num);
因此需要添加完成重定向到 addCartItemSuccess 的controller 查询redis 在跳转到显示页面

/**
     * 添加商品到购物车
     * RedirectAttributes.addFlashAttribute():将数据放在session中，可以在页面中取出，但是只能取一次
     * RedirectAttributes.addAttribute():将数据放在url后面
     * @return
     */

    @RequestMapping("/addCartItem")
    public String addCartItem(@RequestParam("skuId") Long skuId, @RequestParam("num") Integer num, RedirectAttributes attributes) {
        cartService.addCartItem(skuId, num);
        attributes.addAttribute("skuId", skuId);
        return "redirect:http://cart.gulishop.cn/addCartItemSuccess";
    }

    @RequestMapping("/addCartItemSuccess")
    public String addCartItemSuccess(@RequestParam("skuId") Long skuId,Model model) {
        CartItemVo cartItemVo = cartService.getCartItem(skuId);
        model.addAttribute("cartItem", cartItemVo);
        return "success";
    }

4.购物车商品的查询

@Override
    public CartVo getCart() {
        CartVo cartVo = new CartVo();
        UserInfoTo userInfoTo = CartInterceptor.threadLocal.get();
        //1 用户未登录，直接通过user-key获取临时购物车
        List<CartItemVo> tempCart = getCartByKey(userInfoTo.getUserKey());
        if (StringUtils.isEmpty(userInfoTo.getUserId())) {
            List<CartItemVo> cartItemVos = tempCart;
            cartVo.setItems(cartItemVos);
        } else {
            //2 用户登录
            //2.1 查询userId对应的购物车
            List<CartItemVo> userCart = getCartByKey(userInfoTo.getUserId().toString());
            //2.2 查询user-key对应的临时购物车，并和用户购物车合并
            if (tempCart != null && tempCart.size() > 0) {
                //BoundHashOperations<String, Object, Object> ops = redisTemplate.boundHashOps(CartConstant.CART_PREFIX + userInfoTo.getUserId());
                for (CartItemVo cartItemVo : tempCart) {
                    userCart.add(cartItemVo);
                    //2.3 在redis中更新数据
                    addCartItem(cartItemVo.getSkuId(), cartItemVo.getCount());
                }
            }
            cartVo.setItems(userCart);
            //2.4 删除临时购物车数据
            redisTemplate.delete(CartConstant.CART_PREFIX + userInfoTo.getUserKey());
        }

        return cartVo;
    }

5.购物车商品的删除

删除临时购物车数据
redisTemplate.delete(CartConstant.CART_PREFIX + userInfoTo.getUserKey());
登录用户购物车商品删除

@Override
    public void deleteItem(Long skuId) {
        BoundHashOperations<String, Object, Object> ops = getCartItemOps();
        ops.delete(skuId.toString());
    }

二、消息队列RabbitMQ

1.场景理解

和前面学习的异步编排原理类似，通过使用中间件消息队列，完成快速响应客户并把待处理的消息放入队列供其他订阅者消费

快速响应：无序等待异步结果，让队列任务慢慢执行即可
应用解耦：如订单系统，调用库存系统的接口减库存，若库存系统接口变化升级，就要修改订单系统的接口，现在可以将订单消息放入消息队列，库存系统及其他用到订单的系统直接订阅消息，完成对应的逻辑操作
流量控制：如秒杀业务，前端即使能接受并发，后台的业务代码对于大量请求也会阻塞导致最后资源耗尽。现在可以使用消息队列存放大量全部请求，然后订阅者慢慢消费。

谷粒商城项目篇12_分布式高级篇_购物车功能、消息队列RabbitMQ_List_04

谷粒商城项目篇12_分布式高级篇_购物车功能、消息队列RabbitMQ_List_05

2.概述

大多应用中，可通过消息服务中间件来提升系统异步通信、扩展解耦能力
消息服务中两个重要概念：
消息代理（message broker）和目的地（destination）
当消息发送者发送消息以后，将由消息代理接管，消息代理保证消息传递到指定目的地。
消息队列主要有两种形式的目的地
队列（queue）：点对点消息通信（point-to-point）
主题（topic）：发布（publish）/订阅（subscribe）消息通信
点对点式：
• 消息发送者发送消息，消息代理将其放入一个队列中，消息接收者从队列中获
取消息内容，消息读取后被移出队列
• 消息只有唯一的发送者和接受者，但并不是说只能有一个接收者
发布订阅式：
• 发送者（发布者）发送消息到主题，多个接收者（订阅者）监听（订阅）这个
主题，那么就会在消息到达时同时收到消息
JMS（Java Message Service）JAVA消息服务：
• 基于JVM消息代理的规范。ActiveMQ、HornetMQ是JMS实现
AMQP（Advanced Message Queuing Protocol）
• 高级消息队列协议，也是一个消息代理的规范，兼容JMS
• RabbitMQ是AMQP的实现
Spring支持
• spring-jms提供了对JMS的支持
• spring-rabbit提供了对AMQP的支持
• 需要ConnectionFactory的实现来连接消息代理
• 提供JmsTemplate、RabbitTemplate来发送消息
• @JmsListener（JMS）、@RabbitListener（AMQP）注解在方法上监听消息代理发布的消息
• @EnableJms、@EnableRabbit开启支持
Spring Boot自动配置
• JmsAutoConfiguration
• RabbitAutoConfiguration
市面的MQ产品
• ActiveMQ、RabbitMQ、RocketMQ、Kafka

核心概念

Message
消息，消息是不具名的，它由消息头和消息体组成。消息体是不透明的，而消息头则由一系列的可选属性组成，
这些属性包括routing-key（路由键）、priority（相对于其他消息的优先权）、delivery-mode（指出该消息可
能需要持久性存储）等。
Publisher
消息的生产者，也是一个向交换器发布消息的客户端应用程序。
Exchange
交换器，用来接收生产者发送的消息并将这些消息路由给服务器中的队列。
Exchange有4种类型：direct(默认)，fanout, topic, 和headers，不同类型的Exchange转发消息的策略有所区别
Queue
消息队列，用来保存消息直到发送给消费者。它是消息的容器，也是消息的终点。一个消息可投入一个或多个队列。消息一直
在队列里面，等待消费者连接到这个队列将其取走。
Binding
绑定，用于消息队列和交换器之间的关联。一个绑定就是基于路由键将交换器和消息队列连接起来的路由规则，所以可以将交换器理解成一个由绑定构成的路由表。Exchange 和Queue的绑定可以是多对多的关系。
Connection
网络连接，比如一个TCP长连接。包含多个信道
Channel
信道，多路复用连接中的一条独立的双向数据流通道。信道是建立在真实的TCP连接内的虚拟连接，AMQP 命令都是通过信道发出去的，不管是发布消息、订阅队列还是接收消息，这些动作都是通过信道完成。因为对于操作系统来说建立和销毁 TCP 都是非常昂贵的开销，所以引入了信道的概念，以复用一条 TCP 连接。
Consumer
消息的消费者，表示一个从消息队列中取得消息的客户端应用程序。
Virtual Host
虚拟主机，表示一批交换器、消息队列和相关对象。虚拟主机是共享相同的身份认证和加
密环境的独立服务器域。每个 vhost 本质上就是一个 mini 版的 RabbitMQ 服务器，拥
有自己的队列、交换器、绑定和权限机制。vhost 是 AMQP 概念的基础，必须在连接时
指定，RabbitMQ 默认的 vhost 是 / 。
Broker
表示消息队列服务器实体

谷粒商城项目篇12_分布式高级篇_购物车功能、消息队列RabbitMQ_java_06

3.Docker安装RabbitMQ

注意阿里云服务器需要配置安全组策略

docker run -d --name rabbitmq -p 5671:5671 -p 5672:5672 -p 4369:4369 -p 25672:25672 -p 15671:15671 -p 15672:15672 rabbitmq:management

// 设置自启
docker update rabbitmq --restart=always

端口

4369, 25672 (Erlang发现&集群端口)
5672, 5671 (AMQP端口)
15672 (web管理后台端口)
61613, 61614 (STOMP协议端口)
1883, 8883 (MQTT协议端口)

参考：https://www.rabbitmq.com/networking.htm

谷粒商城项目篇12_分布式高级篇_购物车功能、消息队列RabbitMQ_spring_07

谷粒商城项目篇12_分布式高级篇_购物车功能、消息队列RabbitMQ_List_08

交换机类型

direct:根据指定的route-key到指定队列（每个队列对应一个route-key）
fanout:全部绑定的队列都会收到，不受route-key的影响
topic:按照指定的模式，也就是路由键支持模糊匹配格式(# ，*)，发布订阅场景

4.整合SpringBoot

引入 spring-boot-starter-amqp
application.yml配置
测试RabbitMQ

AmqpAdmin：管理组件
RabbitTemplate：消息发送处理组件
@RabbitListener 监听消息的方法可以有三种参数（不分数量，顺序）
• Object content, Message message, Channel channel

<!--  rabbitmq-->

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

spring:
  rabbitmq:
    host: xxx
    port: 5672
    virtual-host: /

1.基本使用

RabbitAutoConfiguration自动配置类，自动配置了

（1）RabbitTemplate

发送消息send()、convertAndSend()将发送的对象转化为字节流数据

@Bean
        @ConditionalOnSingleCandidate(ConnectionFactory.class)
        @ConditionalOnMissingBean({RabbitOperations.class})
        public RabbitTemplate rabbitTemplate(RabbitTemplateConfigurer configurer, ConnectionFactory connectionFactory) {
            RabbitTemplate template = new RabbitTemplate();
            configurer.configure(template, connectionFactory);
            return template;
        }

（2）AmqpAdmin

声明交换机declareExchange()
创建队列delcareQueue()

@Bean
		@ConditionalOnSingleCandidate(ConnectionFactory.class)
		@ConditionalOnProperty(prefix = "spring.rabbitmq", name = "dynamic", matchIfMissing = true)
		@ConditionalOnMissingBean
		public AmqpAdmin amqpAdmin(ConnectionFactory connectionFactory) {
			return new RabbitAdmin(connectionFactory);
		}

//
// Source code recreated from a .class file by IntelliJ IDEA
// (powered by Fernflower decompiler)
//

package org.springframework.amqp.core;

import java.util.Properties;
import org.springframework.lang.Nullable;

public interface AmqpAdmin {
    void declareExchange(Exchange var1);

    boolean deleteExchange(String var1);

    @Nullable
    Queue declareQueue();

    @Nullable
    String declareQueue(Queue var1);

    boolean deleteQueue(String var1);

    void deleteQueue(String var1, boolean var2, boolean var3);

    void purgeQueue(String var1, boolean var2);

    int purgeQueue(String var1);

    void declareBinding(Binding var1);

    void removeBinding(Binding var1);

    @Nullable
    Properties getQueueProperties(String var1);

    @Nullable
    QueueInformation getQueueInfo(String var1);

    default void initialize() {
    }
}

交换机类

//
// Source code recreated from a .class file by IntelliJ IDEA
// (powered by Fernflower decompiler)
//

package org.springframework.amqp.core;

import java.util.Map;

public interface Exchange extends Declarable {
    String getName();

    String getType();

    boolean isDurable();

    boolean isAutoDelete();

    Map<String, Object> getArguments();

    boolean isDelayed();

    boolean isInternal();
}

（3）RabbitMessagingTemplate

配置序列化方式

@Configuration(proxyBeanMethods = false)
@ConditionalOnClass(RabbitMessagingTemplate.class)
@ConditionalOnMissingBean(RabbitMessagingTemplate.class)
@Import(RabbitTemplateConfiguration.class)
protected static class MessagingTemplateConfiguration {

	@Bean
	@ConditionalOnSingleCandidate(RabbitTemplate.class)
	public RabbitMessagingTemplate rabbitMessagingTemplate(RabbitTemplate rabbitTemplate) {
		return new RabbitMessagingTemplate(rabbitTemplate);
	}

}

2.如果发送的消息是对象，会默认的序列化机制，可以设置json序列化

谷粒商城项目篇12_分布式高级篇_购物车功能、消息队列RabbitMQ_List_09

package henu.soft.xiaosi.order.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 MyRabbitConfig {


    /**
     * 配置rabbitmq的对象消息序列化机制为json
     * @return
     */
    
    @Bean
    public MessageConverter messageConverter(){
        return new Jackson2JsonMessageConverter();
    }
    
}

3.监听队列消息@RabbitListener、@RabbitHandler

@RabbitListener该注解可以写在类、方法上面，用于绑定队列
@RabbitHandler只能标注方法上,可以用于方法重载区分收到的不同信息
原声方式获取消息内容：使用 Message message (类似ResponseBody自动获取),然后在message.getBody()获取Object然后一系列转化。
快捷的方式：直接使用对应实体类来映射获取数据

@RabbitListener(queues = {"xxxQueue"})
    public void receiveMessage(Message message, SkuInfoTo skuInfoTo, Channel channel){
        // 1. 得到字节数组，需要自己转化
        byte[] body = message.getBody();
        MessageProperties messageProperties = message.getMessageProperties();


        // 2. 直接得到对应实体对象
        System.out.println(skuInfoTo);

        // 3. 得到当前传输数据的通道
        System.out.println(channel);


    }

4.场景举栗

举栗

如订单微服务为集群部署，都监听指定xxxQueue消息队列
像队列发送消息，哪台服务器上的订单服务能收到该数据
像队列发送消息，业务耗时是否影响从队列接受数据

结果

同一个消息只能有一个客户端收到
耗时的业务处理完成才会从队列取任务

5.消息确认机制

保证消息不丢失，可靠抵达，可以使用事务消息（事务消息导致性能下降），性能下降250倍，为此引入确认
机制
• publisher confirmCallback 确认模式，确认生产消息到服务器
• publisher returnCallback 未投递到 queue 退回模式（类似成功回调、失败回调），确认生产消息到消息队列
• consumer ack机制

谷粒商城项目篇12_分布式高级篇_购物车功能、消息队列RabbitMQ_java_10

1.定制RabbitTemplate实现（消息正确到达mq服务器）确认回调

package henu.soft.xiaosi.order.config;


import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import javax.annotation.PostConstruct;

@Configuration
public class MyRabbitConfig {


    /**
     * 配置rabbitmq的对象消息序列化机制为json
     * @return
     */

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


    @Autowired
    RabbitTemplate rabbitTemplate;

    /**
     * 设置确认回调（生产消息--》服务器)
     */

    @PostConstruct
    public void initRabbitTemplate(){
        // 设置确认回调


        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            /**
             *
             * @param correlationData 当前消息的唯一关联数据（消息的唯一id）
             * @param b 消息是否成功到达rabbitmq服务器
             * @param s 失败的原因
             */
            @Override
            public void confirm(CorrelationData correlationData, boolean b, String s) {

            }
        });
    }
}

2.定制RabbitTemplate实现（消息正确到达mq队列）抵达回调

spring:

  rabbitmq:
    host: xxx
    port: 5672
    virtual-host: /
    # publisher-confirms: true 抵达消息服务器回调
    # 抵达队列回调
    publisher-returns: true
    template:
      mandatory: true

package henu.soft.xiaosi.order.config;


import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.ReturnedMessage;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import javax.annotation.PostConstruct;

@Configuration
public class MyRabbitConfig {


    /**
     * 配置rabbitmq的对象消息序列化机制为json
     * @return
     */

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


    @Autowired
    RabbitTemplate rabbitTemplate;



    @PostConstruct
    public void initRabbitTemplate(){

        /**
         * 1. 设置确认回调（生产消息--》服务器)
         */
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            /**
             *
             * @param correlationData 当前消息的唯一关联数据（消息的唯一id）
             * @param b 消息是否成功到达rabbitmq服务器
             * @param s 失败的原因
             */
            @Override
            public void confirm(CorrelationData correlationData, boolean b, String s) {

            }
        });

        /**
         * 2. 设置抵达队列回调（生产消息---》队列)
         */
        rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {
            /**
             *
             * @param returnedMessage
             */
            @Override
            public void returnedMessage(ReturnedMessage returnedMessage) {

            }
        });



        rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {

            /**
             *  已过时
             * @param message 失败消息详细信息
             * @param i 回复的状态码
             * @param s 回复的文本内容
             * @param s1 目标交换机
             * @param s2 目标路由键
             */
            @Override
            public void returnedMessage(Message message, int i, String s, String s1, String s2) {

            }
        });

    }



}

2.（消息正确到达消费者）手动ACK

•消费者获取到消息，成功处理，可以回复Ack给Broker
•basic.ack用于肯定确认；broker将移除此消息
•basic.nack用于否定确认；可以指定broker是否丢弃此消息，可以批量
•basic.reject用于否定确认；同上，但不能批量
•默认自动ack，消息被消费者收到，就会从broker的queue中移除，消息未处理完宕机，全部消息都会丢失
•queue无消费者，消息依然会被存储，直到消费者消费
•消费者收到消息，默认会自动ack。但是如果无法确定此消息是否被处理完成，
或者成功处理。我们可以开启手动ack模式
•消息处理成功，ack()，接受下一个消息，此消息broker就会移除
•消息处理失败，nack()/reject()，重新发送给其他人进行处理，或者容错处理后ack
•消息一直没有调用ack/nack方法，broker认为此消息正在被处理，不会投递给别人，此时客户
端断开，消息不会被broker移除，会投递给别人

默认的方式是自动ack,这样可能服务器宕机导致消息丢失，因此需要开启手动ack机制，手动缺人之后，才算被消费

谷粒商城项目篇12_分布式高级篇_购物车功能、消息队列RabbitMQ_List_11

// 手动ack
        try {
            /**
             * 消费确认
             * 后面参数为是否批量接受
             */
            channel.basicAck(tag,false);

            /**
             * 消费拒绝
             * 后面参数为是否批量接受，拒绝消费的消息如何处理（true：发回服务器，false:丢弃)
             */
            channel.basicNack(tag,false,false);


            /**
             * 消费拒绝
             * 后面参数为是否批量接受
             */
            channel.basicReject(tag,false);


        } catch (IOException e) {
            e.printStackTrace();
        }

//@RabbitListener(queues = {"xxxQueue"})
    public void receiveMessage(Message message, SkuInfoTo skuInfoTo, Channel channel){
        // 1. 得到字节数组，需要自己转化
        byte[] body = message.getBody();
        MessageProperties messageProperties = message.getMessageProperties();


        // 2. 直接得到对应实体对象
        System.out.println(skuInfoTo);

        // 3. 得到当前传输数据的通道
        System.out.println(channel);
        // 通道内自增，按照消息传送的数量自增
        long tag = messageProperties.getDeliveryTag();

        // 手动ack
        try {
            /**
             * 消费确认
             * 后面参数为是否批量接受
             */
            channel.basicAck(tag,false);

            /**
             * 消费拒绝
             * 后面参数为是否批量接受，拒绝消费的消息如何处理（true：发回服务器，false:丢弃)
             */
            channel.basicNack(tag,false,false);


            /**
             * 消费拒绝
             * 后面参数为是否批量接受
             */
            channel.basicReject(tag,false);


        } catch (IOException e) {
            e.printStackTrace();
        }


    }

标签：12,return,String,队列,RabbitMQ,购物车,消息,import,public
From： https://blog.51cto.com/u_15790456/6569644