RabbitMQ创建队列及交换机、重试、ACK机制的使用

拼搏现实的明天。 2022-12-13 01:22 237阅读 0赞

#### 一、 前言： ####

> 之前使用rabbitMQ时只是简单的应用，停留在demo的概念上，本片博客是结合实际使用时会遇到的问题，在之前那篇文章[springboot整合RabbitMQ简单使用][springboot_RabbitMQ]基础上，继续进行讨论，有兴趣的可以看看之前那篇文章，里面有一些基本的概念介绍。

#### 二、摘要： ####

本篇文章将尝试回答这么几个问题：

*  **SpringBoot中创建RabbitMQ的队列、交换机、及绑定的几种方式?**
 *  **RabbitMQ中的ack机制是什么、如何使用?**
 *  **RabbitMQ中confirm-callback与return-callback的作用、如何使用？**
 *  **配合redis ,如何保证消息不丢失？**
 *  **RabbitMQ中重试机制，重试后转到其他队列**
 *  **SpringBoot中RabbitMQ的自动配置原理，如何自己扩展配置？**

> **其他问题：**  
> **a.当生产者发送消息时的队列并不存在时，springboot是否会自动创建？**
> 
> **b.当创建的队列或者交换机已经存在时，会如何处理？**

#### 三、创建队列、交换机、及绑定的几种方式 ####

> **方法一：直接进入RabbitMQ后台可视化处理**

如下图所示:  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzczMjk1NQ_size_16_color_FFFFFF_t_70_pic_center]  
一般这种需要运维配合，缺点：因为这都是线上如果操作不当删除其他业务的队列，后果很严重。所以运维一般不愿意采用这种方式，除非是已经发版后，需要紧急修改，不过自己写demo的时候，直接到可视化管理会比较方便。(上篇博客中有稍微具体一点的介绍)。

--------------------

> **方法二：容器启动时创建**

*  第一种采用AmqpAdmin管理员类进行操作 **(这个类其实就是上面可视化功能抽象出来的类、相关的操作都是这个类所具有都行为，所以用他能够完成一系列配置。)**

/** * @author liuzihao * @create 2020-10-09-11:07 */
    @Configuration
    public class RabbitMqAdminConfig { 
    
        public Queue testQueue() { 
            return new Queue("test-admin1");
        }
        public DirectExchange myExchange() { 
            return new DirectExchange("directexchange-admin1");
        }
        public Binding binding() { 
            return  BindingBuilder.bind(testQueue()).to(myExchange()).with("test-key-admin1");
        }
    
    	
        @Bean
        public AmqpAdmin amqpAdmin(ConnectionFactory connectionFactory) { 
        	/** * 如果不自己创建amqpAdmin，自动配置类中会创建 */
            AmqpAdmin amqpAdmin =  new RabbitAdmin(connectionFactory);
            amqpAdmin.declareQueue(testQueue());
            /* * Exchange是一个接口实现类如下： * 1.DirectExchange * 2.FanoutExchange * 3.TopicExchange * */
            amqpAdmin.declareExchange(myExchange());
            /** * Binding可以设置的参数： * destination 目的地 * destinationType 绑定的类型 是跟队列绑定 还是交换机绑定 * exchange 交换机的name * routingKey 路由的key * arguments; 参数 可以不设置 */
            amqpAdmin.declareBinding(binding());
            return amqpAdmin;
        }
    }

**上述这种方法在声明Queue和Exchange时网上很多的方法都会在上面使用`@Bean`注解放到容器中、其实使用AmqpAdmin这个类时，可以不用放到容器中，直接用admin对象配置基本组件，当然注入也不会出错，接下来介绍的这种方法就是直接放到容器中即可**

*  第二种是直接配置相关对象，注入到容器中 **(简单粗暴)** 容器在启动过程中会自己帮我我们创建

/** * @author liuzihao * @create 2020-10-08-22:13 * */
    @Configuration
    public class RabbitMqConfig { 
        @Bean
        public Queue testQueue() { 
            return new Queue("test3");
        }
        @Bean
        public DirectExchange myExchange() { 
            return new DirectExchange("directexchange3");
        }
        @Bean
        public  Binding binding() { 
            return  BindingBuilder.bind(testQueue()).to(myExchange()).with("test-key3");
        }
    }

--------------------

> **方法三：消费者使用注解@RabbitListener进行创建**

因为多播模式下，当某个消费者需要生产者的消息时，按照业务逻辑必然要自己创建队列，并且绑定交换机，这样当有生产者产生消息时，消息就会路由到自己所创建多队列中。（**多播模式下由消费者自己维护队列**）

@Component
    public class QueueListener{ 
        //进行监听
        @RabbitListener(bindings = { 
                @QueueBinding(
                        value = @Queue(value = "lister-test", durable = "true"),
                        exchange = @Exchange(value = "lister-exchange", type = ExchangeTypes.DIRECT, durable = "true"),
                        key = "key-listener"
                )}
        )
        public void receive(Map map){ 
            System.out.println("监听接收到消息"+map);
        }
    }

#### 四、RabbitMQ中的ack机制 ####

> ack 机制是一种确认机制，是消费者从队列中消费消息后的一种应答机制。

*  默认：采用的方式是自动ack，也就是说，消费者从队列中拿到消息进行业务处理后，队列中的消息就被删除了，如果此刻消费者处理该消息的时候发生异常，导致业务没有正常处理，那么这个该处理的消息也就丢失了。
 *  手动ack：当正确消费消息后，调用channel.basicAck()，进行手动确认；如果在处理消息过程中发送异常，可以调用channel.basicNack，表示拒绝消息，并且通过参数设置，可以让没有消费的消息再次回到队列，下次消费（**但是不断的重入队列，虽然消息不会丢失、却会造成消息队列阻塞、可以采用重试机制，或者转移到另外的队列如：死信队列中**）。 这样的话，控制了消息处理过程中的正确性。

消息的确认，其实对于生产者、与消费者可以理解都有他们的`ack确认机制`

*  生产者： 保证消息到达交换机 、消息再从交换机到队列
 *  消费者：消息正确从队列中被消费

**那么RabbitMQ都提供了哪些机制来解决这些问题呢?**

*  生产者

##### 4.1 confirm-callback 回调 #####

`confirm-callback`回调是检验消息是否达到交换机的一个机制。（具体config配置见最后）

// 如果消息没有到exchange, 则confirm回调, ack=false; 如果消息到达exchange, 则confirm回调, ack=true
            rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> { 
    
                if (ack) { 
                    log.info("消息确认到达exchange!");
                    RedisUtils.del(correlationData.getId());
    
                } else { 
                    log.error("消息推送MQ Exchange失败, 请检查MQ是否异常, 消息ID: {}", correlationData.getId());
                }
            });

效果：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzczMjk1NQ_size_16_color_FFFFFF_t_70_pic_center 1]

##### 4.2 Return-Callback 回调 #####

`Return-Callback`是校验消息是否到达队列的一个机制。（具体config配置见最后）

// 如果exchange到queue成功, 则不回调return; 如果exchange到queue失败, 则回调return(需设置mandatory=true, 否则不会回调, 消息就丢了)
            rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> { 
    
                log.error("消息推送MQ Queue失败, 请检查路由键准确, 交换机:{}, 路由键:{}", exchange, routingKey);
    
               // 这里其实可以采用重试机制
               // 1、休眠3秒
               // 2、重试 RabbitMQUtils.sendMessageToMQ(exchange, routingKey, messageObj);
    
            });

*  消费者

##### 4.3 消费者ack 与Nack #####

生产者保证消息无误的到达队列后，那么它的使命已经完成、消费者要做的就是如果正确的消费它。在上文中在介绍ack的时候已经提到，如果自动ack可能会导致消息丢失。那么可以采用这几种方案

*  拒绝消息，让其重回队列，等待下一个消费者消费（可能会阻塞队列）
 *  抛出异常 启用重试机制、到达重试阈值后，消息转到其他队列（利用RepublishMessageRecoverer类）（不能开启重入，会阻塞）
 *  拒绝无法处理的消息转移到死信队列中特殊处理（需要配置死信队列）
 *  拒绝消息、不重入队列，记录日志后续处理（与重试机制类似，少了重试）

--------------------

*  `拒绝消息`**basicNack(long deliveryTag, boolean multiple, boolean requeue)**

/** * @param long deliveryTag 消息的标志 * @param boolean multiple 是否批量处理 * @param boolean requeue 是否重入队列 */
                 basicNack(long deliveryTag, boolean multiple, boolean requeue)

*  `接受消息` **basicAck(long deliveryTag, boolean multiple)**

/** * @param long deliveryTag 消息的标志 * @param boolean multiple 是否批量处理 */
                 basicAck(long deliveryTag, boolean multiple)

例子：

@Component
    public class QueueListener{ 
        @RabbitListener(queues = { "my.amqpAdmin.queue"})
        /** * @Payload Employee employee 表示消息的载体 (也可不用注解标注) * @Header(AmqpHeaders.CHANNEL) 取Channel信息 (也可不用注解标注) * @Header(AmqpHeaders.DELIVERY_TAG) Long tag 获取消息的标志id * (tag 也可以在方法产生通过注入 Message 然后通过message.getMessageProperties().getDeliveryTag()进行获取) */
        public void getMessage(@Payload Employee employee,
                               @Header(AmqpHeaders.CHANNEL) Channel channel1,
                               @Header(AmqpHeaders.DELIVERY_TAG) Long tag) throws Exception{ 
            /** * 假设当消息Employee age为5的时候拒绝消息 */
            if (employee.getAge() == 5) { 
                /** * @param long deliveryTag 消息的标志 * @param boolean multiple 是否批量处理 * @param boolean requeue 是否重入队列 */
                System.out.println("消费者，拒绝消息" + employee);
                channel1.basicNack(tag, false, false);
                /** * 要想重试机制生效，进行相关配置后、还需要抛出异常，并且requeue = fasle(不然会一直阻塞) * throw new RuntimeException("111"); */
               
            }else { 
                /** * @param long deliveryTag 消息的标志 * @param boolean multiple 是否批量处理 */
                channel1.basicAck(tag, false);
                System.out.println("消费者，成功处理应答==>"+employee);
    
            }
        }
    }

##### 4.4 确认机制图解 #####

画图不容易啊  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzczMjk1NQ_size_16_color_FFFFFF_t_70_pic_center 2]

##### 4.5 消息不丢失方案讨论 #####

虽然生产者一方存在，消息到达回调机制，但是仍有可能消息还没到达mq时，消息就丢失了，所以我们可以采用

*  **redis缓存 + 定时任务补偿的方案** 如下图：（画图不易，可以点个赞）  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzczMjk1NQ_size_16_color_FFFFFF_t_70_pic_center 3]

> 一、消息如果没有到达exchange、定时任务会根据redis里的缓存进行补偿  
> 二、消息如果没有到队列，会有重发机制

采用redis缓存，再加上回调重试机制，可以说保证了消息不丢失送达队列，

> **上述方案，会有消息重发的风险，所以需要消费者方根据业务键，做好幂等处理**

#### **五、MQ重试机制** ####

> **- mq的重试是针对消费者的，并不是说消息重新进入队列,然后被消费。二者没有因果关系、没有因果关系、没有因果关系.**  
> **是消费者抛出异常后的一种重试机制，想要触发异常需要把异常抛出来**

*  配置（配置类代码见最后）：

# 开启重试
    spring.rabbitmq.listener.simple.retry.enabled=true
    # 重试次数,默认为3次
    spring.rabbitmq.listener.simple.retry.max-attempts=5
    # 也可以设置重试时间间隔

但是发送重试后，消息还是没有处理成功的话，就会被丢弃，继续处理其他消息。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzczMjk1NQ_size_16_color_FFFFFF_t_70_pic_center 4]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzczMjk1NQ_size_16_color_FFFFFF_t_70_pic_center 5]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzczMjk1NQ_size_16_color_FFFFFF_t_70_pic_center 6]  
查看一下消息队列里的消息：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzczMjk1NQ_size_16_color_FFFFFF_t_70_pic_center 7]  
已经被全部check了，那条拒绝的消息也就丢失了。为了解决这个问题，拒绝的消息是不是可以转移到另外的一个队列中呢？大家想到的可能就是死信队列了，死信队列是一种方式，其实我也可以让他转移到我买自定义的业务队列，比如：自定义一个回收队列 rec-queue，然后再针对该队列中无法处理的消息做特殊处理。

*  方式1:死信队列（如果配置了死信队列，并且nack :requeue = false,会自动发送到死信）
 *  方式2:自定义回收器、转发到自定义到业务队列中。下面介绍方式2

建立一个回收队列和回收交换机

/** * 创建回收队列 * @return */
        @Bean
        public DirectExchange errorExchange(){ 
            return new DirectExchange("rec-exchange",true,false);
        }
    
        @Bean
        public Queue errorQueue(){ 
            return new Queue("rec-queue", true);
        }
    
        @Bean
        public Binding errorBinding(Queue errorQueue, DirectExchange errorExchange){ 
            return BindingBuilder.bind(errorQueue).to(errorExchange).with("routing-key");
        }

建立完回收队列，那么如果让重试后到消息回收到改队列呢？那么下面了解一个接口：**MessageRecoverer**，名字就是消息回收器，主要是当所有尝试都失败时回回调。

public interface MessageRecoverer { 
    
    	// 回调被消耗但所有重试尝试都失败的消息。
    	void recover(Message message, Throwable cause);
    
    }

他有两个实现类：`RepublishMessageRecoverer` 和`RejectAndDontRequeueRecoverer(默认使用该回收器)`,只要创建RepublishMessageRecoverer类到容器中即可。

/** * 消息回收器 * @param rabbitTemplate * @return */
        @Bean
        public MessageRecoverer messageRecoverer(RabbitTemplate rabbitTemplate){ 
            return new RepublishMessageRecoverer(rabbitTemplate,"rec-exchange","routing-key");
        }

**这样在重试次数，超过阈值时，将会转移到自定义的 回收队列当中**如下图：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzczMjk1NQ_size_16_color_FFFFFF_t_70_pic_center 8]  
再看一下我们定义的`回收队列中`也就有了重试之后无法处理的消息，我们可以特殊处理。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzczMjk1NQ_size_16_color_FFFFFF_t_70_pic_center 9]

#### 六、配置原理 ####

*  自动配置类：RabbitAutoConfiguration 见博客 [springboot整合RabbitMQ简单使用][springboot_RabbitMQ]
 *  注解@RabbitListener ，配置起作用原于，后置处理器`RabbitListenerAnnotationBeanPostProcessor`，发挥作用是，对象创建完成，初始化前后。此时解析注解里的内容进行创建交换机，并绑定。部分代码如下：  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzczMjk1NQ_size_16_color_FFFFFF_t_70_pic_center 10]

#### 七、其他回答 ####

> **其他问题：**  
> **a.当生产者发送消息时的队列并不存在时，springboot是否会自动创建？**  
> 答：不会，会抛出异常
> 
> **b.当创建的队列或者交换机已经存在时，会如何处理？**  
> 答：保持原来的队列，或交换机，不会覆盖。（具体源码还没找到，如果有了解的，感谢留言告知）

#### 八、配置代码 ####

（因为我生产者和消费者用了两个服务，所以会有两个配置类）

*  生产者配置类

/** * @author liuzihao * @create 2020-10-08-22:13 * */
    @Configuration
    @Slf4j
    public class RabbitMqConfig { 
        @Bean
        public Queue testQueue() { 
            return new Queue("my.amqpAdmin.queue");
        }
        @Bean
        public DirectExchange myExchange() { 
            return new DirectExchange("my.direct");
        }
        @Bean
        public  Binding binding() { 
            return  BindingBuilder.bind(testQueue()).to(myExchange()).with("my.key");
        }
        
        
        @Bean
        public MessageConverter messageConverter(){ 
            return  new Jackson2JsonMessageConverter();
        }
        /** * MQ发送方配置 * * @param connectionFactory * @return */
        @Bean
        @Primary
        public RabbitTemplate rabbitTemplate(CachingConnectionFactory connectionFactory) { 
    
            // 若使用confirm-callback, 必须设置publisherConfirms为true
            connectionFactory.setPublisherConfirms(true);
            // 若使用return-callback, 必须设置publisherReturns为true
            connectionFactory.setPublisherReturns(true);
    
            RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
            // 使用return-callback, 必须设置mandatory为true
            rabbitTemplate.setMandatory(true);
            rabbitTemplate.setMessageConverter(messageConverter());
    
            // 如果消息没有到exchange, 则confirm回调, ack=false; 如果消息到达exchange, 则confirm回调, ack=true
            rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> { 
    
                if (ack) { 
                    log.info("消息确认到达exchange!");
                    /** * * 删除缓存在redis中的消息(保证消息不丢失，可以在发送消息的时候缓存在redis中) * RedisUtils.del(correlationData.getId()); */
    
                } else { 
                    log.error("消息推送MQ Exchange失败, 请检查MQ是否异常, 消息ID: {}", correlationData.getId());
                }
            });
    
            // 如果exchange到queue成功, 则不回调return; 如果exchange到queue失败, 则回调return(需设置mandatory=true, 否则不会回调, 消息就丢了)
            rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> { 
    
                log.error("消息推送MQ Queue失败, 请检查路由键准确, 交换机:{}, 路由键:{}", exchange, routingKey);
    
                /** * 重试机制 * 1、消息反序列化 * Object messageObj = new Jackson2JsonMessageConverter().fromMessage(message); * 2、休眠 * 3、重新发送到mq * RabbitMQUtils.sendMessageToMQ(exchange, routingKey, messageObj); */
    
    
            });
    
            return rabbitTemplate;
        }
    
    }

*  消费者配置类

@Configuration
    public class MyConfig { 
        @Bean
        public MessageConverter messageConverter(){ 
            return  new Jackson2JsonMessageConverter();
        }
        @Bean
        public SimpleRabbitListenerContainerFactory simpleRabbitListenerContainerFactory(SimpleRabbitListenerContainerFactoryConfigurer configurer, ConnectionFactory connectionFactory) { 
            SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
            configurer.configure(factory, connectionFactory);
            // 手动应答
            factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
            // 重试需要设置为false
            factory.setDefaultRequeueRejected(false);
            return factory;
        }
    
    
        /** * 创建回收队列 * @return */
        @Bean
        public DirectExchange errorExchange(){ 
            return new DirectExchange("rec-exchange",true,false);
        }
    
        @Bean
        public Queue errorQueue(){ 
            return new Queue("rec-queue", true);
        }
    
        @Bean
        public Binding errorBinding(Queue errorQueue, DirectExchange errorExchange){ 
            return BindingBuilder.bind(errorQueue).to(errorExchange).with("routing-key");
        }
    
        /** * 消息回收器 * @param rabbitTemplate * @return */
        @Bean
        public MessageRecoverer messageRecoverer(RabbitTemplate rabbitTemplate){ 
            return new RepublishMessageRecoverer(rabbitTemplate,"rec-exchange","routing-key");
        }
    }

*  github：[代码地址:注意是在master分支，默认是main分支][master_main]

[springboot_RabbitMQ]: https://blog.csdn.net/weixin_43732955/article/details/91357455
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzczMjk1NQ_size_16_color_FFFFFF_t_70_pic_center]: /images/20221123/c67202530232472298ea77cdfd73dbe4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzczMjk1NQ_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221123/2a55ed7d6f4746e588a00de6719754b9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzczMjk1NQ_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20221123/a912fd1240e041499c5845c086d1e46b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzczMjk1NQ_size_16_color_FFFFFF_t_70_pic_center 3]: /images/20221123/c90317f85d384a95b1b261e51dbc76db.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzczMjk1NQ_size_16_color_FFFFFF_t_70_pic_center 4]: /images/20221123/570f362f5dc949e4aba4bf61025efd80.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzczMjk1NQ_size_16_color_FFFFFF_t_70_pic_center 5]: /images/20221123/468d38b5740f48bc910b48289340f2d1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzczMjk1NQ_size_16_color_FFFFFF_t_70_pic_center 6]: /images/20221123/99568f8adfb24790996a402bf3561bab.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzczMjk1NQ_size_16_color_FFFFFF_t_70_pic_center 7]: /images/20221123/10f402b0daf646cdb19d283a1c0acef4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzczMjk1NQ_size_16_color_FFFFFF_t_70_pic_center 8]: /images/20221123/d426e07e43974c8982a9cc931d066b0f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzczMjk1NQ_size_16_color_FFFFFF_t_70_pic_center 9]: /images/20221123/599e5fb7ca1b412ea2abf8968111691c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzczMjk1NQ_size_16_color_FFFFFF_t_70_pic_center 10]: /images/20221123/963cee064bfd4e4a817670e3f8374277.png
[master_main]: https://github.com/Liuzihao169/springboot.git