分布式解决方案与实战

发布时间:2023年12月18日

分布式多线程性能调优

使用多线程优化接口

在这里插入图片描述
在这里插入图片描述
在这里插入图片描述

	//下单业务
    public Object order( long userId){
        long start = System.currentTimeMillis();//方法的开始时间戳(ms)

        JSONObject orderInfo = remoteService.createOrder(userId);

        Callable<JSONObject> callable1 = new Callable<JSONObject>() {
            @Override
            public JSONObject call() throws Exception {
                JSONObject goodsInfo = remoteService.dealGoods(orderInfo);
                return goodsInfo;
            }
        };
        Callable<JSONObject> callable2 = new Callable<JSONObject>() {
            @Override
            public JSONObject call() throws Exception {
                JSONObject pointsInfo = remoteService.dealPoints(orderInfo);
                return pointsInfo;
            }
        };
        Callable<JSONObject> callable3 = new Callable<JSONObject>() {
            @Override
            public JSONObject call() throws Exception {

                JSONObject deliverInfo = remoteService.dealDeliver(orderInfo);
                return deliverInfo;
            }
        };
        LeeFutureTask<JSONObject> task1 = new LeeFutureTask(callable1);
        LeeFutureTask<JSONObject> task2 = new LeeFutureTask(callable2);
        LeeFutureTask<JSONObject> task3 = new LeeFutureTask(callable3);

        Thread thread1 =new Thread(task1);
        Thread thread2 =new Thread(task2);
        Thread thread3 =new Thread(task3);

        thread1.start();
        thread2.start();
        thread3.start();

        try {
            orderInfo.putAll(task1.get());
            orderInfo.putAll(task2.get());
            orderInfo.putAll(task3.get());
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        } catch (ExecutionException e) {
            throw new RuntimeException(e);
        }

        long end = System.currentTimeMillis();//方法的开始时间戳(ms)
        System.out.println(end-start);//打印这个方法的执行时间
        return orderInfo;
    }

后台批处理的优化

在这里插入图片描述

public JSONObject orderFastbatch (long userId) throws Exception{

        JSONObject orderInfo = remoteService.createOrderFast(userId);

        //JSONObject goodsInfo = remoteService.dealGoodsFast(orderInfo); //这里是单个的请求,想做成批量+异步的方式
        //orderInfo.putAll(goodsInfo);
        CompletableFuture<JSONObject> future = new CompletableFuture<>();
        Request request = new Request();
        request.future =future;
        request.object = orderInfo;
        queue.add(request);
        return future.get(); //这里类似于FutureTask  的get方法,去异步的方式拿结果
    }

    //定义一个JUC中的MQ
    LinkedBlockingQueue<Request> queue = new LinkedBlockingQueue();
    class Request{
        JSONObject object;
        CompletableFuture<JSONObject> future;
    }

    @PostConstruct
    public void DoBiz(){
        ScheduledExecutorService executorService = Executors.newScheduledThreadPool(1);
        executorService.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                int size =queue.size();
                if(size ==0) return;//没有数据在queue中,定时任务不需要执行什么

                if(size >1000) size=1000;//这里限制每次批量封装的最多是1000

                List<JSONObject> ListJSONRepuest = new ArrayList<>();
                List<Request>  ListRequest = new ArrayList<>();
                for( int i =0 ;i <size;i++){
                    Request request =queue.poll();//从MQ中拉取
                    ListRequest.add(request);
                    ListJSONRepuest.add(request.object);
                }
                //调用了多少次批量接口
                List<JSONObject> ListJSONReponse  = remoteService.dealGoodsFastBatch(ListJSONRepuest);
                System.out.println("调用批量接口,本地组装的数据:"+size+"条");
                for(JSONObject  JSONReponse:ListJSONReponse){//这里可以使用hashmap 的方式减少一轮遍历。
                    for(Request  request:ListRequest){
                        String  request_OrderId =request.object.get("orderId").toString();
                        String  response_OrderId =JSONReponse.get("orderId").toString();
                       if(request_OrderId.equals(response_OrderId)){
                           request.future.complete(JSONReponse);
                       }
                    }
                }
            }
        }, 3000, 50, TimeUnit.MILLISECONDS);

    }
//处理库存信息 (批量接口)  1000,
    public List<JSONObject> dealGoodsFastBatch( List<JSONObject> objectList) {
        List<JSONObject> list = objectList;
        Iterator it = list.iterator();
        while(it.hasNext()){
            JSONObject result =(JSONObject)it.next();
            result.put("dealGoods", "ok");
        }
        return  list;
    }

批处理与MySQL的综合性优化

如果做了批量处理的话。
1W个请求,高并发请求,其实里面针对的修改库存会集中在一些热点数据8000个在一个商品。
应用层基于批量的接口进行优化。
伪代码:
for循环遍历list,把所有相同的goods_id放到hashmap进行扣减计数即可。

分布式锁

单机锁:sync、lock
MySQL去实现分布式锁–for update (行锁)

Redis

分布式锁的第一种常见方案:Redis来实现分布式锁。Redis key-value键值对的数据库–内存。

Redis的分布式锁的实现逻辑:
1、加锁,setnx key value
1)为了避免死锁问题setnx完之后设置TTL失效时间
2)为了TTL的失效的时候业务还未完成导致的多个应用抢到锁的BUG,这里可以使用一个守护线程,来不断的去续锁(延长key的TTL)

2、解锁del key
无论是加锁,还是解锁,这里涉及到多个命令。要解决原子性问题:
1、复合命令实现加锁。set lock 9527 ex 10 nx
2、解锁的逻辑中:在del之前一定要判断:只有持有锁的应用或线程,才能去解锁成功,否则都是失败。value做文章。存一个唯一标识。

if (get ==应用的保存的值)del
else释放锁失败

使用Lua脚本

锁的可重入。A抢到了锁,没有释放锁之前,依然可以lock进入加锁逻辑的。

Zookeeper

在这里插入图片描述
1、连接ZK、创建分布式锁的根节点/lock
2、一个线程获取锁时,create ()在/lock西面创建1个临时顺序节点3、使用getChildren()获取当前所有字节点,并且对子节点进行排序
4、判断当前创建的节点是否是所有子节点中最小的节点,如果是,则获取锁->执行对应的业务逻辑->在结束的时候使用delete()方法删除该节点。
否则需要等待其他的节点的释放锁、
5、当一个线程需要释放锁的,删除该节点,其他的线程获取当前节点前的一个节点来获取锁。

分布式缓存

在这里插入图片描述

缓存的收益:
成本(包括学习的成本)、收益。

成本:读成本、写成本。
原始的读成本(没有做缓存之前)

缓存位置:介于请求方法,数据提供方之间。

key-value
计算key的时间,查询key的时间,转换值的时间。命中率§
所有的数据查询的时间:计算key的时间+查询key的时间+转换值的时间+(1-P)*原始的查询时间

所有的数据查询的时间<<<<<所有数据的原始查询时间。
额外的成本:缓存的一致性(双删之类的策略)。

适合的场景:耗时特别长的查询复用(查询出来的数据不止一次)读多写少

缓存的键值的设计:K-V

Key:要做转换(hash)
单向函数:给定输入,很容易、很快计算出结果。给你结果,很难计算出输入。
正向快速,逆向困难。输入敏感(key改动一点点,得到结果变化很大),避免冲突。
不考虑安全:考虑业务

前缀_业务关键信息_后缀。公司统一规范。
user_order_XXXX
user-order-XXXX
user+order+XXXX

value:序列化\反序列化。
成本。hash> > String(效率)

总结:无碰撞。高效生成。高效查询。
以上所有的:被中间件提高的API封装了(Redis、项目)
查询key的时间:物理位置(要不要走网络<内网、外网>、内存/磁盘)。

缓存的更新机制:
一、查的时候更新:
非常简单
1、客户端查询数据,缓存中没有,从提供方获取,写入缓存(有一个过期时间t)
2、在t的时间内,所有的查询,都有缓存来提供。
3、当缓存数据t过期了,回到了第一步。

使用场景:对于数据准确性和实时性要求不高的场景。商品的关注人数。热度。
问题:缓存于实际数据不一致性的情况

二、主动更新(改的更新)

更新缓存、删除缓存。先后顺序(DB,缓存)
1、先更新缓存。再更新数据库
2、先更新数据库,再更新缓存
3、先删除缓存,再更新数据库
4、先更新数据库,再删除缓存

Redis内存淘汰算法

当Redis 内存超出物理内存限制时,内存的数据会开始和磁盘产生频繁的交换(swap)。交换会让Redis的性能急剧下降,对于访问量比较频繁的Redis来说,这样龟速的存取效率基本上等于不可用。
maxmemory
在生产环境中我们是不允许Redis出现交换行为的,为了限制最大使用内存,Redis 提供了配置参数maxmemory来限制内存超出期望大小。
当实际内存超出 maxmemory时,Redis 提供了几种可选策略(maxmemory-policy)来让用户自己决定该如何腾出新的空间以继续提供读写服务。

Noeviction
noeviction不会继续服务写请求(DEL请求可以继续服务),读请求可以继续进行。这样可以保证不会丢失数据,但是会让线上的业务不能持续进行。这是默认的淘汰策略。

volatile-lru
volatile-Iru尝试淘汰设置了过期时间的key,最少使用的 key优先被淘汰。没有设置过期时间的 key 不会被淘汰,这样可以保证需要持久化的数据不会突然丢失。

volatile-ttl
volatile-ttl 跟上面一样,除了淘汰的策略不是LRU,而是key的剩余寿命ttl 的值,ttl越小越优先被淘汰。

volatile-random
volatile-random跟上面一样,不过淘汰的key是过期key集合中随机的 key.

allkeys-Iru
allkeys-lru区别于volatile-lru,这个策略要淘汰的key对象是全体的key集合,而不只是过期的 key集合。这意味着没有设置过期时间的key也会被淘汰。

allkeys-random
allkeys-random跟上面一样,不过淘汰的策略是随机的key。
volatile-xx策略只会针对带过期时间的key进行淘汰,allkeys-xxx策略会对所有的 key进行淘汰。如果你只是拿Redis做缓存,那应该使用allkeys-xx,客户端写缓存时不必携带过期时间。如果你还想同时使用Redis的持久化功能,那就使用volatile-xxx 策略,这样可以保留没有设置过期时间的 key,它们是永久的key 不会被LRU 算法淘汰。

LRU算法
实现 LRU算法除了需要key/value 字典外,还需要附加一个链表,链表中的元素按照一定的顺序进行排列。当空间满的时候,会踢掉链表尾部的元素。当字典的某个元素被访问时,它在链表中的位置会被移动到表头。所以链表的元素排列顺序就是元素最近被访问的时间顺序。
位于链表尾部的元素就是不被重用的元素,所以会被踢掉。位于表头的元素就是最近刚刚被人用过的元素,所以暂时不会被踢。
在这里插入图片描述

近似LRU算法
Redis使用的是一种近似LRU算法,它跟LRU算法还不太一样。之所以不使用LRU算法,是因为需要消耗大量的额外的内存,需要对现有的数据结构进行较大的改造。近似LRU算法则很简单,在现有数据结构的基础上使用随机采样法来淘汰元素,能达到和LRU算法非常近似的效果。Redis为实现近似LRU算法,它给每个key增加了一个额外的小字段,这个字段的长度是24个bit,也就是最后一次被访问的时间戳。

当Redis 执行写操作时,发现内存超出maxmemory,就会执行一次LRU淘汰算法。这个算法也很简单,就是随机采样出5(可以配置maxmemory-samples)个 key,然后淘汰掉最旧的 key,如果淘汰后内存还是超出maxmemory,那就继续随机采样淘汰,直到内存低于maxmemory为止。

LFU算法
LFU算法是Redis4.0里面新加的一种淘汰策略。它的全称是Least Frequently Used,它的核心思想是根据key的最近被访问的频率进行淘汰,很少被访问的优先被淘汰,被访问的多的则被留下来。

分布式ID

在这里插入图片描述
为什么需要分布式ID
1、传统的业务方式:直接插入数据库:MySQL 并发量600/s对比一下:互联网公司的ID现实、淘宝
如果是淘宝的订单号,生成要求!!!1、全局唯一
2、高并发(前几年的双十一:一天13亿单)算12个小时,每秒钟平均并发:3W
3、高可用
4、安全性
5、可读性(趋势递增)

分布式ID的方案,哪家强!!!

1、数据库自增ID(段号,批量处理)
2、UUID
3、Redis key 的incr
4、雪花算法

雪花算法原理

41位时间戳+10位机器ID+12位序号(自增),转换成长度为18的长整型
在这里插入图片描述
Java中SnowFlake算法生成ID,使用long来存储。64位第一位: bit不用。
41位的时间戳:2的41的,每个数,代表毫秒级别(范围):2的41次方/10006060*24=69年
10位的机器ID:2的10次方=1024台机器。(一般是5位数据中心,5位机器标识)
12位的序列号:4096个数

时钟回拨:
10点1分1秒1毫秒(200个)
实际的时间:10点1分1秒2毫秒(100个),通过机器来获取时间戳(10点1分1秒1毫秒)
修改时间戳为9点呢。。。

public class SnowFlake {
    private long workerId; //工作机器ID(0~31)
    private long datacenterId; //数据中心ID(0~31)
    private long sequence = 0L; //1毫秒内序号(0~4095)
    private long twepoch = 1288834974657L;
    private long workerIdBits = 5L; //机器id所占的位数
    private long datacenterIdBits = 5L;//数据中心所占的位数
    private long maxWorkerId = -1L ^ (-1L << workerIdBits); //支持的最大机器id,结果是31
    private long maxDatacenterId = -1L ^ (-1L << datacenterIdBits); //支持的最大数据中心id,结果是31
    private long sequenceBits = 12L; //序列在id中占的位数
    private long workerIdShift = sequenceBits; //机器ID向左移12位
    private long datacenterIdShift = sequenceBits + workerIdBits; //数据中心id向左移17位(12+5)
    private long timestampLeftShift =   + workerIdBits + datacenterIdBits;//时间戳向左移22位(5+5+12)
    private long sequenceMask = -1L ^ (-1L << sequenceBits); //自增长最大值4095,0开始
    private long lastTimestamp = -1L; //上次生成ID的时间截
    //构造函数(机器id 、数据中心ID )
    public SnowFlake(long workerId, long datacenterId) {
        if (workerId > maxWorkerId || workerId < 0) {
            throw new IllegalArgumentException(String.format("workerId can't be greater than %d or less than 0", maxWorkerId));
        }
        if (datacenterId > maxDatacenterId || datacenterId < 0) {
            throw new IllegalArgumentException(String.format("datacenterId can't be greater than %d or less than 0", maxDatacenterId));
        }
        this.workerId = workerId;
        this.datacenterId = datacenterId;
    }
    //获得下一个ID
    public synchronized long nextId() {
        long timestamp = timeGen();
        //如果当前时间戳小于上次ID的时间戳,说明系统出现了“时钟回拨”,抛出异常(并不完善)
        if (timestamp < lastTimestamp) {
            throw new RuntimeException(String.format("Clock moved backwards.  Refusing to generate id for %d milliseconds", lastTimestamp - timestamp));
        }
        //如果是同一时间戳,则需要进行毫秒内的序号排序
        if (lastTimestamp == timestamp) {
            sequence = (sequence + 1) & sequenceMask;
            if (sequence == 0) {//同一毫秒的序号数达到最大,只能等待下一毫秒
                timestamp = tilNextMillis(lastTimestamp);
            }
        } else {
            sequence = 0L;//时间戳改变(到了下一毫秒),毫秒内序列重置
        }
        lastTimestamp = timestamp;
        // 移位并通过或运算拼到一起组成64位的ID
        return ((timestamp - twepoch) << timestampLeftShift) |  // 时间戳左移22位
                (datacenterId << datacenterIdShift) | //数据标识左移17位
                (workerId << workerIdShift) | //机器id标识左移12位
                sequence;
    }
    //阻塞到下一个毫秒,直到获得新的时间戳
    protected long tilNextMillis(long lastTimestamp) {
        long timestamp = timeGen();
        while (timestamp <= lastTimestamp) { //乐观锁的机制确保是获取下一毫秒
            timestamp = timeGen();
        }
        return timestamp;
    }
    //返回以毫秒为单位的当前时间
    protected long timeGen() {
        return System.currentTimeMillis();
    }
}

Redis的分布式ID以及改良:

public class IDTest {

    static CountDownLatch ctl = new CountDownLatch(1000);

    static SnowFlake snowFlake = new SnowFlake(9,20);
    public static void main(String[] args) {

        for (int i =0 ; i <1000;i++){ //同时启动1000个线程
            new Thread(()->{
                try {
                    ctl.await();
                }catch (Exception e){
                    e.printStackTrace();
                }
                getRedis();
            }).start();
            ctl.countDown();
        }
    }

    public static void  getUUID(){
        UUID uuid= UUID.randomUUID();
        System.out.println("insert into order(order_id) values('"+uuid+"')");
    }

    public static void  getRedis(){
        Jedis jedis = new Jedis("127.0.0.1",6379);
        String key ="orderid";
        String prefix =getPrefix(new Date());
        long value= jedis.incr(key); //加入  淘宝 订单  用户id+日期->long+order_id ->Redis中的key
        String id = prefix+String.format("%1$05d",value);
        System.out.println("insert into order(order_id) values('"+id+"')");
    }
    //加入时间和日期的前缀
    private  static String getPrefix(Date date){
        Calendar c = Calendar.getInstance();
        c.setTime(date);
        int year =c.get(Calendar.YEAR);
        int day =c.get(Calendar.DAY_OF_YEAR); //一年中的第多少天
        int hour = c.get(Calendar.HOUR_OF_DAY);//一天当中的第几个小时
        String dayString =String.format("%1$03d",day); //转成3位字符串
        String hourString =String.format("%1$02d",hour);//转成2位字符串
        return (year-2000)+dayString+hourString;
    }

    public static void  getsonwflake(){
        long value= snowFlake.nextId(); //雪花算法
        System.out.println("insert into order(order_id) values('"+value+"')");
    }
}

分布式限流

为什么要限流!!!
互联网业务,场景突发性,秒杀,抢购。特点:突然性高并发。>>系统承受能力。做限制,限流

限流算法—固定算法

计数器:1个小时1000次。i++。优点:实现建立。
缺点:临界问题。

在这里插入图片描述

滑动窗口算法

为了解决固定窗口的临界问题。所以有了滑动窗口算法
滑动窗口也是有缺陷。为了让限流更加平滑,画出更多格子。

在这里插入图片描述

漏桶算法

在这里插入图片描述

令牌桶

在这里插入图片描述
具体sentinel限流查看springcloudalibaba部分

分布式数据存储技术

1、NoSQL
2、关系型数据库
3、分布式数据存储技术

分布式分流

CDN

什么是CDN?
内容分发网络

适合CDN的:静态资源包括图片,文档, 多媒体
动态内容(代码)不推荐使用CDN

要个根据不同的CDN缓存时间:头像1天,其他的30分钟。
http cache-control
public . private no-cache
max-age

流量的处理:封顶的配置
5分钟1G的CDN流量(可能有人在盗刷)—返回404

DNS

1、DNS与IP地址
2、域名结构树
3、域名服务器
4、域名解析过程

DNS的作用 域名解析

www.baidu.com ->IP启动QQ,里面通过IP去访问
浏览器–通过域名去访问(最终需要解析成IP) DNS
在这里插入图片描述
根域名

在这里插入图片描述
域名解析过程
在这里插入图片描述

Nginx负载均衡与算法

负载均衡:
定义:将负载变得均衡
负载(请求)、均衡(算法)

负载均衡的算法:
1、Round Robin:轮训(默认)
2、Round Robin+权重
3、Least connections:最少连接数
4、IP hash
确保同一个地址请求到同一服务器
5、hash
6、最短时间
7、随机

微服务中的负载均衡与算法

在这里插入图片描述

负载均衡的策略:
RandomRule:随机选择一个Server
RetryRule:重试机制。
RoundRobinRule:轮训,
AvailabilityFilteringRule:过滤掉那些状态异常的。
BestAvailableRule:选择一个最小的并发的Server
WeightedResponseTimeRule:根据响应时间加权重。
(响应时间越长,权重越小,被选中的可能性越低:能者多劳)
ZoneAvoidanceRule:默认的负载均衡器:类似于轮训。
(Server所在区域的性能+Server可用性Server)
NacosRule:同集群优先调用

分布式消息

1、分布式系统加入RocketMQ实现异步与解耦功能

异步

 		<!--RocketMQ-->
        <dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-spring-boot-starter</artifactId>
            <version>2.1.0</version>
        </dependency>

        <!--RocketMQ-->
        <dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-client</artifactId>
            <version>4.8.0</version>
        </dependency>
        <!--gson-->
        <dependency>
            <groupId>com.google.code.gson</groupId>
            <artifactId>gson</artifactId>
            <version>2.8.3</version>
        </dependency>
# RocketMQ
rocketmq.name-server=127.0.0.1:9876
rocketmq.producer.group=orderProducerGroup


mq.order.topic=orderTopic
mq.order.consumer.group.name=orderTopic_shop_order

mq.delay.topic=delayOrder
mq.delay.consumer.group.name=delayOrder_shop_order

mq.order.cancel.topic=cancelOrder

controller

@Autowired
    private OrderServiceImpl orderService;
    //  http://localhost:8080/submitOrder?userId=1&goodsId=13&goodsNumber=1&couponId=1
    //提交订单
    @RequestMapping("/submitOrder")
    public String submitOrder(@RequestParam("userId")long userId,@RequestParam("goodsId")long goodsId,@RequestParam("goodsNumber")int goodsNumber,@RequestParam("couponId")long couponId){
        long orderid ;
        try {
            //check() 略过
            //确认订单
            ShopOrder shopOrder = new ShopOrder();
            shopOrder.setUserId(userId);
            shopOrder.setGoodsId(goodsId);
            shopOrder.setGoodsNumber(goodsNumber);
            shopOrder.setCouponId(couponId);
            shopOrder.setOrderStatus(0);
            shopOrder.setPayStatus(1);
            shopOrder.setShippingStatus(0);
            shopOrder.setAddTime(new Date());
            orderid =orderService.submitOrder(shopOrder);//核心
        } catch (Exception e) {
            e.printStackTrace();
            return FAILUER;
        }
        if(orderid>0){
            return SUCCESS+":"+orderid;
        }else{
            logger.error("提交订单失败:[" + orderid + "]");
            return FAILUER;
        }
    }

service

	@Autowired
    private RocketMQTemplate rocketMQTemplate;



    private static final String SUCCESS = "success";
    private static final String FAILUER = "failure";
    @Autowired
    private RestTemplate restTemplate;

    @Value("${mq.order.topic}")
    private String topic;

    @Value("${mq.order.cancel.topic}")
    private String canceltopic;


	public long submitOrder(ShopOrder shopOrder) {
        long orderid=0;
        try {
            shopOrderMapper.insert(shopOrder);
            if( shopOrder!=null && shopOrder.getOrderId()!=null) {
                orderid = shopOrder.getOrderId();
            }
            if(orderid<=0){
                return orderid;
            }
            //这里是同步、耦合的方式
            //去调用商品系统(扣减库存)
            // restUpdateGoods(shopOrder);
            //去调用用户系统(处理优惠券)
            //restUseCoupon(shopOrder);

            //发送普通消息
             MQShopOrder(shopOrder);
        }catch (Exception e){
            logger.error("提交订单失败:[" + orderid + "]");
            e.printStackTrace();
        }
        return orderid;
    }

	//生成订单--把订单信息发送到MQ
    public int MQShopOrder(ShopOrder shopOrder) throws Exception {
        //TODO 使用Gson序列化
        Gson gson = new Gson();
        String txtMsg = gson.toJson(shopOrder);
        Message message = new Message(topic,"",shopOrder.getOrderId().toString(),txtMsg.getBytes());
        SendResult sendResult = rocketMQTemplate.getProducer().send(message);
        if(sendResult.getSendStatus() == SendStatus.SEND_OK){
            return  1;
        }else{
            logger.error("MQ发送消息失败:[" + shopOrder.getOrderId() + "]");
            return  -1;
        }
    }


	//生成限时订单
    public int MQDelayOrder(ShopOrder shopOrder) throws Exception {
        //TODO 使用Gson序列化
        Gson gson = new Gson();
        String txtMsg = gson.toJson(shopOrder);
        Message message = new Message("delayOrder","",shopOrder.getOrderId().toString(),txtMsg.getBytes());
        // delayTimeLevel:(1~18个等级)"1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h"
        message.setDelayTimeLevel(5);//1分钟不支付,就触发延时消息,就会把订单置为无效,还有回退。。。。
        SendResult sendResult = rocketMQTemplate.getProducer().send(message);
        if(sendResult.getSendStatus() == SendStatus.SEND_OK){
            return  1;
        }else{
            logger.error("MQ发送演示消息失败:[" + shopOrder.getOrderId() + "]");
            return  -1;
        }
    }

消费者部分

@Component
@RocketMQMessageListener(topic = "${mq.order.topic}",consumerGroup = "${mq.order.consumer.group.name}",messageModel = MessageModel.CLUSTERING)
public class GoodsListener implements RocketMQListener<MessageExt> {
    private static final Logger logger = LoggerFactory.getLogger(GoodsListener.class);

    @Autowired
    private GoodsServiceImpl goodsService;

    @Override
    public void onMessage(MessageExt messageExt) {
        try {
            //1.解析消息内容
            String body = new String(messageExt.getBody(),"UTF-8");
            //TODO 使用GSON反序列化
            Gson gson = new Gson();
            ShopOrder order = (ShopOrder)gson.fromJson(body, ShopOrder.class);
            long orderId =order.getOrderId();
            long goodsId =order.getGoodsId();
            Integer goodsNumber=order.getGoodsNumber();
            goodsService.updateGoods(orderId,goodsId,goodsNumber);
        } catch (UnsupportedEncodingException e) {
            logger.error("订阅消息:${mq.order.topic} 失败:[" + messageExt.getBody().toString() + "]");
            e.printStackTrace();
        }

    }
}

削峰填谷实战
改造前
在这里插入图片描述

@Autowired
    private DBService dbService;

    @RequestMapping("/buyTicket")
    public String buyTicket()throws Exception{
        //模拟出票……
        System.out.println("开始购票业务------");
        return dbService.useDb("select ticket ");
    }
/**
 * 类说明:手写的方式:模拟数据库服务
 */
@Component
public class DBService {
    @Autowired
    private DBPool dbPool;
    private static class SQLResult{
        private final String sql;
        private String result;

        public SQLResult(String sql) {
            this.sql = sql;
        }
        public String getSql() {
            return sql;
        }
        public String getResult() {
            return result;
        }
        public void setResult(String result) {
            this.result = result;
        }
    }

    // 模拟数据库服务,如果200ms内无法获取到,将会返回null
    public String useDb(String sql)throws Exception {
        // 从线程池中获取连接,如果1000ms内无法获取到,将会返回null
        Connection connection = dbPool.fetchConn(1000);
        if (connection != null) {
            try {
                connection.createStatement();
                connection.commit();
            } finally {
                dbPool.releaseConn(connection);
            }
        } else {
            throw new Exception("无法获取到数据库连接");
        }
        return sql+"被正确处理";
    }
}
/**
 * 类说明:数据库连接池手动实现,限定了连接池为20个大小,如果超过了则返回null
 */
@Component
public class DBPool {

	//数据库池的容器
	private static LinkedList<Connection> pool = new LinkedList<>();
	//限定了连接池为20个大小
	private final static int CONNECT_CONUT = 20;
	static{
		for(int i=0;i<CONNECT_CONUT;i++) {
			pool.addLast(SqlConnectImpl.fetchConnection());
		}
	}

	//在mills时间内还拿不到数据库连接,返回一个null,超时机制
	public Connection fetchConn(long mills) throws InterruptedException {
		synchronized (pool) {
			if (mills<0) {//没有时间限制
				while(pool.isEmpty()) {//如果连接池为空,则等待
					pool.wait();
				}
				return pool.removeFirst();
			}else {//有时间限制
				long overtime = System.currentTimeMillis()+mills;
				long remain = mills;
				while(pool.isEmpty()&&remain>0) {//如果连接池为空,则等待
					pool.wait(remain);
					remain = overtime - System.currentTimeMillis();
				}
				Connection result  = null;
				if(!pool.isEmpty()) {//如果连接池不为空,则取出连接
					result = pool.removeFirst();
				}
				return result;
			}
		}
	}

	//放回数据库连接,通知其他等待线程
	public void releaseConn(Connection conn) {
		if(conn!=null) {
			synchronized (pool) {
				pool.addLast(conn);
				pool.notifyAll();
			}
		}
	}
}

RocketMQ改造后
在这里插入图片描述

    @Autowired
    private GoodsServiceImpl goodsService;

	//改造后的:通过MQ进行异步、缓冲处理,达到削峰填谷的效果
    @RequestMapping("/rocketmq/buyTicket")
    public String buyTicketToRocket()throws Exception{
        if(goodsService.MQShopOrder("123")==1){
            return "success";
        }else {
            throw new Exception("购票失败");
        }
    }
//生成订单--把订单信息发送到MQ
    public int MQShopOrder(String ticket) throws Exception {
        Message message = new Message("ticket","",ticket,ticket.getBytes());
        SendResult sendResult = rocketMQTemplate.getProducer().send(message);
        if(sendResult.getSendStatus() == SendStatus.SEND_OK){
            return  1;
        }else{
            return  -1;
        }
    }

消费者

@Component
@RocketMQMessageListener(topic = "ticket",consumerGroup = "ticket-goods",messageModel = MessageModel.CLUSTERING)
public class TicketListener implements RocketMQListener<MessageExt> {
    private static final Logger logger = LoggerFactory.getLogger(TicketListener.class);

    @Autowired
    private DBService dbService;

    @Override
    public void onMessage(MessageExt messageExt) {
        try {
            //1.解析消息内容
            String body = new String(messageExt.getBody(),"UTF-8");
            //模拟出票……
            dbService.useDb("select ticket ");
            System.out.println("短信或其他的形式通知用户!");
        } catch (Exception e) {
            logger.error("订阅消息:${mq.order.topic} 失败:[" + messageExt.getBody().toString() + "]");
            e.printStackTrace();
        }

    }
}

2、利用RocketMQ的延时消息实现限时订单功能

既可以使用RocketMQ的延时消息也可以使用RabbitMQ利用死信队列来实现(消息写到队列A–》A中必须5分钟之内消费,死信消息->绑定一个死信消息交换器->队列B)
kafka无法实现延时订单的功能

		//发送延时消息
        MQDelayOrder(shopOrder);
	//生成限时订单
    public int MQDelayOrder(ShopOrder shopOrder) throws Exception {
        //TODO 使用Gson序列化
        Gson gson = new Gson();
        String txtMsg = gson.toJson(shopOrder);
        Message message = new Message("delayOrder","",shopOrder.getOrderId().toString(),txtMsg.getBytes());
        // delayTimeLevel:(1~18个等级)"1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h"
        message.setDelayTimeLevel(5);//1分钟不支付,就触发延时消息,就会把订单置为无效,还有回退。。。。
        SendResult sendResult = rocketMQTemplate.getProducer().send(message);
        if(sendResult.getSendStatus() == SendStatus.SEND_OK){
            return  1;
        }else{
            logger.error("MQ发送演示消息失败:[" + shopOrder.getOrderId() + "]");
            return  -1;
        }
    }

listener

mq.delay.topic=delayOrder
mq.delay.consumer.group.name=delayOrder_shop_order
@Component
@RocketMQMessageListener(topic = "${mq.delay.topic}",consumerGroup = "${mq.delay.consumer.group.name}",messageModel = MessageModel.CLUSTERING)
public class DelayOrderListener implements RocketMQListener<MessageExt> {
    private static final Logger logger = LoggerFactory.getLogger(DelayOrderListener.class);

    @Autowired
    private OrderServiceImpl orderService;
    @Override
    public void onMessage(MessageExt messageExt) {
        try {
            //1.解析消息内容
            String body = new String(messageExt.getBody(),"UTF-8");
            //TODO 使用GSON反序列化
            Gson gson = new Gson();
            ShopOrder order = (ShopOrder)gson.fromJson(body, ShopOrder.class);
            orderService.dealDealyOrder(order);
        } catch (Exception e) {
            logger.error("订阅消息:${mq.order.topic} 失败:[" + messageExt.getBody().toString() + "]");
            e.printStackTrace();
        }
    }
}
	//超时订单处理
    public int dealDealyOrder(ShopOrder shopOrder) throws Exception {
    //查看订单状态是否是已支付的
        //再从数据库中查一下订单的实时状态  调用第三方的接口  -》 订单是否真的支付
        ShopOrder shopOrderReal= shopOrderMapper.selectByPrimaryKey(shopOrder.getOrderId());
        if(shopOrderReal ==null) return -1;
        if(shopOrderReal.getPayStatus()==2 ||shopOrderReal.getOrderStatus()==2
                ||shopOrderReal.getOrderStatus()==3
                ||shopOrderReal.getOrderStatus()==4){
            //logger.info("该订单已经付款:[" + shopOrderReal.getOrderId() + "]");
            return 1;
        }
        shopOrderReal.setOrderStatus(3);//订单状态(订单超时没有支付,支付失败) --3无效
        if(shopOrderMapper.updateByPrimaryKeySelective(shopOrderReal)>0){
            //logger.info("该订单已经超时,改为无效:[" + shopOrderReal.getOrderId() + "]");
            return  1;
        }else{
            logger.error("修改订单超时失败:[" + shopOrderReal.getOrderId() + "]");
            return  -1;
        }
    }

扩展接口(暂时与该业务无关)

    
	//支付确认订单
    public int ConfirmOrder(long orderid) throws Exception {
        ShopOrder shopOrder=shopOrderMapper.selectByPrimaryKey(orderid);
        shopOrder.setOrderStatus(1);
        shopOrder.setPayStatus(2);
        if(  shopOrderMapper.updateByPrimaryKey(shopOrder)>0){
            return  1;
        }else{
            logger.error("确认订单失败:[" + orderid + "]");
            return  -1;
        }
    }

	

3、使用RocketMQ消息重复消费的幂等性处理方案

如何解决消息的重复问题–消费幂等性处理
两种方案:1、MVCC(版本号)、2、去重表的方案

MVCC方式
坏处:对业务侵入较大,一般不推荐
在这里插入图片描述
去重表方案代码

生成订单发送消息

public long submitOrder(ShopOrder shopOrder) {
        long orderid=0;
        try {
            shopOrderMapper.insert(shopOrder);
            if( shopOrder!=null && shopOrder.getOrderId()!=null) {
                orderid = shopOrder.getOrderId();
            }
            if(orderid<=0){
                return orderid;
            }

            //发送普通消息
             MQShopOrder(shopOrder);

            //发送普通消息  2次  造成消息的重复
            MQShopOrder(shopOrder);
            //发送延时消息
             MQDelayOrder(shopOrder);
        }catch (Exception e){
            logger.error("提交订单失败:[" + orderid + "]");
            e.printStackTrace();
        }
        return orderid;
    }

//生成订单--把订单信息发送到MQ
    public int MQShopOrder(ShopOrder shopOrder) throws Exception {
        //TODO 使用Gson序列化
        Gson gson = new Gson();
        String txtMsg = gson.toJson(shopOrder);
        Message message = new Message(topic,"",shopOrder.getOrderId().toString(),txtMsg.getBytes());
        SendResult sendResult = rocketMQTemplate.getProducer().send(message);
        if(sendResult.getSendStatus() == SendStatus.SEND_OK){
            return  1;
        }else{
            logger.error("MQ发送消息失败:[" + shopOrder.getOrderId() + "]");
            return  -1;
        }
    }

监听器类

@Component
@RocketMQMessageListener(topic = "${mq.order.topic}",consumerGroup = "${mq.order.consumer.group.name}",messageModel = MessageModel.CLUSTERING)
public class GoodsListener implements RocketMQListener<MessageExt> {
    private static final Logger logger = LoggerFactory.getLogger(GoodsListener.class);

    @Autowired
    private GoodsServiceImpl goodsService;

    @Override
    public void onMessage(MessageExt messageExt) {
        try {
            //1.解析消息内容
            String body = new String(messageExt.getBody(),"UTF-8");
            //TODO 使用GSON反序列化
            Gson gson = new Gson();
            ShopOrder order = (ShopOrder)gson.fromJson(body, ShopOrder.class);
            long orderId =order.getOrderId();
            long goodsId =order.getGoodsId();
            Integer goodsNumber=order.getGoodsNumber();
            goodsService.updateGoods(orderId,goodsId,goodsNumber);
        } catch (UnsupportedEncodingException e) {
            logger.error("订阅消息:${mq.order.topic} 失败:[" + messageExt.getBody().toString() + "]");
            e.printStackTrace();
        }

    }
}

具体去重代码

	@Autowired  //去重表
    private shopGoodsUniqueMapper shopGoodsUniqueMapper;

	//2、幂等性问题
    public synchronized   int updateGoods(long orderId, long goodsId, int goodsNumber){
        try{
        	//去重表
            shopGoodsUnique shopGoodsUnique = new shopGoodsUnique();
            shopGoodsUnique.setOrderId(orderId);
            shopGoodsUniqueMapper.insert(shopGoodsUnique);//如果这个地方消息重复了\异常
        }catch (Exception e){//如果异常了呢\
            logger.error("重复的修改库存信息:[" + orderId + "]");
            //完美的 又写一个topic 的数据(重复的),这个数据 可以用来分析
            return 1;
        }

		//业务操作
        //对共享一个东东 进行多线程操作呢?
        ShopGoods shopGoods =shopGoodsMapper.selectByPrimaryKey(goodsId);
        Integer goodnumber = shopGoods.getGoodsNumber()-goodsNumber;
        shopGoods.setGoodsNumber(goodnumber);
        if(shopGoodsMapper.updateByPrimaryKey(shopGoods)>=0){
            //logger.info("修改库存成功:[" + orderId + "]");
            return 1;
        }else{
            logger.error("修改库存失败:[" + orderId + "]");
            return -1;
        }
    }

分布式事务RocketMQ处理方案

在这里插入图片描述
整体方案
在这里插入图片描述

流程
在这里插入图片描述
具体代码

	@Autowired
    TransactionProducer producer;

 @RequestMapping("/trans-order")
    public String TransOrder(@RequestParam("userId")long userId, @RequestParam("goodsId")long goodsId, @RequestParam("goodsNumber")int goodsNumber, @RequestParam("couponId")long couponId) {
        long orderid;
        try {
            ShopOrder shopOrder = new ShopOrder();
            shopOrder.setUserId(userId);
            shopOrder.setGoodsId(goodsId);
            shopOrder.setGoodsNumber(goodsNumber);
            shopOrder.setCouponId(couponId);
            shopOrder.setOrderStatus(0);
            shopOrder.setPayStatus(1);
            shopOrder.setShippingStatus(0);
            shopOrder.setAddTime(new Date());

            //TODO 使用Gson序列化
            Gson gson = new Gson();
            String txtMsg = gson.toJson(shopOrder);
            //发送半事务消息
            SendResult  sendResult =producer.send(txtMsg,"trans-order");
            if(sendResult.getSendStatus() == SendStatus.SEND_OK){
                return  SUCCESS;
            }else{
                logger.error("MQ发送消息失败:[" + shopOrder.getOrderId() + "]");
                return  FAILUER;
            }
        } catch (Exception e) {
            e.printStackTrace();
            return FAILUER;
        }
    }
	@Component
public class TransactionProducer {
    
    private String producerGroup = "order_trans_group";

    // 事务消息
    private TransactionMQProducer producer;

    //用于执行本地事务和事务状态回查的监听器
    @Autowired
    OrderTransactionListener orderTransactionListener;
    //执行(定时回查)任务的线程池
    ThreadPoolExecutor executor = new ThreadPoolExecutor(5, 10, 60,
            TimeUnit.SECONDS, new ArrayBlockingQueue<>(50));
            
    @PostConstruct
    public void init(){
        producer = new TransactionMQProducer(producerGroup);
        producer.setNamesrvAddr("127.0.0.1:9876");
        producer.setSendMsgTimeout(Integer.MAX_VALUE);
        producer.setExecutorService(executor);
        producer.setTransactionListener(orderTransactionListener);
        this.start();
    }
    private void start(){
        try {
            this.producer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }
    //事务消息发送 
    public TransactionSendResult send(String data, String topic) throws MQClientException {
        Message message = new Message(topic,data.getBytes());
        return this.producer.sendMessageInTransaction(message, null);
    }
}

OrderTransactionListener类

@Component
public class OrderTransactionListener implements TransactionListener {

    @Autowired
    OrderServiceImpl orderService;

    @Autowired
    TransactionLogDao transactionLogDao;

    /**
     * 发送half msg 返回send ok后调用的方法
     * @param message
     * @param o
     * @return
     */
    @Override
    public LocalTransactionState executeLocalTransaction(Message message, Object o) {
        System.out.println("开始执行本地事务....");
        LocalTransactionState state;
        try{
            //1.解析消息内容
            String body = new String(message.getBody(),"UTF-8");
            //TODO 使用GSON反序列化
            Gson gson = new Gson();
            ShopOrder order = (ShopOrder)gson.fromJson(body, ShopOrder.class);
            //2、插入order表(订单表、事务表)
            orderService.createOrder(order,message.getTransactionId());
            // 同步的处理---返回commit后,消息能被消费者消费
            state = LocalTransactionState.COMMIT_MESSAGE;


           // 异步的处理---
           // state=LocalTransactionState.UNKNOW;
            System.out.println("本地事务已提交:"+message.getTransactionId());

        }catch (Exception e){
            System.out.println("执行本地事务失败:"+e);
            state = LocalTransactionState.ROLLBACK_MESSAGE;
        }
        return state;
    }

    /**
     * 定时回查的方法
     * @param messageExt
     * @return
     */
    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {

        // 回查多次失败 人工补偿。提醒人。发邮件的。
        System.out.println("开始回查本地事务状态:"+messageExt.getTransactionId());
        LocalTransactionState state;
        String transactionId = messageExt.getTransactionId();
        //这里如果发现事务表已经插入成功了,那么事务回查  commit
        if (transactionLogDao.selectCount(transactionId)>0){
            state = LocalTransactionState.COMMIT_MESSAGE;
        }else {
            //这里一般不会直接返回失败
            state = LocalTransactionState.UNKNOW;
        }
        System.out.println("结束本地事务状态查询:"+state);
        return state;
    }
}

orderService类

 //执行本地事务时调用,将订单数据和事务日志写入本地数据库
    @Transactional
    public long createOrder(ShopOrder shopOrder,String transactionId) {
        long orderid=0;
        try {
            //1.创建订单
            shopOrderMapper.insert(shopOrder);
            //2.写入事务日志
            TransactionLog log = new TransactionLog();
            log.setId(transactionId);
            log.setBusiness("order");
            log.setForeignKey(String.valueOf(shopOrder.getOrderId()));
            transactionLogMapper.insert(log);

        }catch (Exception e){
            logger.error("提交订单失败:[" + orderid + "]");
            e.printStackTrace();
        }
        return orderid;
    }

消费者部分代码

@Component
public class OrderListener implements MessageListenerConcurrently {

    @Autowired
    private GoodsServiceImpl goodsService;

    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext context) {
        try{
            //1.解析消息内容
            for (MessageExt message:list) {
                String body = new String(message.getBody(),"UTF-8");
                //TODO 使用GSON反序列化
                Gson gson = new Gson();
                if(1==1){
                    throw new Exception();
                }
                ShopOrder order = (ShopOrder)gson.fromJson(body, ShopOrder.class);
                long goodsId =order.getGoodsId();
                Integer goodsNumber=order.getGoodsNumber();
                goodsService.updateGoods(goodsId,goodsNumber);
                System.out.println("处理消费者数据:成功:"+message.getTransactionId());
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }catch (Exception e){
            System.out.println("处理消费者数据发生异常"+e);
            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
        }
    }
}
@Component
public class Consumer {

    String consumerGroup = "consumer-group";
    DefaultMQPushConsumer consumer;

    @Autowired
    OrderListener orderListener;
    
    @PostConstruct
    public void init() throws MQClientException {
        consumer = new DefaultMQPushConsumer(consumerGroup);
        consumer.setNamesrvAddr("127.0.0.1:9876");
        consumer.subscribe("trans-order","*");
        consumer.registerMessageListener(orderListener);
        // 2次重试就进死信队列
        consumer.setMaxReconsumeTimes(2);
        consumer.start();
    }
}

死信队列部分

@Component
public class OrderListenerDld implements MessageListenerConcurrently {


    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext context) {
        System.out.println("死信队列:消费者线程监听到消息。");
        try{
            for (MessageExt message:list) {
                System.out.println("死信队列的消息发邮件,要采取补偿策略");
                //手动补偿,邮件或钉钉告警
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }catch (Exception e){
            System.out.println("死信队列:处理消费者数据发生异常"+e);
            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
        }
    }
}
@Component
public class ConsumerDld {

    String consumerGroup = "consumer-group1";
    DefaultMQPushConsumer consumer;

    @Autowired
    OrderListenerDld orderListener;

    @PostConstruct
    public void init() throws MQClientException {
        consumer = new DefaultMQPushConsumer(consumerGroup);
        consumer.setNamesrvAddr("127.0.0.1:9876");
        consumer.subscribe("%DLQ%consumer-group","*");
        consumer.registerMessageListener(orderListener);

        consumer.setMaxReconsumeTimes(2);
        consumer.start();
    }
}

分布式文件

为什么需要分布式文件系统
存储 图片、视频、CSS样式、静态文件之类不经常发生变化的

比较老的:FastDFS ? . MinlO (go)
1、管理后端–推荐MinlO(可视化的管理后端)
2、功能上–推荐MinlO (MinlO更加的强大)
3、便捷—推荐MinlO(安装、部署一键式处理docker)
4、文档和社区支持—推荐MinlO

在这里插入图片描述

FastDFS

FastDFS是一个轻量级的开源分布式文件系统。2008年4月份开始启动。类似google FS的一个轻量级分布式文件系统,纯C实现,支持Linux、FreeBSD、AIX等UNIX系统。

FastDFS只能通过Client API访问,不支持POSIX访问方式。

?FastDFS特别适合大中型网站使用,用来存储资源文件(如:图片、文档、音频、视频等等),但是 FastDFS没有官网。

MinIO

MinIO是全球领先的对象存储先锋,目前在全世界有数百万的用户。

  • 高性能 ,在标准硬件上,读/写速度上高达183GB/秒和171GB/秒,拥有更高的吞吐量和更低的延迟
  • 可扩展性 ,为对象存储带来了简单的缩放模型,通过添加更多集群可以扩展空间
  • 简单 ,极简主义是MinIO的指导性设计原则,即可在几分钟内安装和配置
  • 与Amazon S3兼容 ,亚马逊云的 S3 API(接口协议)是在全球范围内达到共识的对象存储的协议,是全世界内大家都认可的标准
  • 数据安全 ,使用纠删码来保护数据免受硬件故障和无声数据损坏

纠删码

??纠删码是一种恢复丢失和损坏数据的数学算法, Minio默认采用 Reed-Solomon code将数据拆分成N/2个数据块和N/2个奇偶校验块。这就意味着如果是16块盘,一个对象会被分成8个数据块、8个奇偶校验块,你可以丢失任意8块盘(不管其是存放的数据块还是校验块),你仍可以从剩下的盘中的数据进行恢复。

Minio和FastDFS的对比

  1. 安装难度
  2. 文档
  3. 性能
  4. 容器化支持
  5. SDK支持

Minio安装

  • 1.安装DockerCompose
    ?安装DockerCompose的前提是先安装一个Docker环境,如果还没安装的参考波哥的博客地址:https://blog.csdn.net/qq_38526573/category_9619681.html

??Compose 是用于定义和运行多容器 Docker 应用程序的工具。通过 Compose,您可以使用 YML 文件来配置应用程序需要的所有服务。然后,使用一个命令,就可以从 YML 文件配置中创建并启动所有服务。

一键启动所有的服务

curl -L https://get.daocloud.io/docker/compose/releases/download/1.25.0/docker-compose-`uname -s`-`uname -m` > /usr/local/bin/docker-compose

修改文件夹权限

chmod +x /usr/local/bin/docker-compose

建立软连接

ln -s /usr/local/bin/docker-compose /usr/bin/docker-compose

检查是否安装成功

docker-compose --version

安装Minio集群
官方推荐 docker-compose.yaml

稍加修改,内容如下:

version: '3.7'

# 所有容器通用的设置和配置
x-minio-common: &minio-common
  image: minio/minio
  command: server --console-address ":9001" http://minio{1...4}/data
  expose:
    - "9000"
  # environment:
    # MINIO_ROOT_USER: minioadmin
    # MINIO_ROOT_PASSWORD: minioadmin
  healthcheck:
    test: ["CMD", "curl", "-f", "http://localhost:9000/minio/health/live"]
    interval: 30s
    timeout: 20s
    retries: 3

# 启动4个docker容器运行minio服务器实例
# 使用nginx反向代理9000端口,负载均衡, 你可以通过9001、9002、9003、9004端口访问它们的web console
services:
  minio1:
    <<: *minio-common
    hostname: minio1
    ports:
      - "9001:9001"
    volumes:
      - ./data/data1:/data

  minio2:
    <<: *minio-common
    hostname: minio2
    ports:
      - "9002:9001"
    volumes:
      - ./data/data2:/data

  minio3:
    <<: *minio-common
    hostname: minio3
    ports:
      - "9003:9001"
    volumes:
      - ./data/data3:/data

  minio4:
    <<: *minio-common
    hostname: minio4
    ports:
      - "9004:9001"
    volumes:
      - ./data/data4:/data

  nginx:
    image: nginx:1.19.2-alpine
    hostname: nginx
    volumes:
      - ./config/nginx.conf:/etc/nginx/nginx.conf:ro
    ports:
      - "9000:9000"
    depends_on:
      - minio1
      - minio2
      - minio3
      - minio4

接着新建文件夹 config,新建配置 nginx.conf

user  nginx;
worker_processes  auto;

error_log  /var/log/nginx/error.log warn;
pid        /var/run/nginx.pid;

events {
    worker_connections  4096;
}

http {
    include       /etc/nginx/mime.types;
    default_type  application/octet-stream;

    log_format  main  '$remote_addr - $remote_user [$time_local] "$request" '
                      '$status $body_bytes_sent "$http_referer" '
                      '"$http_user_agent" "$http_x_forwarded_for"';

    access_log  /var/log/nginx/access.log  main;
    sendfile        on;
    keepalive_timeout  65;

    # include /etc/nginx/conf.d/*.conf;

    upstream minio {
        server minio1:9000;
        server minio2:9000;
        server minio3:9000;
        server minio4:9000;
    }

    server {
        listen       9000;
        listen  [::]:9000;
        server_name  localhost;

        # To allow special characters in headers
        ignore_invalid_headers off;
        # Allow any size file to be uploaded.
        # Set to a value such as 1000m; to restrict file size to a specific value
        client_max_body_size 0;
        # To disable buffering
        proxy_buffering off;

        location / {
            proxy_set_header Host $http_host;
            proxy_set_header X-Real-IP $remote_addr;
            proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
            proxy_set_header X-Forwarded-Proto $scheme;

            proxy_connect_timeout 300;
            # Default is HTTP/1, keepalive is only enabled in HTTP/1.1
            proxy_http_version 1.1;
            proxy_set_header Connection "";
            chunked_transfer_encoding off;

            proxy_pass http://minio;
        }
    }

}

然后执行对应的命令

docker-compose up -d

访问控制台:http://192.168.56.100:9000
在这里插入图片描述
账号密码为:minioadmin

image.png

分布式搜索

1、通过数据库来检索(MySQL…)—数据库条件(B+树–排序,范围查询比较友好)
2、通过中间件搜索(ES) --全文检索

ES做分布式搜索

全文检索:没有搜索边界,讲究相关性,和搜索词相关的任何的结构,约定: 谐音、英文、拼音、网络热梗、同义词
在这里插入图片描述
在这里插入图片描述

电商经典场景----秒杀

整体流程

在这里插入图片描述

秒杀活动技术关注点

1、独立的服务部署(不影响其他的服务)
2、防止恶意攻击。(防止工作人员自己去秒杀、防止链接暴露)
3、库存的处理(1、预热。2、快速扣减)︰(不用去数据库校验库存:库存数据—>Redis、信号量控制秒杀的数量)
4、动静分离:只有动态请求才会进入后端服务。通过CDN服务,分担集群的压力。
5、恶意的请求的拦截:识别非法的请求,拦截。
6、流量错峰:采用各种手段、校验码、购物车,其他的等等,讲访问的流量,分摊到更长的时间线。
7、限流、熔断、降级的处理。
8、削峰填谷:秒杀的请求->消息中间件中,秒杀的订单服务从中间件获取,订单创建、库存其他的扣减。
9、限时订单的功能:(一个订单过了一段时间不支付,需要把这个库存还有其他的都要及时释放)

秒杀与Redis的综合运用

1)秒杀商品的定时上架。
后端管理界面:维护活动。商品1,商品2.。起止时间。库存(10)定时任务触发
2)秒杀的业务:请求-》访问Redis,在Redis中完成校验,完成信号量扣减。数据进入RocketMQ快速生成订单。

文章来源:https://blog.csdn.net/Forbidden_City/article/details/134997680
本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。