简栈

Feign简介

​ 在上一篇文章中分析了Eureka的注册、续约、服务剔除、服务自我保护等机制，地址在https://blog.csdn.net/lgq2626/article/details/80288992 。这篇分析SpringCloud的feign。SpringCloud微服务项目之间调用是通过httprest请求来进行服务调用的，之前我们会用到HttpClient等工具来进行服务请求，Spring对这种请求进行了处理，封装成了可声明式的web客户端，使得编写web客户端更容易，feign还支持可插拔的编码器和解码器，Spring在用的时候增加了对@requestMapping的处理，同时，SpringCloud还对feign集成了注册中心（eureka）和客户端负载均衡（ribbon），使得我们拥有一个客户端负载均衡的web请求客户端。

Feign在项目中的配置和使用

​ 在Springcloud中使用feign的时候，需要在配置类中加入一个@EnableFeignClients注解。代码如下：

1
2
3
4
5
6
7
8

@SpringBootApplication//springboot 启动类
@EnableFeignClients//开启eureka扫描
@EnableDiscoveryClient//开启eureka客户端
public class Application {
    public static void main( String[] args ) throws ClassNotFoundException {
        SpringApplication.run(Application.class, args);
    }
}

配置feign调用客户端

1
2
3
4
5

@FeignClient(value = "xxx-server",configuration = FeignConfiguration.class)
public interface ConsumerSmsService extends SMSService{
    @RequestMapping(value = "/sms/smsMessage", method = RequestMethod.POST)
    RespSMSDto sendSms(ReqSMSDto smsReqDto);
}

经过上面的配置，直接在项目里面注入容器调用接口就可以了。

Feign源码分析

​ 在@EnableFeignClients标签中，import了一个FeignClientsRegistrar类，那么这个FeignClientsRegistrar#registerBeanDefinitions()在什么时候调用的呢？跟着Spring的源码走下去，看过源码的人都会直接看到AbstractApplicationContext#refresh()方法，整体整理一下代码：

前文已经讲过，CMS是老年代垃圾收集器，在收集过程中可以与用户线程并发操作。它可以与Serial收集器和Parallel New收集器搭配使用。CMS牺牲了系统的吞吐量来追求收集速度，适合追求垃圾收集速度的服务器上。

CMS相关参数

触发条件

周期性GC

由后台线程ConcurrentMarkSweepThread循环判断（默认2s）是否需要触发。

如果没有设置-XX:+UseCMSInitiatingOccupancyOnly，虚拟机会根据收集的数据决定是否触发
老年代使用率达到阈值 CMSInitiatingOccupancyFraction，默认92%。
永久代的使用率达到阈值 CMSInitiatingPermOccupancyFraction，默认92%，前提是开启 CMSClassUnloadingEnabled。
新生代的晋升担保失败。

主动触发

YGC过程发生Promotion Failed，进而对老年代进行回收
比如执行了System.gc()，前提是没有参数ExplicitGCInvokesConcurrent

收集过程

前言

线程池ThreadPoolExecutor在运行的过程中，业务并发量变动，需要不停服务调整线程池的线程数，ThreadPoolExecutor支持动态调整corePoolSize与maximumPoolSize的值。

示例demo

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

public class ThreadChangeTest {
    public static void main(String[] args) throws InterruptedException {
        ThreadPoolExecutor executor = new ThreadPoolExecutor(3,
                10,
                10l, TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(10));
		int count = 0;
        while (true) {
            Thread.sleep(1000l);
            for (int i = 0; i < 9; i++) {
                executor.execute(() -> {
                    /*try {
                        Thread.sleep(1l);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }*/
                    System.out.println("------------core:\t" + executor.getCorePoolSize() + "\tactive:\t" + executor.getActiveCount() + "\tmax:\t" + executor.getMaximumPoolSize());
                });
            }

            count++;
            if (count == 20) {
                executor.setCorePoolSize(2);
                executor.setMaximumPoolSize(9);
                System.out.println("----------------------------------------");
            }

            if (count == 100) {
                executor.shutdown();
                System.out.println("=============================================");
                break;
            }
        }
        Thread.currentThread().join();
    }
}

在程序运行中动态修改线程池corePoolSize与maximumPoolSize的值

源码分析

线程池参数调大

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

public void setCorePoolSize(int corePoolSize) {
    if (corePoolSize < 0)
        throw new IllegalArgumentException();
    int delta = corePoolSize - this.corePoolSize;
    this.corePoolSize = corePoolSize;
    //核心线程调小，中断空闲任务，否则线程池的当前任务结束，自动调小
    if (workerCountOf(ctl.get()) > corePoolSize)
        interruptIdleWorkers();
    //核心线程数调大后，从队列取任务
    else if (delta > 0) {
        // We don't really know how many new threads are "needed".
        // As a heuristic, prestart enough new workers (up to new
        // core size) to handle the current number of tasks in
        // queue, but stop if queue becomes empty while doing so.
        //队列大小是否可以取任务
        int k = Math.min(delta, workQueue.size());
        //队列有任务就取，否则break
        while (k-- > 0 && addWorker(null, true)) {
            if (workQueue.isEmpty())
                break;
        }
    }
}
 
public void setMaximumPoolSize(int maximumPoolSize) {
    if (maximumPoolSize <= 0 || maximumPoolSize < corePoolSize)
        throw new IllegalArgumentException();
    this.maximumPoolSize = maximumPoolSize;
    //中断空闲任务，否则线程池的当前任务结束，自动调小
    if (workerCountOf(ctl.get()) > maximumPoolSize)
        interruptIdleWorkers();
}

源码看出：线程池的调节时直接设置corePoolSize与maximumPoolSize的值

其中

在数据仓库的建设中，一般都会围绕着星型模型和雪花模型来设计表关系或者结构。下面我们先来理解这两种模型的概念。

星型模型

​ 星型模是一种多维的数据关系，它由一个事实表和一组维表组成。每个维表都有一个维作为主键，所有这些维的主键组合成事实表的主键。强调的是对维度进行预处理，将多个维度集合到一个事实表，形成一个宽表。这也是我们在使用hive时，经常会看到一些大宽表的原因，大宽表一般都是事实表，包含了维度关联的主键和一些度量信息，而维度表则是事实表里面维度的具体信息，使用时候一般通过join来组合数据，相对来说对OLAP的分析比较方便。

雪花模型

​ 当有一个或多个维表没有直接连接到事实表上，而是通过其他维表连接到事实表上时，其图解就像多个雪花连接在一起，故称雪花模型。雪花模型是对星型模型的扩展。它对星型模型的维表进一步层次化，原有的各维表可能被扩展为小的事实表，形成一些局部的 “层次 “ 区域，这些被分解的表都连接到主维度表而不是事实表。雪花模型更加符合数据库范式，减少数据冗余，但是在分析数据的时候，操作比较复杂，需要join的表比较多所以其性能并不一定比星型模型高。

星型模型和雪花模型对比

应用场景

聚集（clustered）索引，也叫聚簇索引

定义：数据行的物理顺序与列值（一般是主键的那一列）的逻辑顺序相同，一个表中只能拥有一个聚集索引。

注：第一列的地址表示该行数据在磁盘中的物理地址，后面三列才是我们SQL里面用的表里的列，其中id是主键，建立了聚集索引。

结合上面的表格就可以理解这句话了吧：数据行的物理顺序与列值的顺序相同，如果我们查询id比较靠后的数据，那么这行数据的地址在磁盘中的物理地址也会比较靠后。而且由于物理排列方式与聚集索引的顺序相同，所以也就只能建立一个聚集索引了。

从上图可以看出聚集索引的好处了，索引的叶子节点就是对应的数据节点（MySQL的MyISAM除外，此存储引擎的聚集索引和非聚集索引只多了个唯一约束，其他没什么区别），可以直接获取到对应的全部列的数据，而非聚集索引在索引没有覆盖到对应的列的时候需要进行二次查询，后面会详细讲。因此在查询方面，聚集索引的速度往往会更占优势。

创建聚集索引

如果不创建索引，系统会自动创建一个隐含列作为表的聚集索引。

• 创建表的时候指定主键（注意：SQL Sever默认主键为聚集索引，也可以指定为非聚集索引，而MySQL里主键就是聚集索引）

前提

近段时间，业务系统架构基本完备，数据层面的建设比较薄弱，因为笔者目前工作重心在于搭建一个小型的数据平台。优先级比较高的一个任务就是需要近实时同步业务系统的数据（包括保存、更新或者软删除）到一个另一个数据源，持久化之前需要清洗数据并且构建一个相对合理的便于后续业务数据统计、标签系统构建等扩展功能的数据模型。基于当前团队的资源和能力，优先调研了Alibaba开源中间件Canal的使用。

这篇文章简单介绍一下如何快速地搭建一套Canal相关的组件。

关于Canal

简介

下面的简介和下一节的原理均来自于Canal项目的README：

Canal[kə'næl]，译意为水道/管道/沟渠，主要用途是基于MySQL数据库增量日志解析，提供增量数据订阅和消费。早期阿里巴巴因为杭州和美国双机房部署，存在跨机房同步的业务需求，实现方式主要是基于业务trigger获取增量变更。从 2010 年开始，业务逐步尝试数据库日志解析获取增量变更进行同步，由此衍生出了大量的数据库增量订阅和消费业务。

基于日志增量订阅和消费的业务包括：

1. Kafka简介

Kafka是一种分布式的，基于发布/订阅的消息系统。主要设计目标如下：

1. 以时间复杂度为O(1)的方式提供消息持久化能力，即使对TB级以上数据也能保证常数时间复杂度的访问性能
2. 高吞吐率。即使在非常廉价的商用机器上也能做到单机支持每秒100K条以上消息的传输
3. 支持Kafka Server间的消息分区，及分布式消费，同时保证每个Partition内的消息顺序传输
4. 同时支持离线数据处理和实时数据处理
5. Scale out：支持在线水平扩展

2. Kafka架构

Broker：Kafka集群包含一个或多个服务器，这种服务器被称为broker

Topic：每条发布到Kafka集群的消息都有一个类别，这个类别被称为Topic。（物理上不同Topic的消息分开存储，逻辑上一个Topic的消息虽然保存于一个或多个broker上但用户只需指定消息的Topic即可生产或消费数据而不必关心数据存于何处）

Partition：Parition是物理上的概念，每个Topic包含一个或多个Partition.

Producer：负责发布消息到Kafka broker

Consumer：消息消费者，向Kafka broker读取消息的客户端。

Consumer Group：每个Consumer属于一个特定的Consumer Group（可为每个Consumer指定group name，若不指定group name则属于默认的group）。

背景

为了后续的沟通方便，在20200213的早上创建了一个**《简栈-Java技术交流群》**，也方便大家通过扫二维码积极的参与进来。

如果是二维码已经过期，大家可以添加简栈文化-小助手的微信号（lastpass4u），然后让他拉大家进群进群。我们保持着小而美的精神，宁缺毋滥。

然后早上群里就有人提了一个问题：

执行计划里面的扫描函数跟执行时间不匹配，比如查询优化器发现，扫描a索引行数更多，所以更慢，因此优化器选择了索引b, 但实际上走b索引的时候比a更慢，走a索引大概是4秒左右，b是8秒。

这个问题激发起了大家的讨论，有的人建议说：

1、这种可以强制指定索引执行的吧

2、这个扫描行数都是预估的不一定准的，能操作shell的话执行analyse table看看。

3、看一下你的index，DDL，explain等等

但提问者明显这些都是已经自己搞清楚了的，他关心的是底层的优化器成本规则等。这类我才意识到EXPLAIN出来的是结果，其实数据库底层本身是有优化器的，而最终选择谁，是否过索引等都是有它的规则的。这其中都涉及到效率与成本问题。

1. 背景

1.1. 惊人的性能数据

最近一个圈内朋友通过私信告诉我，通过使用 Netty4 + Thrift 压缩二进制编解码技术，他们实现了 10W TPS（1K 的复杂 POJO 对象）的跨节点远程服务调用。相比于传统基于 Java 序列化 +BIO（同步阻塞 IO）的通信框架，性能提升了 8 倍多。

事实上，我对这个数据并不感到惊讶，根据我 5 年多的 NIO 编程经验，通过选择合适的 NIO 框架，加上高性能的压缩二进制编解码技术，精心的设计 Reactor 线程模型，达到上述性能指标是完全有可能的。

下面我们就一起来看下 Netty 是如何支持 10W TPS 的跨节点远程服务调用的，在正式开始讲解之前，我们先简单介绍下 Netty。

1.2. Netty 基础入门

Netty 是一个高性能、异步事件驱动的 NIO 框架，它提供了对 TCP、UDP 和文件传输的支持，作为一个异步 NIO 框架，Netty 的所有 IO 操作都是异步非阻塞的，通过 Future-Listener 机制，用户可以方便的主动获取或者通过通知机制获得 IO 操作结果。

作为当前最流行的 NIO 框架，Netty 在互联网领域、大数据分布式计算领域、游戏行业、通信行业等获得了广泛的应用，一些业界著名的开源组件也基于 Netty 的 NIO 框架构建。

2. Netty 高性能之道

2.1. RPC 调用的性能模型分析

作者：吃饭睡觉撸代码

来源：https://fangjian0423.github.io/2017/05/10/springboot-context-refresh/

前言

Spring容器创建之后，会调用它的refresh方法，refresh的时候会做很多事情：比如完成配置类的解析、各种BeanFactoryPostProcessor和BeanPostProcessor的注册、国际化配置的初始化、web内置容器的构造等等。

我们来分析一下这个refresh过程。

还是以web程序为例，那么对应的Spring容器为AnnotationConfigEmbeddedWebApplicationContext。它的refresh方法调用了父类AbstractApplicationContext的refresh方法：

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

public void refresh() throws BeansException, IllegalStateException {
  // refresh过程只能一个线程处理，不允许并发执行
  synchronized (this.startupShutdownMonitor) {
    prepareRefresh();
    ConfigurableListableBeanFactory beanFactory = obtainFreshBeanFactory();
    prepareBeanFactory(beanFactory);
    try {
      postProcessBeanFactory(beanFactory);
      invokeBeanFactoryPostProcessors(beanFactory);
      registerBeanPostProcessors(beanFactory);
      initMessageSource();
      initApplicationEventMulticaster();
      onRefresh();
      registerListeners();
      finishBeanFactoryInitialization(beanFactory);
      finishRefresh();
    }
    catch (BeansException ex) {
      if (logger.isWarnEnabled()) {
        logger.warn("Exception encountered during context initialization - " +
            "cancelling refresh attempt: " + ex);
      }
      destroyBeans();
      cancelRefresh(ex);
      throw ex;
    }
    finally {
      resetCommonCaches();
    }
  }
}

prepareRefresh方法

表示在真正做refresh操作之前需要准备做的事情：

1. 设置Spring容器的启动时间，撤销关闭状态，开启活跃状态。
2. 初始化属性源信息(Property)
3. 验证环境信息里一些必须存在的属性