发表于

这是我参与11月更文挑战的第10天，活动详情查看：2021最后一次更文挑战

基础介绍

Resilience4j是一款轻量级，易于使用的容错库，其灵感来自于Netflix Hystrix，但是专为Java 8和函数式编程而设计。轻量级，因为库只使用了Vavr，它没有任何其他外部依赖下。相比之下，Netflix Hystrix对Archaius具有编译依赖性，Archaius具有更多的外部库依赖性，例如Guava和Apache Commons Configuration。

使用Resilience4j

要使用Resilience4j，不需要引入所有依赖，只需要选择你需要的。Resilience4j提供了以下的核心模块和拓展模块:

核心模块

  • resilience4j-circuitbreaker: Circuit breaking
  • resilience4j-ratelimiter: Rate limiting
  • resilience4j-bulkhead: Bulkheading
  • resilience4j-retry: Automatic retrying (sync and async)
  • resilience4j-cache: Result caching
  • resilience4j-timelimiter: Timeout handling

Circuitbreaker

CircuitBreaker通过具有三种正常状态的有限状态机实现：CLOSED，OPEN和HALF_OPEN以及两个特殊状态DISABLED和FORCED_OPEN。

  • 当熔断器关闭时，所有的请求都会通过熔断器。
  • 如果失败率超过设定的阈值，熔断器就会从关闭状态转换到打开状态，这时所有的请求都会被拒绝。
  • 当经过一段时间后，熔断器会从打开状态转换到半开状态，这时仅有一定数量的请求会被放入，并重新计算失败率，如果失败率超过阈值，则变为打开状态，如果失败率低于阈值，则变为关闭状态。

Ring Bit Buffer(环形缓冲区)

Resilience4j记录请求状态的数据结构和Hystrix不同，Hystrix是使用滑动窗口来进行存储的，而Resilience4j采用的是Ring Bit Buffer(环形缓冲区)。

Ring Bit Buffer在内部使用BitSet这样的数据结构来进行存储，BitSet的结构如下图所示：

每一次请求的成功或失败状态只占用一个bit位，与boolean数组相比更节省内存。BitSet使用long[]数组来存储这些数据，意味着16个值(64bit)的数组可以存储1024个调用状态。

执行监控范围

计算失败率需要填满环形缓冲区。如果环形缓冲区的大小为10，则必须至少请求满10次，才会进行故障率的计算，如果仅仅请求了9次，即使9个请求都失败，熔断器也不会打开。

请求拦截控制

但是CLOSE状态下的缓冲区大小设置为10并不意味着只会进入10个请求，在熔断器打开之前的所有请求都会被放入。

状态转换机制

  • 当故障率高于设定的阈值时，熔断器状态会从由CLOSE变为OPEN。这时所有的请求都会抛出CallNotPermittedException异常。
  • 当经过一段时间后，熔断器的状态会从OPEN变为HALF_OPEN，HALF_OPEN状态下同样会有一个Ring Bit Buffer，用来计算HALF_OPEN状态下的故障率，如果高于配置的阈值，会转换为OPEN，低于阈值则装换为CLOSE。
  • CLOSE状态下的缓冲区不同的地方在于，HALF_OPEN状态下的缓冲区大小会限制请求数，只有缓冲区大小的请求数会被放入。
  • DISABLED（始终允许访问）和FORCED_OPEN（始终拒绝访问）。这两个状态不会生成熔断器事件（除状态装换外），并且不会记录事件的成功或者失败。退出这两个状态的唯一方法是触发状态转换或者重置熔断器。

SpringBoot的整合方式

resilience4j-spring-boot集成了circuitbeaker、retry、bulkhead、ratelimiter几个模块，因为后续还要学习其他模块，就直接引入resilience4j-spring-boot依赖。

maven 的配置 pom.xml

测试使用的IDE为idea，使用的springboot进行学习测试，首先引入maven依赖：

1
2
3
4
5
java复制代码<dependency>
    <groupId>io.github.resilience4j</groupId>
    <artifactId>resilience4j-spring-boot</artifactId>
    <version>0.9.0</version>
</dependency>
application.yml配置
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
yaml复制代码resilience4j:
  circuitbreaker:
    configs:
      default:
        ringBufferSizeInClosedState: 5 # 熔断器关闭时的缓冲区大小
        ringBufferSizeInHalfOpenState: 2 # 熔断器半开时的缓冲区大小
        waitDurationInOpenState: 10000 # 熔断器从打开到半开需要的时间
        failureRateThreshold: 60 # 熔断器打开的失败阈值
        eventConsumerBufferSize: 10 # 事件缓冲区大小
        registerHealthIndicator: true # 健康监测
        automaticTransitionFromOpenToHalfOpenEnabled: false # 是否自动从打开到半开，不需要触发
        recordFailurePredicate:    com.example.resilience4j.exceptions.RecordFailurePredicate # 谓词设置异常是否为失败
        recordExceptions: # 记录的异常
          - com.hyts.resilience4j.exceptions.Service1Exception
          - com.hyts.resilience4j.exceptions.Service2Exception
        ignoreExceptions: # 忽略的异常
          - com.example.resilience4j.exceptions.BusinessAException
    instances:
      service1:
        baseConfig: default
        waitDurationInOpenState: 5000
        failureRateThreshold: 20
      service2:
        baseConfig: default

可以配置多个熔断器实例，使用不同配置或者覆盖配置。

保护的后端服务

以一个后端服务为例，利用熔断器保护该服务。

1
2
3
java复制代码interface RemoteService {
    List<User> process() throws TimeoutException, InterruptedException;
}
连接器调用该服务

这是调用远端服务的连接器，我们通过调用连接器中的方法来调用后端服务。

1
2
3
4
5
6
7
java复制代码public RemoteServiceConnector{
    public List<User> process() throws TimeoutException, InterruptedException {
        List<User> users;
        users = remoteServic.process();
        return users;
    }
}
监控熔断器状态及事件

各个配置项的作用，需要获取特定时候的熔断器状态：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
java复制代码@Log4j2
public class CircuitBreakerUtil {

    /**
     * @Description: 获取熔断器的状态
     */
    public static void getCircuitBreakerStatus(String time, CircuitBreaker circuitBreaker){
        CircuitBreaker.Metrics metrics = circuitBreaker.getMetrics();
        // Returns the failure rate in percentage.
        float failureRate = metrics.getFailureRate();
        // Returns the current number of buffered calls.
        int bufferedCalls = metrics.getNumberOfBufferedCalls();
        // Returns the current number of failed calls.
        int failedCalls = metrics.getNumberOfFailedCalls();
        // Returns the current number of successed calls.
        int successCalls = metrics.getNumberOfSuccessfulCalls();
        // Returns the max number of buffered calls.
        int maxBufferCalls = metrics.getMaxNumberOfBufferedCalls();
        // Returns the current number of not permitted calls.
        long notPermittedCalls = metrics.getNumberOfNotPermittedCalls();
        log.info(time + "state=" +circuitBreaker.getState() + " , metrics[ failureRate=" + failureRate +
                ", bufferedCalls=" + bufferedCalls +
                ", failedCalls=" + failedCalls +
                ", successCalls=" + successCalls +
                ", maxBufferCalls=" + maxBufferCalls +
                ", notPermittedCalls=" + notPermittedCalls +
                " ]"
        );
    }

    /**
     * @Description: 监听熔断器事件
     */
    public static void addCircuitBreakerListener(CircuitBreaker circuitBreaker){
        circuitBreaker.getEventPublisher()
                .onSuccess(event -> log.info("服务调用成功：" + event.toString()))
                .onError(event -> log.info("服务调用失败：" + event.toString()))
                .onIgnoredError(event -> log.info("服务调用失败，但异常被忽略：" + event.toString()))
                .onReset(event -> log.info("熔断器重置：" + event.toString()))
                .onStateTransition(event -> log.info("熔断器状态改变：" + event.toString()))
                .onCallNotPermitted(event -> log.info(" 熔断器已经打开：" + event.toString()))
        ;
    }

调用方法

CircuitBreaker支持两种方式调用，一种是程序式调用，一种是AOP使用注解的方式调用。

程序式的调用方法

在CircuitService中先注入注册器，然后用注册器通过熔断器名称获取熔断器。如果不需要使用降级函数，可以直接调用熔断器的executeSupplier方法或executeCheckedSupplier方法：

1
2
3
4
5
6
7
8
9
java复制代码public class CircuitBreakerServiceImpl{
    @Autowired
    private CircuitBreakerRegistry circuitBreakerRegistry;
    public List<User> circuitBreakerNotAOP() throws Throwable {
        CircuitBreaker circuitBreaker = circuitBreakerRegistry.circuitBreaker("service1");
        CircuitBreakerUtil.getCircuitBreakerStatus("执行开始前：", circuitBreaker);
        circuitBreaker.executeCheckedSupplier(remotServiceConnector::process);
    }
}

如果需要使用降级函数，则要使用decorate包装服务的方法，再使用Try.of().recover()进行降级处理，同时也可以根据不同的异常使用不同的降级方法：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
java复制代码public class CircuitBreakerServiceImpl {
    @Autowired
    private RemoteServiceConnector remoteServiceConnector;
    @Autowired
    private CircuitBreakerRegistry circuitBreakerRegistry;
    public List<User> circuitBreakerNotAOP(){
        // 通过注册器获取熔断器的实例
        CircuitBreaker circuitBreaker = circuitBreakerRegistry.circuitBreaker("service1");
        CircuitBreakerUtil.getCircuitBreakerStatus("执行开始前：", circuitBreaker);
        // 使用熔断器包装连接器的方法
        CheckedFunction0<List<User>> checkedSupplier = CircuitBreaker.
            decorateCheckedSupplier(circuitBreaker, remoteServiceConnector::process);
        // 使用Try.of().recover()调用并进行降级处理
        Try<List<User>> result = Try.of(checkedSupplier).
                    recover(CallNotPermittedException.class, throwable -> {
                        log.info("熔断器已经打开，拒绝访问被保护方法~");
                        CircuitBreakerUtil
                        .getCircuitBreakerStatus("熔断器打开中:", circuitBreaker);
                        List<User> users = new ArrayList();
                        return users;
                    })
                    .recover(throwable -> {
                        log.info(throwable.getLocalizedMessage() + ",方法被降级了~~");
                        CircuitBreakerUtil
                        .getCircuitBreakerStatus("降级方法中:",circuitBreaker);
                        List<User> users = new ArrayList();
                        return users;
                    });
            CircuitBreakerUtil.getCircuitBreakerStatus("执行结束后：", circuitBreaker);
            return result.get();
    }
}
AOP式的调用方法

首先在连接器方法上使用@CircuitBreaker(name=””,fallbackMethod=””)注解，其中name是要使用的熔断器的名称，fallbackMethod是要使用的降级方法，降级方法必须和原方法放在同一个类中，且降级方法的返回值需要和原方法相同，输入参数需要添加额外的exception参数，类似这样：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
java复制代码public RemoteServiceConnector{
    
    @CircuitBreaker(name = "backendA", fallbackMethod = "fallBack")
    public List<User> process() throws TimeoutException, InterruptedException {
        List<User> users;
        users = remoteServic.process();
        return users;
    }
    
    private List<User> fallBack(Throwable throwable){
        log.info(throwable.getLocalizedMessage() + ",方法被降级了~~");
        CircuitBreakerUtil.getCircuitBreakerStatus("降级方法中:", circuitBreakerRegistry.circuitBreaker("backendA"));
        List<User> users = new ArrayList();
        return users;
    }
    
    private List<User> fallBack(CallNotPermittedException e){
        log.info("熔断器已经打开，拒绝访问被保护方法~");
        CircuitBreakerUtil.getCircuitBreakerStatus("熔断器打开中:", circuitBreakerRegistry.circuitBreaker("backendA"));
        List<User> users = new ArrayList();
        return users;
    }
    
}

可使用多个降级方法，保持方法名相同，同时满足的条件的降级方法会触发最接近的一个（这里的接近是指类型的接近，先会触发离它最近的子类异常），例如如果process()方法抛出CallNotPermittedException，将会触发fallBack(CallNotPermittedException e)方法而不会触发fallBack(Throwable throwable)方法。

之后直接调用方法就可以了：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
java复制代码public class CircuitBreakerServiceImpl {
    
    @Autowired
    private RemoteServiceConnector remoteServiceConnector;
    
    @Autowired
    private CircuitBreakerRegistry circuitBreakerRegistry;
    
    public List<User> circuitBreakerAOP() throws TimeoutException, InterruptedException {
        CircuitBreakerUtil
            .getCircuitBreakerStatus("执行开始前：",circuitBreakerRegistry.circuitBreaker("backendA"));
        List<User> result = remoteServiceConnector.process();
        CircuitBreakerUtil
            .getCircuitBreakerStatus("执行结束后：", circuitBreakerRegistry.circuitBreaker("backendA"));
        return result;
    }
}
使用测试

接下来进入测试，首先我们定义了两个异常，异常A同时在黑白名单中，异常B只在黑名单中：

recordExceptions: # 记录的异常

  • com.example.resilience4j.exceptions.BusinessBException
  • com.example.resilience4j.exceptions.BusinessAException
    ignoreExceptions: # 忽略的异常
  • com.example.resilience4j.exceptions.BusinessAException
    然后对被保护的后端接口进行如下的实现：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
java复制代码public class RemoteServiceImpl implements RemoteService {
    
    private static AtomicInteger count = new AtomicInteger(0);

    public List<User> process() {
        int num = count.getAndIncrement();
        log.info("count的值 = " + num);
        if (num % 4 == 1){
            throw new BusinessAException("异常A，不需要被记录");
        }
        if (num % 4 == 2 || num % 4 == 3){
            throw new BusinessBException("异常B，需要被记录");
        }
        log.info("服务正常运行，获取用户列表");
        // 模拟数据库的正常查询
        return repository.findAll();
    }
}

使用CircuitBreakerServiceImpl中的AOP或者程序式调用方法进行单元测试，循环调用10次：

1
2
3
4
5
6
7
8
9
10
11
12
13
java复制代码public class CircuitBreakerServiceImplTest{
    
    @Autowired
    private CircuitBreakerServiceImpl circuitService;
    
    @Test
    public void circuitBreakerTest() {
        for (int i=0; i<10; i++){
            // circuitService.circuitBreakerAOP();
            circuitService.circuitBreakerNotAOP();
        }
    }
}

同时也可以看出白名单所谓的忽略，是指不计入缓冲区中（即不算成功也不算失败），有降级方法会调用降级方法，没有降级方法会抛出异常，和其他异常无异。

public class CircuitBreakerServiceImplTest{

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
ini复制代码@Autowired
private CircuitBreakerServiceImpl circuitService;

@Test
public void circuitBreakerThreadTest() throws InterruptedException {
    ExecutorService pool = Executors.newCachedThreadPool();
    for (int i=0; i<15; i++){
        pool.submit(
            // circuitService::circuitBreakerAOP
            circuitService::circuitBreakerNotAOP);
    }
    pool.shutdown();

    while (!pool.isTerminated());

    Thread.sleep(10000);
    log.info("熔断器状态已转为半开");
    pool = Executors.newCachedThreadPool();
    for (int i=0; i<15; i++){
        pool.submit(
            // circuitService::circuitBreakerAOP
            circuitService::circuitBreakerNotAOP);
    }
    pool.shutdown();

    while (!pool.isTerminated());
    for (int i=0; i<10; i++){
        
    }
}

}

1
2
3
4
5
yaml复制代码resilience4j:
  circuitbreaker:
    configs:
      myDefault:
        automaticTransitionFromOpenToHalfOpenEnabled: true # 是否自动从打开到半开

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

发表于

「这是我参与11月更文挑战的第10天，活动详情查看：2021最后一次更文挑战」

前言

  大家好，明天就是双十一了，多少IT人值守在机房和战场上，为迎接一年一度的大卖会准备着。又有多少人在双十一进行了“剁手”，疯狂的采购。

  今年的双十一来的格外的早，在10月20号就开始了双十一的预热，今年也暴雷出很多商家在原件基础上进行加价，然后再进行打折优惠，很多都比日常中的价格高很多，大家在网购的时候，注意价格变动，可以查查价格。合适的再购买。

  在项目开发过程中，在一个项目中有很多任务需要在不同的时间节点运行，因此，本篇将给大家介绍一个实用的任务调度平台XXL-JOB。下面开始正题。

什么是XXL-JOB

  XXL-JOB是一个分布式任务调度平台，XXL-JOB具有操作简单、动态配置、动态配置、弹性扩容缩容、自动注册、各种测量配置、任务进度监控、跨语言、全异步、自定义任务参数、邮件报警、故障转移、用户管理、权限控制、数据加密、动态分片等各种特性，为我们在项目中提供了一个强大的全面的任务调度平台。

  XXL-JOB起始于2015年，经过近几年的发展和完善，已经越发的全面强大。小编今天在github中查看，目前有19.8k的star,8.4k的fork,有1692次的commits，至今，XXL-JOB已接入多家公司的线上产品线，接入场景如电商业务，O2O业务和大数据作业等，据不完全统计，目前有近500家大中型企业子在使用。

github地址

  使用XXL-JOB主要分为两步配置即可完成一个简单的任务调度中心，第一步，配置调度中心，第二步配置执行器。下面开始部署。

快速开始：

  本次Spring Boot 集成XXL-JOB分布式任务调度平台使用的环境是：

1
2
3
4
markdown复制代码    Maven
    Jdk1.8
    Mysql5.7
    2.1.2 xxl-job

配置调度中心

  下载完XXL-JOB源码之后，在项目文件中找到数据库脚本信息，执行一个数据库脚本，为配置中心初始化数据。

打开项目

  如下图是下载完成之后的XXL-JOB项目结构。

图片.png

执行SQL脚本

  数据库脚本信息截图如下：图片.png
  数据库主要是初始化配置中心和初始化一个管理员用户，基本表就包含用户表、执行器表、日志表、分组表、配置表、线程锁表等相关表。表中也有相关字段的描述，比较全面介绍。

配置参数

  执行完SQL脚本之后，根据自己需要修改application.properties配置文件，主要有端口、项目名称、报警邮件配置、数据库配置、调度线程池最大线程配置、调度中心日志表数据配值等，配值方式与Spring boot配置方式相同。基本上采用默认的配置即可，只需要修改一个数据源。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
js复制代码### web
server.port=8080
server.context-path=/xxl-job-admin

### actuator
management.context-path=/actuator
management.health.mail.enabled=false

### resources
spring.mvc.servlet.load-on-startup=0
spring.mvc.static-path-pattern=/static/**
spring.resources.static-locations=classpath:/static/

### freemarker
spring.freemarker.templateLoaderPath=classpath:/templates/
spring.freemarker.suffix=.ftl
spring.freemarker.charset=UTF-8
spring.freemarker.request-context-attribute=request
spring.freemarker.settings.number_format=0.##########

### mybatis
mybatis.mapper-locations=classpath:/mybatis-mapper/*Mapper.xml
### xxl-job, datasource
spring.datasource.url=jdbc:mysql://127.0.0.1:3306/xxl_job?useUnicode=true&characterEncoding=utf8&zeroDateTimeBehavior=convertToNull&useSSL=false&serverTimezone=UTC
spring.datasource.username=test
spring.datasource.password=12345678test
spring.datasource.driver-class-name=com.mysql.jdbc.Driver

spring.datasource.type=org.apache.tomcat.jdbc.pool.DataSource
spring.datasource.tomcat.max-wait=10000
spring.datasource.tomcat.max-active=30
spring.datasource.tomcat.test-on-borrow=true
spring.datasource.tomcat.validation-query=SELECT 1
spring.datasource.tomcat.validation-interval=30000

## xxl-job, triggerpool max size
xxl.job.triggerpool.fast.max=200
xxl.job.triggerpool.slow.max=100

### xxl-job, log retention days
xxl.job.logretentiondays=30

启动调度中心

  配置好所有的参数之后，启动XxlJobAdminApplication，任务调度中心启动完成之后，浏览器输入如下地址，进入管理平台。

  默认地址为：http://127.0.0.1:8080/xxl-job-admin

进入后台

  输入默认的用户名密码admin/123456，即可登陆到平台。
图片.png
  默认首页是数据报表，可以一目了然的看到相关任务调度信息。
图片.png

配置执行器

  本次配置执行器是基于Spring Boot搭建一个执行器，在项目源码中作者也给出了相关的示例，大家可以参考。首先在IDEA中搭建一个基础的Spring Boot项目。搭建一个最基础的就可以，本次将以现成的项目进行介绍，如需要Spring Boot项目可以参考往期的文章链接。

【Spring Boot 快速入门】三、Spring Boot集成MyBatis，可以连接数据库啦！

引入pom

  在Spring Boot项目中引入xxl-job-core包，注意版本需要与Spring Boot 版本相互兼容，否则会报很多异常。

1
2
3
4
5
6
xml复制代码    <!--定时调度项目-->
        <dependency>
            <groupId>com.xuxueli</groupId>
            <artifactId>xxl-job-core</artifactId>
            <version>2.1.2</version>
        </dependency>

XxlJobConfig配置类

  创建一个XxlJobConfig配置类，便于读取配置文件中的配置信息。加入@ComponentScan注解，扫描的包是执行器所在的位置包名。这样一个简单的XxlJobConfig配置类就完成了。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
js复制代码@Configuration
@ConditionalOnProperty(prefix = "xxl", value = "enable", havingValue = "true")
@ComponentScan(basePackages = "com.juejin.scheduling")
public class XxlJobConfig {
    private Logger logger = LoggerFactory.getLogger(XxlJobConfig.class);

    @Value("${xxl.job.admin.addresses}")
    private String adminAddresses;

    @Value("${xxl.job.executor.appname}")
    private String appName;

    @Value("${xxl.job.executor.ip}")
    private String ip;

    @Value("${xxl.job.executor.port}")
    private int port;

    @Value("${xxl.job.accessToken}")
    private String accessToken;

    @Value("${xxl.job.executor.logpath}")
    private String logPath;

    @Value("${xxl.job.executor.logretentiondays}")
    private int logRetentionDays;

    @Bean()
    public XxlJobSpringExecutor xxlJobExecutor() {
        logger.info(">>>>>>>>>>> xxl-job config init.");
        XxlJobSpringExecutor xxlJobSpringExecutor = new XxlJobSpringExecutor();
        xxlJobSpringExecutor.setAdminAddresses(adminAddresses);
        xxlJobSpringExecutor.setAppName(appName);
        xxlJobSpringExecutor.setIp(ip);
        xxlJobSpringExecutor.setPort(port);
        xxlJobSpringExecutor.setAccessToken(accessToken);
        xxlJobSpringExecutor.setLogPath(logPath);
        xxlJobSpringExecutor.setLogRetentionDays(logRetentionDays);
        return xxlJobSpringExecutor;
    }
}

编写执行器

  在上一节中已经制定了执行器所在的包，下面在执行此包中创建一个执行器，需要加入@Component注解。在类中创建执行器方法，加入@XxlJob(“TestXXLScheduled”)，其中指定执行器方法名称，需要在调度中心后台中配置这个注解方法，以便调度中心调用此方法。在方法中可以加入自己的业务逻辑即可。这样一个简单的执行器几配置好了。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
js复制代码@Slf4j
@Component
public class TestXXLScheduled {


    @XxlJob("TestXXLScheduled")
    public ReturnT<String> TestXXLScheduled(String param){
     XxlJobLogger.log("测试XXL-JOB分布式任务调度平台---开始");
        //业务逻辑代码块
        System.out.println("测试XXL-JOB分布式任务调度平台执行了");
       
      XxlJobLogger.log("测试XXL-JOB分布式任务调度平台---结束");
    }
}

添加执行器

  登录XXL-JOB分布式任务调度平台，进入到执行器管理中，选择新增执行器，依次输入AppName、名称、排序等信息，由于是测试使用，选择手动注册，直接填写127.0.0.1：8081本地的执行器项目的地址即可。然后点击保存。

图片.png
  在任务管理中新增一个任务，选择刚刚创建的执行器，依次输入任务名称，执行时间间隔、选择路由策略、执行策略、然后输入 @XxlJob(“TestXXLScheduled”)中的名称TestXXLScheduled到JobHandler中，输入负责人即可，任务就配置完成了。
图片.png

运行执行器

  为了便于测试，本次将执行间隔设置为5秒，分别启动控制中心和执行器两个项目。可以看到执行器中输出了日志信息，代表执行器创建成功，任务在执行了。

图片.png

结语

  好了，以上就是Spring Boot 集成XXL-JOB分布式任务调度平台的详细过程，感谢您的阅读，希望您喜欢，如对您有帮助，欢迎点赞收藏。如有不足之处，欢迎评论指正。下次见。

  作者介绍：【小阿杰】一个爱鼓捣的程序猿，JAVA开发者和爱好者。公众号【Java全栈架构师】维护者，欢迎关注阅读交流。

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

发表于

这是我参与11月更文挑战的第10天，活动详情查看：2021最后一次更文挑战

前言

作为一名Java开发人员，我想Java字节码文件是无论如何都会接触到的，也是要读懂的。面试或者是自己开始研究Java的一些底层原理，大都会遇上要字节码文件的时候。

接下里咱们一起来聊聊如何idea有那几种方式查看字节码文件。

idea查看字节码文件

1.1、javap命令的使用

在jdk工具包的bin目录下，有一个java可执行文件javap，该工具可以查看java编译后的class文件。使用命令如下命令进行查看:

image-20211110195828154

这个每个Jdk中都会有的。（配置了环境变量就可以直接在idea中使用）

随便写个程序，然后点开idea下部的Terminal，转到编译完后的class目录中, 用javap -c StringTest.class 就可以直接打印出字节码文件。

image-20211110200211930

但是这样子去看，仍然要我们自己一点点找，并且去分析，不能非常的直观的看。

所以一下子就可以想到用idea插件来查看了。（idea插件不要装太多了，容易导致idea卡顿，但是这个我觉得在字节码文件方面还是挺香的，不想用的话，卸掉即可）

1.2、Idea插件 jclasslibBytecodeViewer

image-20211110201129918

直接搜索，然后安装即可。

安装完成之后，我们像之前一样编译代码，这次如何查看勒？

我们点击idea顶部菜单栏中的view中，会出现一个 jclasslibBytecodeViewer标识。

image-20211110201738988

点击会在右边展示出一个

image-20211110201859610

在这边idea都帮你分好类，不用担心因为代码太长，从而导致分析麻烦。

image-20211110202005625

另外我们不认识这些字节码命令，但是只要点击一下，它就会直接跳到浏览器的jdk官网处的字节码命令去(不过是英文版本)

image-20211110202151445

我觉得这点对于刚学的小伙伴，是非常非常实用的。

自言自语

纸上得来终觉浅，绝知此事要躬行。

大家好，我是博主宁在春：主页

一名喜欢文艺却踏上编程这条道路的小青年。

希望：我们，待别日相见时，都已有所成。

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

发表于

这是我参与11月更文挑战的第10天，活动详情查看：2021最后一次更文挑战

本系列代码地址：github.com/JoJoTec/spr…

在前面一节，我们实现了 FeignClient 粘合 resilience4j 的 Retry 实现重试。细心的读者可能会问，为何在这里的实现，不把断路器和线程限流一起加上呢：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
ini复制代码
@Bean
public FeignDecorators.Builder defaultBuilder(
        Environment environment,
        RetryRegistry retryRegistry
) {
    //获取微服务名称
    String name = environment.getProperty("feign.client.name");
    Retry retry = null;
    try {
        retry = retryRegistry.retry(name, name);
    } catch (ConfigurationNotFoundException e) {
        retry = retryRegistry.retry(name);
    }

    //覆盖其中的异常判断，只针对 feign.RetryableException 进行重试，所有需要重试的异常我们都在 DefaultErrorDecoder 以及 Resilience4jFeignClient 中封装成了 RetryableException
    retry = Retry.of(name, RetryConfig.from(retry.getRetryConfig()).retryOnException(throwable -> {
        return throwable instanceof feign.RetryableException;
    }).build());

    return FeignDecorators.builder().withRetry(
            retry
    );
}

主要原因是，这里增加断路器以及线程隔离，其粒度是微服务级别的，这样的坏处是：

  • 微服务中只要有一个实例一直异常，整个微服务就会被断路
  • 微服务只要有一个方法一直异常，整个微服务就会被断路
  • 微服务的某个实例比较慢，其他实例正常，但是轮询的负载均衡模式导致线程池被这个实例的请求堵满。由于这一个慢实例，倒是整个微服务的请求都被拖慢

回顾我们想要实现的微服务重试、断路、线程隔离

请求重试

来看几个场景：

1.在线发布服务的时候，或者某个服务出现问题下线的时候，旧服务实例已经在注册中心下线并且实例已经关闭，但是其他微服务本地有服务实例缓存或者正在使用这个服务实例进行调用，这时候一般会因为无法建立 TCP 连接而抛出一个 java.io.IOException，不同框架使用的是这个异常的不同子异常，但是提示信息一般有 connect time out 或者 no route to host。这时候如果重试，并且重试的实例不是这个实例而是正常的实例，就能调用成功。如下图所示：

image

2.当调用一个微服务返回了非 2XX 的响应码：

a) 4XX：在发布接口更新的时候，可能调用方和被调用方都需要发布。假设新的接口参数发生变化，没有兼容老的调用的时候，就会有异常，一般是参数错误，即返回 4XX 的响应码。例如新的调用方调用老的被调用方。针对这种情况，重试可以解决。但是为了保险，我们对于这种请求已经发出的，只重试 GET 方法（即查询方法，或者明确标注可以重试的非 GET 方法），对于非 GET 请求我们不重试。如下图所示：

image

b) 5XX：当某个实例发生异常的时候，例如连不上数据库，JVM Stop-the-world 等等，就会有 5XX 的异常。针对这种情况，重试也可以解决。同样为了保险，我们对于这种请求已经发出的，只重试 GET 方法（即查询方法，或者明确标注可以重试的非 GET 方法），对于非 GET 请求我们不重试。如下图所示：

image

3.断路器打开的异常：后面我们会知道，我们的断路器是针对微服务某个实例某个方法级别的，如果抛出了断路器打开的异常，请求其实并没有发出去，我们可以直接重试。

4.限流异常：后面我们会知道，我们给调用每个微服务实例都做了单独的线程池隔离，如果线程池满了拒绝请求，会抛出限流异常，针对这种异常也需要直接重试。

这些场景在线上在线发布更新的时候，以及流量突然到来导致某些实例出现问题的时候，还是很常见的。如果没有重试，用户会经常看到异常页面，影响用户体验。所以这些场景下的重试还是很必要的。对于重试，我们使用 resilience4j 作为我们整个框架实现重试机制的核心。

微服务实例级别的线程隔离

再看下面一个场景：

image

微服务 A 通过同一个线程池调用微服务 B 的所有实例。如果有一个实例有问题，阻塞了请求，或者是响应非常慢。那么久而久之，这个线程池会被发送到这个异常实例的请求而占满，但是实际上微服务 B 是有正常工作的实例的。

为了防止这种情况，也为了限制调用每个微服务实例的并发（也就是限流），我们使用不同线程池调用不同的微服务的不同实例。这个也是通过 resilience4j 实现的。

微服务实例方法粒度的断路器

如果一个实例在一段时间内压力过大导致请求慢，或者实例正在关闭，以及实例有问题导致请求响应大多是 500，那么即使我们有重试机制，如果很多请求都是按照请求到有问题的实例 -> 失败 -> 重试其他实例，这样效率也是很低的。这就需要使用断路器。

在实际应用中我们发现，大部分异常情况下，是某个微服务的某些实例的某些接口有异常，而这些问题实例上的其他接口往往是可用的。所以我们的断路器不能直接将这个实例整个断路，更不能将整个微服务断路。所以，我们使用 resilience4j 实现的是微服务实例方法级别的断路器（即不同微服务，不同实例的不同方法是不同的断路器）

使用 resilience4j 的断路器和线程限流器

下面我们先来看下断路器的相关配置，来理解下 resilience4j 断路器的原理:

CircuitBreakerConfig.java

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
ini复制代码//判断一个异常是否记录为断路器失败，默认所有异常都是失败，这个相当于黑名单
private Predicate<Throwable> recordExceptionPredicate = throwable -> true;
//判断一个返回对象是否记录为断路器失败，默认只要正常返回对象就不认为是失败
private transient Predicate<Object> recordResultPredicate = (Object object) -> false;
//判断一个异常是否可以不认为是断路器失败，默认所有异常都是失败，这个相当于白名单
private Predicate<Throwable> ignoreExceptionPredicate = throwable -> false;
//获取当前时间函数
private Function<Clock, Long> currentTimestampFunction = clock -> System.nanoTime();
//当前时间的单位
private TimeUnit timestampUnit = TimeUnit.NANOSECONDS;
//异常名单，指定一个 Exception 的 list，所有这个集合中的异常或者这些异常的子类，在调用的时候被抛出，都会被记录为失败。其他异常不会被认为是失败，或者在 ignoreExceptions 中配置的异常也不会被认为是失败。默认是所有异常都认为是失败。
private Class<? extends Throwable>[] recordExceptions = new Class[0];
//异常白名单，在这个名单中的所有异常及其子类，都不会认为是请求失败，就算在 recordExceptions 中配置了这些异常也没用。默认白名单为空。
private Class<? extends Throwable>[] ignoreExceptions = new Class[0];
//失败请求百分比，超过这个比例，`CircuitBreaker`就会变成`OPEN`状态，默认为 50%
private float failureRateThreshold = 50;
//当`CircuitBreaker`处于`HALF_OPEN`状态的时候，允许通过的请求数量
private int permittedNumberOfCallsInHalfOpenState = 10;
//滑动窗口大小，如果配置`COUNT_BASED`默认值100就代表是最近100个请求，如果配置`TIME_BASED`默认值100就代表是最近100s的请求。
private int slidingWindowSize = 100;
//滑动窗口类型，`COUNT_BASED`代表是基于计数的滑动窗口，`TIME_BASED`代表是基于计时的滑动窗口
private SlidingWindowType slidingWindowType = SlidingWindowType.COUNT_BASED;
//最小请求个数。只有在滑动窗口内，请求个数达到这个个数，才会触发`CircuitBreaker`对于是否打开断路器的判断。
private int minimumNumberOfCalls = 100;
//对应 RuntimeException 的 writableStackTrace 属性，即生成异常的时候，是否缓存异常堆栈
//断路器相关的异常都是继承 RuntimeException，这里统一指定这些异常的 writableStackTrace
//设置为 false，异常会没有异常堆栈，但是会提升性能
private boolean writableStackTraceEnabled = true;
//如果设置为`true`代表是否自动从`OPEN`状态变成`HALF_OPEN`，即使没有请求过来。
private boolean automaticTransitionFromOpenToHalfOpenEnabled = false;
//在断路器 OPEN 状态等待时间函数，默认是固定 60s，在等待与时间后，会退出 OPEN 状态
private IntervalFunction waitIntervalFunctionInOpenState = IntervalFunction.of(Duration.ofSeconds(60));
//当返回某些对象或者异常时，直接将状态转化为另一状态，默认是没有配置任何状态转换机制
private Function<Either<Object, Throwable>, TransitionCheckResult> transitionOnResult = any -> TransitionCheckResult.noTransition();
//当慢调用达到这个百分比的时候，`CircuitBreaker`就会变成`OPEN`状态
//默认情况下，慢调用不会导致`CircuitBreaker`就会变成`OPEN`状态，因为默认配置是百分之 100
private float slowCallRateThreshold = 100;
//慢调用时间，当一个调用慢于这个时间时，会被记录为慢调用
private Duration slowCallDurationThreshold = Duration.ofSeconds(60);
//`CircuitBreaker` 保持 `HALF_OPEN` 的时间。默认为 0， 即保持 `HALF_OPEN` 状态，直到 minimumNumberOfCalls 成功或失败为止。
private Duration maxWaitDurationInHalfOpenState = Duration.ofSeconds(0);

然后是线程隔离的相关配置：

ThreadPoolBulkheadConfig.java

1
2
3
4
5
6
7
8
9
10
11
12
ini复制代码//以下五个参数对应 Java 线程池的配置，我们这里就不再赘述了
private int maxThreadPoolSize = Runtime.getRuntime().availableProcessors();
private int coreThreadPoolSize = Runtime.getRuntime().availableProcessors();
private int queueCapacity = 100;
private Duration keepAliveDuration = Duration.ofMillis(20);
private RejectedExecutionHandler rejectedExecutionHandler = new ThreadPoolExecutor.AbortPolicy();
//对应 RuntimeException 的 writableStackTrace 属性，即生成异常的时候，是否缓存异常堆栈
//限流器相关的异常都是继承 RuntimeException，这里统一指定这些异常的 writableStackTrace
//设置为 false，异常会没有异常堆栈，但是会提升性能
private boolean writableStackTraceEnabled = true;
//Java 很多 Context 传递都基于 ThreadLocal，但是这里相当于切换线程了，某些任务需要维持上下文，可以通过实现 ContextPropagator 加入这里即可
private List<ContextPropagator> contextPropagators = new ArrayList<>();

在添加了上一节所说的 resilience4j-spring-cloud2 依赖之后，我们可以这样配置断路器和线程隔离：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
yaml复制代码resilience4j.circuitbreaker:
  configs:
    default:
      registerHealthIndicator: true
      slidingWindowSize: 10
      minimumNumberOfCalls: 5
      slidingWindowType: TIME_BASED
      permittedNumberOfCallsInHalfOpenState: 3
      automaticTransitionFromOpenToHalfOpenEnabled: true
      waitDurationInOpenState: 2s
      failureRateThreshold: 30
      eventConsumerBufferSize: 10
      recordExceptions:
        - java.lang.Exception
resilience4j.thread-pool-bulkhead:
  configs:
    default:
      maxThreadPoolSize: 50
      coreThreadPoolSize: 10
      queueCapacity: 1000

如何实现微服务实例方法粒度的断路器

我们要实现的是每个微服务的每个实例的每个方法都是不同的断路器，我们需要拿到：

  • 微服务名
  • 实例 ID，或者能唯一标识一个实例的字符串
  • 方法名：可以是 URL 路径，或者是方法全限定名。

我们这里方法名采用的是方法全限定名称，而不是 URL 路径，因为有些 FeignClient 将参数放在了路径上面，例如使用 @PathVriable，如果参数是类似于用户 ID 这样的，那么一个用户就会有一个独立的断路器，这不是我们期望的。所以采用方法全限定名规避这个问题。

那么在哪里才能获取到这些呢？回顾下 FeignClient 的核心流程，我们发现需要在实际调用的时候，负载均衡器调用完成之后，才能获取到实例 ID。也就是在 org.springframework.cloud.openfeign.loadbalancer.FeignBlockingLoadBalancerClient 调用完成之后。所以，我们在这里植入我们的断路器代码实现断路器。

另外就是配置粒度，可以每个 FeignClient 单独配置即可，不用到方法这一级别。举个例子如下：

1
2
3
4
5
6
yaml复制代码resilience4j.circuitbreaker:
  configs:
    default:
      slidingWindowSize: 10
    feign-client-1:
      slidingWindowSize: 100

下面这段代码，contextId 即 feign-client-1 这种，不同的微服务实例方法 serviceInstanceMethodId 不同。如果 contextId 对应的配置没找到，就会抛出 ConfigurationNotFoundException，这时候我们就读取并使用 default 配置。

1
2
3
4
5
ini复制代码try {
    circuitBreaker = circuitBreakerRegistry.circuitBreaker(serviceInstanceMethodId, contextId);
} catch (ConfigurationNotFoundException e) {
    circuitBreaker = circuitBreakerRegistry.circuitBreaker(serviceInstanceMethodId);
}

如何实现微服务实例线程限流器

对于线程隔离限流器，我们只需要微服务名和实例 ID，同时这些线程池只做调用，所以其实和断路器一样，可以放在 org.springframework.cloud.openfeign.loadbalancer.FeignBlockingLoadBalancerClient 调用完成之后，植入线程限流器相关代码实现。

微信搜索“我的编程喵”关注公众号，每日一刷，轻松提升技术，斩获各种offer

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

发表于

「这是我参与11月更文挑战的第9天，活动详情查看：2021最后一次更文挑战」。

使用guava作内存缓存，大多数小伙伴应该都使用过，通过CacheBuilder创建LoadingCache一个kv格式的缓存，如果我们需要缓存的只是一个value呢？

针对这种场景，接下来介绍一种基于Supplier来实现的缓存方式

1. Supplier使用姿势

guava的Supplier与jdk的Supplier从接口定义上来看没什么区别，对外只提供了一个get()方法

1
2
3
4
5
6
java复制代码@FunctionalInterface
@GwtCompatible
public interface Supplier<T> extends java.util.function.Supplier<T> {
    @CanIgnoreReturnValue
    T get();
}

重点需要关注的是Supplier创建的姿势，借助Suppliers来实现

下面是几个常见的创建姿势：

  • memoize: delegate为具体的获取值的委托类，需要注意的是，delegate的具体实现只会在首次时调用；这种方式相当于持久缓存
  • memoizeWithExpiration：delegate的返回值，会缓存一段时间；缓存时间过后，会重新调用一下delegate来获取返回值
  • ofInstance: 直接传参
1
2
3
4
5
java复制代码public static <T> Supplier<T> memoize(Supplier<T> delegate)

public static <T> Supplier<T> memoizeWithExpiration(Supplier<T> delegate, long duration, TimeUnit unit)

public static <T> Supplier<T> ofInstance(@Nullable T instance)

基于上面的方法描述，如果我们想实现一个10s缓存，那么可以选择memoizeWithExpiration来实现

1
2
3
4
5
6
7
java复制代码AtomicInteger atomicInteger = new AtomicInteger(1);
Supplier<Integer> cache = Suppliers.memoizeWithExpiration(this::ret, 10, TimeUnit.SECONDS);

private int ret() {
    System.out.println("------- 更新 value --------");
    return atomicInteger.getAndAdd(2) ;
}

上面定义了一个内存缓存cache, 缓存10s，调用时若缓存失效，会重新调用ret()刷新缓存

测试case就比较简单了

1
2
3
4
5
6
7
8
9
java复制代码@Test
public void testSupplier() throws InterruptedException {
    for (int i = 0; i < 10; i++) {
        System.out.print(cache.get() + " | ");
    }
    System.out.println();
    Thread.sleep(10000);
    System.out.println(cache.get());
}

输出如下

1
2
3
4
diff复制代码------- 更新 value --------
1 | 1 | 1 | 1 | 1 | 1 | 1 | 1 | 1 | 1 | 
------- 更新 value --------
3

2. 缓存刷新

使用Supplier当缓存时，需要注意的一点就是没有缓存失效的方法可供调用；对于LoadingCache若是想失效缓存，可以通过调用 invalidate来主动失效指定的缓存，那么Supplier 可以怎么整？

  • 直接重新赋值

比如当我们希望刷新时，可以直接覆盖就的supplier即可

1
2
3
java复制代码public void refresh() {
    cache = Suppliers.memoizeWithExpiration(this::ret, 10, TimeUnit.SECONDS);
}

一灰灰的联系方式

尽信书则不如无书，以上内容，纯属一家之言，因个人能力有限，难免有疏漏和错误之处，如发现bug或者有更好的建议，欢迎批评指正，不吝感激

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

发表于

这是我参与11月更文挑战的第3天，活动详情查看：2021最后一次更文挑战

写在前面

Redis 是完全开源的，高性能的 key-value 数据库，受到越来越多的业务场景应用。对于”发布/订阅”的消息模式，大家也许都比较了解，但是其实现原理及应用是否还存在模糊呢？

今天计划同大家一起，深入浅出讲透 Redis 发布订阅，尽量通俗易懂，让大家轻松上手。

发布/订阅模式

关于发布/订阅模式

在软件架构中，发布/订阅是一种消息模式，消息的发送者（称为发布者）不会将消息直接发送给特定的接收者（称为订阅者），而是通过消息通道广播出去，让订阅该消息主题的订阅者消费到。

Redis提供了发布订阅功能，可以用于消息的传输，Redis的发布订阅机制包括三个部分：发布者（Publisher），订阅者（Subscriber）和频道（Channel）。发布/订阅者模式最大的特点就是实现了松耦合。

Redis发布订阅分类

  • 频道的发布订阅
  • 模式的发布订阅

下面来分别详细阐述一下其实现原理及应用。

频道的发布订阅

实现原理

Redis将所有频道的订阅关系都保存在服务器状态的 pubsub_channels 字典，字典的键是某个被订阅的频道，而对应值则是一个链表，链表里记录了所有订阅这个频道的客户端。

1
2
3
4
5
6
7
8
arduino复制代码struct redisServer{
    //...

    // 保存所有频道订阅关系
    dict *pubsub_channels;

    //...
}

一个pubsub_channels字典示例如下：

  • client-1、client-2、client-3 三个客户端正在订阅 “article.tech” 频道
  • 客户端 client-4 正在订阅 “article.mysql” 频道
  • client-5、client-6 两个客户端正在订阅 “article.redis” 频道

订阅频道

相关命令：

1
css复制代码SUBSCRIBE channel [channel …]

当客户端执行SUBSCRIBE命令订阅某个或某些频道的时候，这个客户端与被订阅频道之间就建立起了一种订阅关系。

建立订阅关系执行分两种情况：

1）该频道已有其他订阅者

该频道在 pubsub_channels 字典中存在订阅者链表，将此客户端添加至订阅者链表末尾即可；

2）该频道暂无订阅者

该频道在 pubsub_channels 字典中不存在订阅者链表，首先在字典中为频道创建一个键，并将这个键的值设置为空链表，然后将客户端添加到链表，成为链表的第一个元素。

参考示例：

客户端client-10086 执行命令：

1
arduino复制代码SUBSCRIBE "article.mysql" "article.java"

执行SUBSCRIBE命令之后的 pubsub_channels 字典：

退订频道

相关命令：

1
css复制代码UNSUBSCRIBE channel [channel …]

当客户端退订某个或某些频道的时候，服务器将从 pubsub_channels 中解除客户端与被退订频道之间的关联。

解除订阅关系执行过程：

1）根据被退订频道的名字，在 pubsub_channels 字典中找到频道对应的订阅者链表，然后从订阅者链表中删除退订客户端的信息；

2）假如删除退订客户端后，频道的订阅者链表变成了空链表，那么说明这个频道已无任何订阅者了，将从 pubsub_channels 字典中删除频道对应的键。

参考示例：

客户端client-10086 执行命令：

1
arduino复制代码UNSUBSCRIBE "article.mysql" "article.java" "article.a"

执行SUBSCRIBE命令之后的pubsub_channels字典：

我们注意到虽然退订频道里包含 “article.a”，但是由于 “article.a” 在 pubsub_channels 字典中不存在，则被忽略。

模式的发布订阅

模式与频道的区别，简单理解模式是多个频道的组合。

实现原理

Redis将所有模式的订阅关系都保存在服务器状态的 pubsub_patterns 链表，链表的每个节点都包含着一个 pubsub Pattern 结构，这个结构的 pattern 属性记录了被订阅的模式，而 client 属性则记录了订阅模式的客户端。

1
2
3
4
5
6
7
8
arduino复制代码struct redisServer{
    //...

    // 保存所有模式订阅关系
    dict *pubsub_patterns;

    //...
}

一个pubsub_patterns链表示例：

  • 客户端 client-7 正在订阅模式 “book.*“
  • 客户端 client-8 正在订阅模式 “column.*“

订阅模式

相关命令：

1
sql复制代码PSUBSCRIBE pattern [pattern …]

当客户端执行 PSUBSCRIBE 命令订阅某个或某些模式的时候，服务器会对每个被订阅的模式执行以下两个操作：

1）新建一个 pubsubPattern结果，将结构的 pattern 属性设置为被订阅的模式，client 属性设置为订阅模式的客户端；

2）将pubsubPattern结构添加到 pubsub_patterns 链表的尾部。

参考示例：

客户端 client-9 执行命令：

1
arduino复制代码PSUBSCRIBE "article.*"

执行 PSUBSCRIBE 命令之后的 pubsub_patterns 链表：

退订模式

相关命令：

1
sql复制代码PUNSUBSCRIBE pattern [pattern …]

当客户端退订某个或某些模式的时候，服务器将从 pubsub_patterns 链表中查找并删除那些 pattern 属性为被退订模式，并且client 属性为执行退订命令的客户端的 pubsubPattern 结构。

简单理解即：查找client、pattern 均相同的 pubsubPattern 并删除。

参考示例：

客户端 client-9 执行命令：

1
arduino复制代码PUNSUBSCRIBE "article.*"

执行 PUNSUBSCRIBE 命令之后的 pubsub_patterns 链表：

发消息

相关命令：

1
xml复制代码PUBLISH <channel> <message>

将消息message 发送给channel 频道的所有订阅者，以及发送给 channel 频道相匹配模式的订阅者。

发消息执行过程：

1）在 pubsub_channels 字典里找到频道 channel 的订阅者列表，然后将消息发送给列表上所有客户端；

2）遍历 pubsub_patterns 链表，查找与channel 频道相匹配的 pattern 模式，并将消息发送给订阅了这些 pattern 模式的客户端。

参考示例：

当前 pubsub_channels 字典状态如下：

当前 pubsub_patterns 链表状态如下：

此时某客户端执行如下命令：

1
arduino复制代码PUBLISH "article.redis" "hello"

发消息执行过程：

  • PUBLISH 命令会先将消息 “hello” 发送给 “articleredis” 频道的所有订阅者（client-5、client-6）；
  • 然后在 pubsub_patterns 链表中查找是否有被订阅的模式与 “article.redis” 频道相匹配，随机找到 “article.*“ 模式，随即将消息 “hello” 发送给client-9。

发布订阅原理小结

发布订阅原理，主要小结如下：

  • pubsub_channels 字典保存了所有频道的订阅关系：SUBSCRIBE 命令负责将客户端与被订阅的频道关联到字典，而UNSUBSCRIBE 命令负责解除客户端和被退订频道之间的关联；
  • pubsub_patterns 链表保存了所有模式的订阅关系：PSUBSCRIBE 命令负责将客户端与被订阅的模式记录到链表，而PUNSUBSCRIBE 命令负责移除客户端和被退订模式在链表中的记录；
  • PUBLISH 命令通过访问pubsub_channels 字典来向频道的所有订阅者发送消息，通过访问 pubsub_patterns 链表向所有匹配频道的模式的订阅者发送消息。

实际应用案例经验分享

背景描述

我们以信息订阅分发网站为例，假设文章结构如下所示：

各 chat 相当于“频道”，前端、后端、测试 等分类可理解为一类频道的组合，成为“模式”。

数据剖析

假如用户进行 chat（频道） 及分类（模式）的订阅：

  • 用户 A 预定了频道 chat-1
  • 用户 B 预定了频道 chat-16 和模式“后端”
  • 用户 C 预定了模式“前端”和频道 chat-101

频道和模式的订阅关系如图所示：

Redis 记录发布订阅频道的数据格式如下：

Redis 记录发布订阅模式的数据格式如下：

操作执行

此时某客户端执行如下命令：

1
arduino复制代码PUBLISH "chat-1" "hello"

执行过程如下：

  • PUBLISH 命令会先将消息 “hello” 发送给 “chat-1” 频道的所有订阅者 用户 A；
  • 然后在 pubsub_patterns 链表中查找是否有被订阅的模式与 “chat-1” 频道相匹配，随机找到 “前端” 模式，随即将消息 “hello” 发送给 用户 C。

其他消息发送执行过程，对于订阅关系及消息发送与上述场景同理，大家可以尝试自行分析。

能力应用

Redis 发布订阅应用场景比较广泛，类似微博/微信公众号这种关注/订阅以及消息推送能力，同样还可以作为实时消息系统（类似聊天/群聊能力支持）。

利用 Redis 发布订阅可以快速实现用户订阅/关注关系维护以及后续消息推送能力，本 文从概念到原理分析，再到具体案例应用讲解，算是带大家基本熟悉了 Redis 发布订阅的全貌，希望对你今后的工作有所帮助，谢谢。

  • END -

作者：架构精进之路，十年研发风雨路，大厂架构师，CSDN 博客专家，专注架构技术沉淀学习及分享，职业与认知升级，坚持分享接地气儿的干货文章，期待与你一起成长。

关注并私信我回复“01”，送你一份程序员成长进阶大礼包，欢迎勾搭。

Thanks for reading!

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

发表于

这是我参与11月更文挑战的第9天，活动详情查看：2021最后一次更文挑战

前两天，通过文章: 《若依中的代码自动生成器研究-表查询篇》以及《若依中的代码生成器-数据库篇》对若依代码生成器的前一段代码的阅读，我们了解了若依代码生成器的一些逻辑，包括通过数据库的information_schema. TABLES查询表信息，以及information_schema. COLUMNS查询指定表的列信息，将其转换到表gen_table与gen_table_column中的数据行，以便后续查询与代码转换。

今天我们继续来看若依系统中是如何自动生成domain代码的。

在系统菜单“系统工具”->”代码生成”中以及可以看到我们导入的表my_user，如下图所示：

image.png

这一行右侧有5个按钮，分别是预览、编辑、删除、同步、生成代码。

预览

我们点击“预览”，在弹出窗口中可以看到可以预览生成的代码包括：domain.java， mapper.java， service.java， serviceImpl.java，controller.java, mapper.xml， sql, api.js, index.vue。

其中domain.java, mapper.java, mapper.xml, sql都是与数据库紧密相关的，domain即生成对应数据库表的类，mapper与sql中则包含数据库基本的增删改查。

我们来研究一下domain的生成逻辑。

通过F12调试，发现点击预览的接口为：/tool/gen/preview/?id， 如下图所示：

image.png

接口代码经过查找，controller中如下：

1
2
3
4
5
6
7
8
9
10
java复制代码/**
     * 预览代码
     */
    @PreAuthorize("@ss.hasPermi('tool:gen:preview')")
    @GetMapping("/preview/{tableId}")
    public AjaxResult preview(@PathVariable("tableId") Long tableId) throws IOException
    {
        Map<String, String> dataMap = genTableService.previewCode(tableId);
        return AjaxResult.success(dataMap);
    }

通过逐层分解，我们找到其中的一些关键代码：

mybatis的collections一对多查询

看过前两篇文章的小伙伴们知道，gen_table中的一行数据对应gen_table_column中的多行数据，那么mybatis是如何查询这种结果的呢？

这里有一个很好的示例。

其关键代码如下所图所示（源代码位于项目ruoyi-generator/resources/GenTableMapper.xml中）

image.png

具体的查询语句如下图所示：

image.png

如此，我们便将数据库中的一对多数据查询并映射为Java中的一个对象GenTable，如此便利于后续的代码生成操作。

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

发表于

这是我参与11月更文挑战的第2天，活动详情查看：2021最后一次更文挑战

数组

一维数组

创建
定义

数组是一组相同类型的元素的集合。那数组的语法形式：

1
2
3
C复制代码type_t arr_name [const_n]
//如：
int arr[10];

type_t 指的是数组元素的类型。

const_n 指的是一个常量表达式，用来指定数组的大小。

此时运行程序的话，系统会报一个警告：未初始化变量。打开调试就会发现系统默认填入一些无意义的数据。

系统默认未初始化值

当然全局数组的话，系统默认初始化为0；

1
2
3
4
C复制代码int arr[10];// 0 0 ... 0
int main(){
  return 0;
}
创建实例
1
2
3
4
5
6
7
8
C复制代码//1.	
int arr[10];
//2.
int count = 10;
int arr2[count];//这样的创建数组可不可以呢？
//3.
float arr3[20];//浮点型数组
char ch[10];

数组的创建必须要[]使用常量，不能使用变量。（ps：虽然C99支持变长数组，但一般用常数创建就已经够用了）同样，我虽然用const_n表示常量，但可千万不要误会为const修饰的变量哦。

为什么呢？

因为数组控制不好容易越界访问非法内存，用变量的话风险太大，所以一直以来都是用常量创建数组的。

初始化

初始化，顾名思义，在创建数组的同时给予一些合理的初识值。如：

1
C复制代码int arr[10] = { 1,2,3 };//不完全初始化

这种是不完全初始化，剩余的元素默认是0

1
C复制代码int arr2[] = { 1,2,3,4 };//利用初始化内容，指定数组大小

这种是省略数组的const_n常量表达式

由初始化内容指定数组的大小

那下面这三个有什么不同呢？

/字符串数组初始化示

第一种是用字符串初始化数组，字符串有\0作为结束标志，虽不算字符串内容，但是可以说是字符串与生俱来的，所以它也被初始化作为数组内容。a b c \0

第二种和第三种是一样的，因为数组元素类型是字符型，且字符'b'的ACSII码值是98，自动将98解析为字符。a b c

使用

数组的访问是通过下标来访问的，默认下标是从0开始。通过下标引用操作符[]我们可以访问到数组元素。

1
2
3
4
5
6
7
8
C复制代码int arr[10] = { 1,2,3,4,5,6,7,8,9,10 };
//数组的下标是从0开始的0~9
int sz = sizeof(arr) / sizeof(arr[0]);//10
for (int i = 0; i < sz; i++)
{
    printf("%d ", arr[i]);
}
//1 2 3 4 5 6 7 8 9 10

对于sizeof操作符，sizeof(arr)，即sizeof+数组名，指的是计算整个数组的大小，算出来是40，然后sizeof(arr[0])是计算数组首元素的大小为4，这样一除就是元素个数啦。

使用变量sz，可以灵活的改变数组的大小，就不用再更改循环条件了。

总结：
  1. 数组是通过下标访问的，下标从0开始
  2. 数组的大小可以通过计算得到
内存存储

在这里插入图片描述

通过printf("&arr[%d]=%p\n", i,&arr[i]);这样的语句我们可以看到该数组在内存中的存储情况。

很明显的是，数组在内存中是连续存放的。

在这里插入图片描述

右边是十六进制的内存编号，可以看见每一个元素之间都相差4个字节，而一个整型元素正好占4个字节。

所以数组在内存中是连续存放的，随着数组下标的增长，地址也在增长，这也正是为什么指针变量+1，可以访问到下一个数组元素。

所以数组的本质是什么？

一组内存中连续存放的相同类型的元素。

二维数组

创建
1
2
3
4
5
c复制代码type_t arr_name[const_n][const_n]
//
int arr[3][5];//3行5列
char ch[4][7];//4行5列
double arr2[2][4]//2行4列

如上述代码所示：二维数组的语法结构就是，类型+数组名+[行][列]。

二维数组描述示意图

如图所示，二维数组在理解上就是这样的3行5列类似于表格的东西。就像线性代数里的矩阵，矩阵的定义就是一组数组成数表。

初始化
1
2
3
4
5
c复制代码//1.
int arr1[3][5] = { 1,2,3,4,5,6,7,8,9,10,11 };

//2.
int arr2[3][5] = { {1,2},{3,4},{5,6,7} };
  1. 第一种初始化，先一行一行填入，第一行是1 2 3 4 5，第二行是6 7 8 9 10，第三行不够就补零11 0 0 0 0 。
  2. 第二种的话，把每一行看成一个一维数组，不够的话还是补零，即第一行1 2 0 0 0 ，第二行3 4 0 0 0 ，第三行5 6 7 0 0 。

二维数组初始化示意

1
2
3
c复制代码char ch1[2][4] = { 'a','b' };
char ch2[2][4] = { {'a'},{'b'} };
char ch3[3][4] = { "abc","def","gh" };

当然用字符串去初始化二维数组的话，也是需要注意\0的问题。

第一行：a b c \0 ；第二行：d e f \0；第三行：g h \0 0

省略
1
c复制代码int arr2[][5] = { {1,2},{3,4},{5,6,7} };

像这样省略行可以，但是不能省略列。

行数可以根据初始化内容来规定，但如果列省略了就会造成歧义。

当然，省略必须在已经初始化的前提之下，不然行和列一概不知，怎么分配空间呢？

使用

当然二维数组同样是用下标访问数组内容的，也是从0开始。如：

二维数组下标示意

我们要去访问这个二维数组的话，我们当然是用两次循环遍历这个数组。

循环遍历二维数组示例

内存存储

当然我们也可以用同样的办法打印出每个元素的地址，如：

二维数组内存存储地址示例

  • 我们还是能发现每一个元素都是在内存中连续存放的。

这样的话，二维数组在内存中的存储形式便是大家想象中的二维的形式，把每一行理解为一个一维数组，这样的话二维数组在内存中的存储形式还是一维的。如下图的对比：

二维数组存储形式对比示意图

  • 从这里我们也可以理解到，二维数组的初始化里，为什么可以省略行不能省略列。

把行省略了，但是我们知道列，一个一个填满就是了，能填到多少行就有多少行。

理解方式
  • 对于二维数组，我们可以理解为每一行为一个元素的一维数组，该一维数组的每一个元素又是一个一维数组。

如数组arr[3][5] ，是有3个元素的一维数组，每个元素是一个有5个元素的一维数组。

指向二维数组的指针+1，指向的是下一行。

对于二维数组在内存存储形式的理解还是很重要的，有了这样的思想，我们就可以通过指针遍历得到数组元素，如：

1
2
3
4
5
6
c复制代码int arr[3][5] = { {1,2,3},{4,5,6},{7,8} };
int* p = &arr[0][0];
for (int i = 0; i < 15; i++)
{
    printf("%d ", *p++);//1 2 3 0 0 4 5 6 0 0 7 8 0 0 0
}

数组越界问题

定义

数组通过下标访问，那么下标也就可以控制数组的访问范围。在数组前后进行访问的话，就是非法访问内存，即数组越界。

1
2
3
4
5
6
c复制代码//1 2 3 4 5 -858993460
int arr[5] = { 1,2,3,4,5 };
for (int i = 0; i <= 5; i++)//越界访问到第6个
{
    printf("%d ", arr[i]);
}

数组越界访问到最后一个元素之后的一块内存，这就属于越界访问，-858993460是vs2019自动生成的随机值。

一般编译器是不会去检查数组越界访问的情况（vs2019太先进），所以我们就要有意识的主动检查。如果编译器提示这样的错误信息，那么一般就是数组越界了：

数组越界访问报错示例

数组作函数参数

在写代码时，我们经常会将数组作为参数，比如接下来的两个应用实例，那么我们这里以冒泡排序的实现作为案例。
在写代码时，我们经常会将数组作为参数，比如接下来的两个应用实例，那么我们这里以冒泡排序的实现作为案例。

排序算法一般有四种：冒泡排序、选择排序、插入排序和快速排序。

冒泡排序的核心思想：两两相邻的元素进行比较。

  • 一趟冒泡排序搞定一个数字，让其来到最终的位置上。
  • nnn 个元素，则总共需要 n−1n-1n−1 趟冒泡排序，每一趟排序需要进行 n−1−in-1-in−1−i 次判断大小。如分析图所示：
    冒泡排序示意图
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
c复制代码void Print(int* arr, int sz)
{
	for (int i = 0; i < sz; i++)
	{
		printf("%d ", *arr++);
	}
}
void Sort(int arr[],int sz)//int* arr
//数组作形参本质是指针
{
	//int sz = sizeof(arr) / sizeof(arr[0]);//err//用指针的sizeof值除以另一个值 = 4 / 4 = 1
	for (int i = 0; i < sz - 1; i++)//n-1趟
	{
		for (int j = 0; j < sz - 1 - i; j++)//n-1-i次
		{
			if (arr[j] > arr[j + 1])//目标升序
			{
				//交换
				int tmp = arr[j];
				arr[j] = arr[j + 1];
				arr[j + 1] = tmp;
			}
		}
	}
}
int main()
{
	int arr[] = { 1,4,6,3,7,9,3,2,8,5 };
	int sz = sizeof(arr) / sizeof(arr[0]);
	//数组名单独放在sizeof中，表示整个数组
	//排序
	Sort(arr,sz);
	//打印
	Print(arr,sz);
	return 0;
}
  1. 定义数组作形参时，本质上是指针。

void Sort(int *arr,int sz)本质上就是void Sort(int arr[],int sz)

所以Sort()函数内，sizeof(arr)也算的就是指针arr的大小，所以只能传参进去。

  1. 数组名arr何时代表整个数组何时代表数组首元素地址呢？
* 代表整个数组的情况：



> 单独放在`sizeof`操作符内部时，如`sizeof(arr);`  。
> 
> 
> 写出`&arr`时，代表的是整个数组，但表面仍为首元素地址。
* 代表首元素地址的情况：



> 除上面两以外其他都是代表首元素的地址。

应用实例

笔者实在没时间写两个应用实例的博客，所以在此将思维导图奉上，一般照着思维导图写就没问题了。感谢李姐和支持~

数组的应用实例1：三子棋

三子棋脑图

数组的应用实例2：扫雷游戏

扫雷脑图

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

发表于

「这是我参与11月更文挑战的第4天，活动详情查看：2021最后一次更文挑战」

那天，小二去马蜂窝面试，面试官老王一上来就甩给了他一道面试题：请问Comparable和Comparator有什么区别？小二差点笑出声，因为三年前，也就是 2021 年，他在《Java 程序员进阶之路》专栏上看到过这题😆。

PS：为了能够帮助更多的 Java 初学者，已将《Java 程序员进阶之路》开源到了 GitHub（本篇已收录）。该专栏目前已经收获了 580 枚星标，如果你也喜欢这个专栏，觉得有帮助的话，可以去点个 star，这样也方便以后进行更系统化的学习！

GitHub 地址：github.com/itwanger/to…

码云地址：沉默王二/toBeBetterJavaer

Comparable 和 Comparator 是 Java 的两个接口，从名字上我们就能够读出来它们俩的相似性：以某种方式来比较两个对象。但它们之间到底有什么区别呢？请随我来，打怪进阶喽！

01、Comparable

Comparable 接口的定义非常简单，源码如下所示。

1
2
3
java复制代码public interface Comparable<T> {
    int compareTo(T t);
}

如果一个类实现了 Comparable 接口（只需要干一件事，重写 compareTo() 方法），就可以按照自己制定的规则将由它创建的对象进行比较。下面给出一个例子。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
java复制代码public class Cmower implements Comparable<Cmower> {
    private int age;
    private String name;

    public Cmower(int age, String name) {
        this.age = age;
        this.name = name;
    }

    @Override
    public int compareTo(Cmower o) {
        return this.getAge() - o.getAge();
    }

    public static void main(String[] args) {
        Cmower wanger = new Cmower(19,"沉默王二");
        Cmower wangsan = new Cmower(16,"沉默王三");

        if (wanger.compareTo(wangsan) < 0) {
            System.out.println(wanger.getName() + "比较年轻有为");
        } else {
            System.out.println(wangsan.getName() + "比较年轻有为");
        }
    }
}

在上面的示例中，我创建了一个 Cmower 类，它有两个字段：age 和 name。Cmower 类实现了 Comparable 接口，并重写了 compareTo() 方法。

程序输出的结果是“沉默王三比较年轻有为”，因为他比沉默王二小三岁。这个结果有什么凭证吗？

凭证就在于 compareTo() 方法，该方法的返回值可能为负数，零或者正数，代表的意思是该对象按照排序的规则小于、等于或者大于要比较的对象。如果指定对象的类型与此对象不能进行比较，则引发 ClassCastException 异常（自从有了泛型，这种情况就少有发生了）。

02、Comparator

Comparator 接口的定义相比较于 Comparable 就复杂的多了，不过，核心的方法只有两个，来看一下源码。

1
2
3
4
java复制代码public interface Comparator<T> {
    int compare(T o1, T o2);
    boolean equals(Object obj);
}

第一个方法 compare(T o1, T o2) 的返回值可能为负数，零或者正数，代表的意思是第一个对象小于、等于或者大于第二个对象。

第二个方法 equals(Object obj) 需要传入一个 Object 作为参数，并判断该 Object 是否和 Comparator 保持一致。

有时候，我们想让类保持它的原貌，不想主动实现 Comparable 接口，但我们又需要它们之间进行比较，该怎么办呢？

Comparator 就派上用场了，来看一下示例。

1）原封不动的 Cmower 类。

1
2
3
4
5
6
7
8
9
java复制代码public class Cmower  {
    private int age;
    private String name;

    public Cmower(int age, String name) {
        this.age = age;
        this.name = name;
    }
}

（说好原封不动，getter/setter 吃了啊）

Cmower 类有两个字段：age 和 name，意味着该类可以按照 age 或者 name 进行排序。

2）再来看 Comparator 接口的实现类。

1
2
3
4
5
6
java复制代码public class CmowerComparator implements Comparator<Cmower> {
    @Override
    public int compare(Cmower o1, Cmower o2) {
        return o1.getAge() - o2.getAge();
    }
}

按照 age 进行比较。当然也可以再实现一个比较器，按照 name 进行自然排序，示例如下。

1
2
3
4
5
6
7
8
9
10
11
java复制代码public class CmowerNameComparator implements Comparator<Cmower> {
    @Override
    public int compare(Cmower o1, Cmower o2) {
        if (o1.getName().hashCode() < o2.getName().hashCode()) {
            return -1;
        } else if (o1.getName().hashCode() == o2.getName().hashCode()) {
            return 0;
        }
        return 1;
    }
}

3）再来看测试类。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
java复制代码Cmower wanger = new Cmower(19,"沉默王二");
Cmower wangsan = new Cmower(16,"沉默王三");
Cmower wangyi = new Cmower(28,"沉默王一");

List<Cmower> list = new ArrayList<>();
list.add(wanger);
list.add(wangsan);
list.add(wangyi);

list.sort(new CmowerComparator());

for (Cmower c : list) {
    System.out.println(c.getName());
}

创建了三个对象，age 不同，name 不同，并把它们加入到了 List 当中。然后使用 List 的 sort() 方法进行排序，来看一下输出的结果。

1
2
3
复制代码沉默王三
沉默王二
沉默王一

这意味着沉默王三的年纪比沉默王二小，排在第一位；沉默王一的年纪比沉默王二大，排在第三位。和我们的预期完全符合。

03、到底该用哪一个呢？

通过上面的两个例子可以比较出 Comparable 和 Comparator 两者之间的区别：

  • 一个类实现了 Comparable 接口，意味着该类的对象可以直接进行比较（排序），但比较（排序）的方式只有一种，很单一。
  • 一个类如果想要保持原样，又需要进行不同方式的比较（排序），就可以定制比较器（实现 Comparator 接口）。
  • Comparable 接口在 java.lang 包下，而 Comparator 接口在 java.util 包下，算不上是亲兄弟，但可以称得上是表（堂）兄弟。

举个不恰当的例子。我想从洛阳出发去北京看长城，体验一下好汉的感觉，要么坐飞机，要么坐高铁；但如果是孙悟空的话，翻个筋斗就到了。我和孙悟空之间有什么区别呢？孙悟空自己实现了 Comparable 接口（他那年代也没有飞机和高铁，没得选），而我可以借助 Comparator 接口（现代化的交通工具）。

好了，关于 Comparable 和 Comparator 我们就先聊这么多。总而言之，如果对象的排序需要基于自然顺序，请选择 Comparable，如果需要按照对象的不同属性进行排序，请选择 Comparator。

这是《Java 程序员进阶之路》专栏的第 67 篇。Java 程序员进阶之路，该专栏风趣幽默、通俗易懂，对 Java 初学者极度友好和舒适😘，内容包括但不限于 Java 语法、Java 集合框架、Java IO、Java 并发编程、Java 虚拟机等核心知识点。

GitHub 地址：github.com/itwanger/to…

码云地址：沉默王二/toBeBetterJavaer

亮白版和暗黑版的 PDF 也准备好了呢，让我们一起成为更好的 Java 工程师吧，一起冲！

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

发表于

这是我参与11月更文挑战的第2天，活动详情查看：2021最后一次更文挑战

👉速看

  • 最近跟着B站视频看完了redis的入门，对于redis的纯理论知识有了初步的了解，本专栏会先更新入门篇，之后还会有项目实战篇(应用在小程序上或者其他的项目上边)，欢迎大家关注本专栏一起学习!

💫参照课程大纲

image.png

本专栏将结合尚硅谷和狂神说两个课程，基本涵盖了包括但不限于图中的知识点。

NoSQL

NoSQL(NoSQL = Not Only SQL )，意即“不仅仅是SQL”，泛指非关系型的数据库。
NoSQL 不依赖业务逻辑方式存储，而以简单的key-value模式存储。因此大大的增加了数据库的扩展能力。

  • 不遵循SQL标准。
  • 不支持ACID。
  • 远超于SQL的性能。

适用场景

  • 对数据高并发的读写
  • 海量数据的读写
  • 对数据高可扩展性的

不适用场景

  • 需要事务支持
  • 基于sql的结构化查询存储，处理复杂的关系,需要即席查询。
  • 用不着sql的和用了sql也不行的情况，请考虑用NoSql

概述

Redis（Remote Dictionary Server )，即远程字典服务 !
是一个开源的使用ANSI C语言编写、支持网络、可基于内存亦可持久化的日志型、Key-Value数据库，并提供多种语言的API。

  • redis会周期性的把更新的数据写入磁盘或者把修改操作写入追加的记录文件，并且在此基础上实现了master-slave(主从)同步。

用途

1、内存存储、持久化，内存中是断电即失、所以说持久化很重要（rdb、aof）
2、效率高，可以用于高速缓存
3、发布订阅系统
4、地图信息分析
5、计时器、计数器（浏览量！）
image.png

特性

1、多样的数据类型

2、持久化

3、集群

4、事务

历史发展

一开始数据量很少，只需要单表处理读和写

90年代，一个基本的网站访问量一般不会太大，单个数据库完全足够！

那个时候，更多的去使用静态网页 Html ~ 服务器根本没有太大的压力！

思考一下，这种情况下：整个网站的瓶颈是什么？

1、数据量如果太大、一个机器放不下了！

2、数据的索引 （B+ Tree），一个机器内存也放不下

3、访问量（读写混合），一个服务器承受不了

Memcached（缓存）+ MySQL + 垂直拆分 （读写分离）

可以一台服务器负责写,如何同步给前台服务器去读?

缓存也解决读的问题

网站80%的情况都是在读，每次都要去查询数据库的话就十分的麻烦！所以说我们希望减轻数据的压力，我们可以使用缓存来保证效率！

发展过程： 优化数据结构和索引–> 文件缓存（IO）—> Memcached（当时最热门的技术！）

分库分表，MySQL集群 + 水平拆分

阅读量也不是实时写到mysql的,是先到缓存,再一定时间统一写进去

早些年MyISAM： 表锁，十分影响效率！高并发下就会出现严重的锁问题

转战Innodb：行锁

慢慢的就开始使用分库分表来解决写的压力！ MySQL 在哪个年代推出了表分区！这个并没有多少公司使用！

MySQL 的集群，很好满足那个年代的所有需求！

Linux版安装

1、官网下载安装包！

2、解压Redis的安装包到你要的目录！

tar -zxvf

3、基本环境安装

yum install gcc-c++

4、进入解压后的目录

make

4、进入src目录

make install

  • 此处如果有问题的话可以自行搜索相关安装教程，根据自己需求下载window版或者linux版

Redis运行

先设置daemonize no改成yes，开启后台启动

修改redis.conf(128行)文件将里面的daemonize no 改成yes，让服务在后台启动

protected-mode 设置成no

默认是设置成yes的， 防止了远程访问，在redis3.2.3版本后

找到# requirepass foobared

删除前面的注释符号#，并把foobared修改成自己的密码

或者另起一行 requirepass 自己的密码

  • 一定要设置密码阿，之前被黑客通过redis黑进服务器了，植入了挖抗病毒麻了🙊(详情见沸点)

要注意是在启动server的时候选择conf!!!!!

./redis-server ../redis.conf

此处要选择conf来启动

./redis-cli -p 端口号

服务器连接

进入后 auth 你设置的密码

验证通过后才能成功操作

Redis关闭与退出

SHUTDOWN

  • 然后exit

image.png

性能分析

redis-benchmark 是一个压力测试工具！
官方自带的性能测试工具！
redis-benchmark 命令参数！
image.png
我们来简单测试下：
image.png
如何查看这些分析呢？
image.png

基础知识

默认16个数据库,可以用select切换

image.png
Redis是很快的，官方表示，Redis是基于内存操作，CPU不是Redis性能瓶颈，Redis的瓶颈是根据
机器的内存和网络带宽，既然可以使用单线程来实现，就使用单线程了！

Redis 是C 语言写的，官方提供的数据为 100000+ 的QPS，完全不比同样是使用 key-vale的Memecache差！

Redis 为什么单线程还这么快？

1、误区1：高性能的服务器一定是多线程的？

2、误区2：多线程（CPU上下文会切换!）一定比单线程效率高！

核心：redis 是将所有的数据全部放在内存中的，所以说使用单线程去操作效率就是最高的，多线程(CPU上下文会切换：耗时的操作！！！) 对于内存系统来说，如果没有上下文切换效率就是最高的！多次读写都是在一个CPU上的，在内存情况下，这个就是最佳的方案！

💠下篇预告

  • 下篇我们将介绍redis中的常见数据类型和相关的常用命令，之后还会解析一下redis的配置文件等相关知识。

参考

  • 尚硅谷Redis6视频
  • 狂神说Redis视频

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

0%