深入理解 RPC 之序列化篇 --Kryo

原创
RPC

一年前，笔者刚刚接触 RPC 框架，从单体式应用向分布式应用的变革无疑是让人兴奋的，同时也对 RPC 背后到底做了哪些工作产生了兴趣，但其底层的设计对新手而言并不是很友好，其涉及的一些常用技术点都有一定的门槛。如传输层常常使用的 netty，之前完全没听过，想要学习它，需要掌握前置知识点 nio；协议层，包括了很多自定义的协议，而每个 RPC 框架的实现都有差异；代理层的动态代理技术，如 jdk 动态代理，虽然实战经验不多，但至少还算会用，而 cglib 则又有一个盲区；序列化层倒还算是众多层次中相对简单的一环，但 RPC 为了追求可扩展性，性能等诸多因素，通常会支持多种序列化方式以供使用者插拔使用，一些常用的序列化方案 hessian，kryo，Protobuf 又得熟知…

这个系列打算就 RPC 框架涉及到的一些知识点进行探讨，本篇先从序列化层的一种选择 –kryo 开始进行介绍。

序列化概述

大白话介绍下 RPC 中序列化的概念，可以简单理解为对象 –> 字节的过程，同理，反序列化则是相反的过程。为什么需要序列化？因为网络传输只认字节。所以互信的过程依赖于序列化。有人会问，FastJson 转换成字符串算不算序列化？对象持久化到数据库算不算序列化？没必要较真，广义上理解即可。

JDK 序列化

可能你没用过 kryo，没用过 hessian，但你一定用过 jdk 序列化。我最早接触 jdk 序列化，是在大二的 JAVA 大作业中，《XX 管理系统》需要把对象保存到文件中（那时还没学数据库），jdk 原生支持的序列化方式用起来也很方便。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
class Student implements Serializable{  
   private String name;  
}  
class Main{
   public static void main(String[] args) throws Exception{  
      // create a Student
      Student st = new Student("kirito");  
     // serialize the st to student.db file  
     ObjectOutputStream oos = new ObjectOutputStream(new FileOutputStream("student.db"));  
     oos.writeObject(st);  
     oos.close();  
     // deserialize the object from student.db
     ObjectInputStream ois = new ObjectInputStream(new FileInputStream("student.db"));  
     Student kirito = (Student) ois.readObject();  
     ois.close();  
    // assert
    assert "kirito".equals(kirito.getName());  
   }  
}

Student 实体类需要实现 Serializable 接口，以告知其可被序列化。

序列化协议的选择通常有下列一些常用的指标：

  1. 通用性。是否只能用于 java 间序列化 / 反序列化，是否跨语言，跨平台。
  2. 性能。分为空间开销和时间开销。序列化后的数据一般用于存储或网络传输，其大小是很重要的一个参数；解析的时间也影响了序列化协议的选择，如今的系统都在追求极致的性能。
  3. 可扩展性。系统升级不可避免，某一实体的属性变更，会不会导致反序列化异常，也应该纳入序列化协议的考量范围。
  4. 易用性。API 使用是否复杂，会影响开发效率。

容易用的模型通常性能不好，性能好的模型通常用起来都比较麻烦。显然，JDK 序列化属于前者。我们不过多介绍它，直接引入今天的主角 kryo 作为它的替代品。

Kryo 入门

引入依赖

1
2
3
4
5
<dependency>
    <groupId>com.esotericsoftware</groupId>
    <artifactId>kryo</artifactId>
    <version>4.0.1</version>
</dependency>

由于其底层依赖于 ASM 技术，与 Spring 等框架可能会发生 ASM 依赖的版本冲突（文档中表示这个冲突还挺容易出现）所以提供了另外一个依赖以供解决此问题

1
2
3
4
5
<dependency>
    <groupId>com.esotericsoftware</groupId>
    <artifactId>kryo-shaded</artifactId>
    <version>4.0.1</version>
</dependency>

快速入门

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
class Student implements Serializable{  
   private String name;  
}  
public class Main {
    public static void main(String[] args) throws Exception{
        Kryo kryo = new Kryo();
        Output output = new Output(new FileOutputStream("student.db"));
        Student kirito = new Student("kirito");
        kryo.writeObject(output, kirito);
        output.close();
        Input input = new Input(new FileInputStream("student.db"));
        Student kiritoBak = kryo.readObject(input, Student.class);
        input.close();
        assert "kirito".equals(kiritoBak.getName());
    }
}

不需要注释也能理解它的执行流程，和 jdk 序列化差距并不是很大。

三种读写方式

Kryo 共支持三种读写方式

  1. 如果知道 class 字节码，并且对象不为空
1
2
3
kryo.writeObject(output, someObject);
// ...
SomeClass someObject = kryo.readObject(input, SomeClass.class);

快速入门中的序列化 / 反序列化的方式便是这一种。而 Kryo 考虑到 someObject 可能为 null，也会导致返回的结果为 null，所以提供了第二套读写方式。

  1. 如果知道 class 字节码，并且对象可能为空
1
2
3
kryo.writeObjectOrNull(output, someObject);
// ...
SomeClass someObject = kryo.readObjectOrNull(input, SomeClass.class);

但这两种方法似乎都不能满足我们的需求，在 RPC 调用中，序列化和反序列化分布在不同的端点，对象的类型确定，我们不想依赖于手动指定参数，最好是…emmmmm… 将字节码的信息直接存放到序列化结果中，在反序列化时自行读取字节码信息。Kryo 考虑到了这一点，于是提供了第三种方式。

  1. 如果实现类的字节码未知，并且对象可能为 null
1
2
3
4
5
6
kryo.writeClassAndObject(output, object);
// ...
Object object = kryo.readClassAndObject(input);
if (object instanceof SomeClass) {
   // ...
}

我们牺牲了一些空间一些性能去存放字节码信息，但这种方式是我们在 RPC 中应当使用的方式。

我们关心的问题

继续介绍 Kryo 特性之前，不妨让我们先思考一下，一个序列化工具或者一个序列化协议，应当需要考虑哪些问题。比如，支持哪些类型的序列化？循环引用会不会出现问题？在某个类增删字段之后反序列化会报错吗？等等等等….

带着我们考虑到的这些疑惑，以及我们暂时没考虑到的，但 Kryo 帮我们考虑到的，来看看 Kryo 到底支持哪些特性。

支持的序列化类型

boolean Boolean byte Byte char
Character short Short int Integer
long Long float Float double
Double byte[] String BigInteger BigDecimal
Collection Date Collections.emptyList Collections.singleton Map
StringBuilder TreeMap Collections.emptyMap Collections.emptySet KryoSerializable
StringBuffer Class Collections.singletonList Collections.singletonMap Currency
Calendar TimeZone Enum EnumSet

表格中支持的类型一览无余，这都是其默认支持的。

1
2
Kryo kryo = new Kryo();
kryo.addDefaultSerializer(SomeClass.class, SomeSerializer.class);

这样的方式，也可以为一个 Kryo 实例扩展序列化器。

总体而言，Kryo 支持以下的类型：

  • 枚举
  • 集合、数组
  • 子类 / 多态
  • 循环引用
  • 内部类
  • 泛型

但需要注意的是，Kryo 不支持 Bean 中增删字段 。如果使用 Kryo 序列化了一个类，存入了 Redis，对类进行了修改，会导致反序列化的异常。

另外需要注意的一点是使用反射创建的一些类序列化的支持。如使用 Arrays.asList(); 创建的 List 对象，会引起序列化异常。

1
Exception in thread "main" com.esotericsoftware.kryo.KryoException: Class cannot be created (missing no-arg constructor): java.util.Arrays$ArrayList

但 new ArrayList() 创建的 List 对象则不会，使用时需要注意，可以使用第三方库对 Kryo 进行序列化类型的扩展。如 https://github.com/magro/kryo-serializers 所提供的。

不支持包含无参构造器类的反序列化 ，尝试反序列化一个不包含无参构造器的类将会得到以下的异常：

1
Exception in thread "main" com.esotericsoftware.kryo.KryoException: Class cannot be created (missing no-arg constructor): moe.cnkirito.Xxx

保证每个类具有无参构造器是应当遵守的编程规范，但实际开发中一些第三库的相关类不包含无参构造，的确是有点麻烦。

线程安全

Kryo 是线程不安全的，意味着每当需要序列化和反序列化时都需要实例化一次，或者借助 ThreadLocal 来维护以保证其线程安全。

1
2
3
4
5
6
7
8
9
10
11
private static final ThreadLocal<Kryo> kryos = new ThreadLocal<Kryo>() {
	protected Kryo initialValue() {
		Kryo kryo = new Kryo();
		// configure kryo instance, customize settings
		return kryo;
	};
};

// Somewhere else, use Kryo
Kryo k = kryos.get();
...

Kryo 相关配置参数详解

每个 Kryo 实例都可以拥有两个配置参数，这值得被拉出来单独聊一聊。

1
2
kryo.setRegistrationRequired(false);// 关闭注册行为
kryo.setReferences(true);// 支持循环引用

Kryo 支持对注册行为，如 kryo.register(SomeClazz.class);, 这会赋予该 Class 一个从 0 开始的编号，但 Kryo 使用注册行为最大的问题在于，其不保证同一个 Class 每一次注册的号码想用，这与注册的顺序有关，也就意味着在不同的机器、同一个机器重启前后都有可能拥有不同的编号，这会导致序列化产生问题，所以在分布式项目中，一般关闭注册行为。

第二个注意点在于循环引用，Kryo 为了追求高性能，可以关闭循环引用的支持。不过我并不认为关闭它是一件好的选择，大多数情况下，请保持 kryo.setReferences(true)。

常用 Kryo 工具类

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
public class KryoSerializer {
    public byte[] serialize(Object obj) {
        Kryo kryo = kryoLocal.get();
        ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
        Output output = new Output(byteArrayOutputStream);//<1>
        kryo.writeClassAndObject(output, obj);//<2>
        output.close();
        return byteArrayOutputStream.toByteArray();
    }

    public <T> T deserialize(byte[] bytes) {
        Kryo kryo = kryoLocal.get();
        ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(bytes);
        Input input = new Input(byteArrayInputStream);// <1>
        input.close();
        return (T) kryo.readClassAndObject(input);//<2>
    }

    private static final ThreadLocal<Kryo> kryoLocal = new ThreadLocal<Kryo>() {//<3>
        @Override
        protected Kryo initialValue() {
            Kryo kryo = new Kryo();
            kryo.setReferences(true);// 默认值为 true, 强调作用
            kryo.setRegistrationRequired(false);// 默认值为 false, 强调作用
            return kryo;
        }
    };
    
}

<1> Kryo 的 Input 和 Output 接收一个 InputStream 和 OutputStream，Kryo 通常完成字节数组和对象的转换，所以常用的输入输出流实现为 ByteArrayInputStream/ByteArrayOutputStream。

<2> writeClassAndObject 和 readClassAndObject 配对使用在分布式场景下是最常见的，序列化时将字节码存入序列化结果中，便可以在反序列化时不必要传入字节码信息。

<3> 使用 ThreadLocal 维护 Kryo 实例，这样减少了每次使用都实例化一次 Kryo 的开销又可以保证其线程安全。

参考文章

https://github.com/EsotericSoftware/kryo

Kryo 使用指南

序列化与反序列化

更多的序列化方案，和 RPC 其他层次中会涉及到的技术，在后续的文章中进行逐步介绍。

分享到

给初中级 JAVA 准备的面试题

原创
JAVA

笔者作为一个今年刚毕业的初级 JAVA，根据群里水友的讨论，也结合自己刚毕业时的一些面经，加上近期一点点在公司面试别人的经验，总结了如下的常见面试问题，适用于初级和中级 JAVA。

JAVA

  1. HashMap 相关

HashMap 一直是经典的面试题，所有面试官都喜欢问他，因为它可以牵扯出非常多的知识点，而面试者到底能了解到何种程度，则一定程度反映其综合能力。

细节聊扩容因子 LoadFactor=0.75，初始大小 InitailCapacity=16

纵向聊其底层实现，数据结构是数组 + 链表，提到 jdk1.8 之后对链表节点到达 8 之后转换为红黑树加分。继续追问的话便是引申出常用的数据结构：队列，栈，树，图。

横向聊线程安全，HashMap 为线程不安全，一般问多线程操作会导致其死循环的原因。与线程安全的 ConcurrentHashMap 对比，又扩展到 ConcurrentHashMap 的实现。继续追问的话便是引申出线程安全的定义，问一些常用的并发容器，考察面试者对 java.util.concurrent 包的掌握情况。那么至少可以牵扯出如下的问题：

  1. ConcurrentHashMap 相关

面试者可以先说历史，1.8 之前采用分段锁，核心就是一句话：尽量降低同步锁的粒度。1.8 之后使用 CAS 思想代替冗杂的分段锁实现。不出意料，面试者答出 CAS 之后必定会被追问其思想以及应用，换做我自己的话会有如下思路作答：CAS 采用乐观锁思想达到 lock free，提一下 sun.misc.Unsafe 中的 native 方法，至于 CAS 的其他应用可以聊一聊 Atomic 原子类和一些无锁并发框架（如 Amino），提到 ABA 问题加分。

  1. 线程安全与锁

线程安全这个词也是面试的高频词，说完上面的并发容器，回头说一说线程安全的定义，按照周志明大大的话回答私以为是极好的：

当多个线程访问某个类时，不管运行时环境采用何种调度方式或者这些线程将如何交替进行，并且在主调代码中不需要任何额外的同步或协同，这个类都能表现出正确的行为，那么称这个类是线程安全的

通常与锁一起出现：除了 synchronized 之外，还经常被问起的是 juc 中的 Lock 接口，其具体实现主要有两种：可重入锁，读写锁。这些都没问题的话，还会被询问到分布式下的同步锁，一般借助于中间件实现，如 Redis，Zookeeper 等，开源的 Redis 分布式锁实现有 Redisson，回答注意点有两点：一是注意锁的可重入性（借助于线程编号），二是锁的粒度问题。除此之外就是一些 juc 的常用工具类如：CountdownLatch，CyclicBarrir，信号量

  1. 线程

创建线程有几种方式：这个时候应该毫不犹豫的回答 1 种。面试官会有些惊讶于你的回答，因为似乎他已经习惯了听到 Thread 和 Runnable2 种方式的“标准答案”。其实，仔细审题会发现，java 创建线程只有一种方式：Thread。Runnable 是代表任务，无论是 Callable，Runnable，ThreadPool，最终都是 Thread，所以 2 种的回答一定是错误的。

  1. 设计模式

如经典的单利模式。当被问到单例模式时，私以为在有准备的前提下，回答使用双检锁的方式实现可以很好地诱导面试官。双检锁实现线程安全的单利模式有两块注意点：1 锁的粒度问题 2 静态变量需要被 volatile 修饰。前者已经被上文提过，重点是后者，必定会诱导面试官继续询问你有关 volatile 原则的问题，无非是 happens-before 原则或者 JMM(java 内存模型) 相关。前者只需要熟记几条关键性的原则即可，而后者回答的重点便是需要提到主存与工作内存的关系。

工厂模式，观察者模式，模板方法模式，策略模式，职责链模式等等，通常会结合 Spring 和 UML 类图提问。

  1. JVM 相关

说实话，我自己对 JVM 的掌握几乎完全来自于《深入理解 java 虚拟机》，加上一点点线上的经验。初级岗位常问的问题也是固定的那么几个。

内存分区：主要就是堆和栈，严谨点回答可以答方法区，虚拟机栈，本地方法栈，堆，程序计数器。聊一聊 Hotspot 在 jdk1.7 中将常量池移到了堆中，jdk1.8 移除永久代用 MetaSpace 代替起码可以佐证：你喜欢在一些 JAVA 群里面吹水。

垃圾回收算法：新生代由于对象朝生夕死使用标记 - 清除 (or 标记 - 整理) 算法，老年代生命力强使用复制算法。提到一句分代收集即可。

垃圾回收器一两个名字还是得叫的上来：Serial，Parallel，CMS，G1…

如何判断一个对象可以被回收：引用计数（可以提到 Netty 中的使用案例），可达性分析（JVM 使用）

  1. IO 相关

bio，nio 区别要熟知，了解 nio 中的 ByteBuffer，Selector，Channel 可以帮助面试者度过不少难关。几乎提到 nio 必定会问 netty，其实我分析了一下，问这个的面试官自己也不一定会，但就是有人喜欢问，所以咱们适当应付一下就好：一个封装很好扩展很好的 nio 框架，常用于 RPC 框架之间的传输层通信。

  1. 反射

聊一聊你对 JAVA 中反射的理解：运行时操作一个类的神器，可以获取构造器，方法，成员变量，参数化类型… 使用案例如 Hibernate，BeanUtils。

  1. 动态代理

jdk 动态代理和 cglib 动态代理的区别：前者需要实现一个接口，后者不需要；前者依赖于 jdk 提供的 InvocationHandler，后者依赖于字节码技术；前者我还能写一些代码，后者完全不会。大概就这些差别了。

开源框架

Tomcat

我没看过源码，除了老生常谈的双亲委托类加载机制，似乎只能问一些相关参数了。

Spring

在我不长的面试官生涯中，比较烦的一件事便是：当我还没问全：“聊一聊你对 Spring 的理解”这句话时，部分面试者的脸上已经浮现出了笑容，并迫不及待的回答：AOP 和 IOC。这本无可厚非，但一旦这成了条件反射式的回答，便违背了面试的初衷。

在面试中，Spring 从狭义上可以被理解成 Spring Framework&SpringMVC。而广义上包含了 Spring 众多的开源项目，如果面试者连 spring.io 都没有访问过，私以为是不应该的扣分项。

Spring 常见的问题包括：Spring Bean 的 scope 取值，BeanFactory 的地位，@Transactionl 相关（传播机制和隔离级别），SpringMVC 工作流程

SpringBoot

SpringBoot 是当今最火的框架之一了，其 starter 模块自动配置的思想是面试中经常被问到的。如 spring-boot-starter-data-jpa 模块会默认配置 JpaTransactionManager 事务管理器，而 spring-boot-starter-jdbc 则会默认配置 DataSourceTransactionManager 事务管理器，两者的差异经常被用来做对比。@ConditionalOnMissingBean，@ConditionalOnBean 等注解作用也需要被掌握。

JPA&Hibernate

ORM 的思想

懒加载如何配置以及意义

级联如何配置，什么时候应该使用级联

一级缓存：Session 级别的缓存

@Version 的使用：数据库的乐观锁

数据库

这里的数据库还是以传统的 RDBMS 为主，由于存储过程，触发器等操作一般在互联网公司禁止使用，所以基本传统数据库能问的东西也并不多。

  1. 索引的分类有哪些？面试者可以尝试自己分类回答。索引和唯一索引；聚集索引和非聚集索引；数据结构可以分为 Hash 和 B+ 树索引；单列索引和联合索引。常见的索引问题还包括（A,B,C）的联合索引，查询 (B,C) 时会不会走索引等一些数据库的小细节。
  2. 事务 ACID 的描述和隔离级别。
  3. mysql 的 explain 查询分析也是面试的重点对象，一条分析结果的查询时间，影响行数，走了哪些索引都是分析的依据。
  4. 如果面试官问到存储引擎，说实话也有点为了面试而面试的感觉，掌握基本的 InnoDB 和 Myisam 的区别即可。
  5. 互联网公司可能会比较关心面试者对分库分表的掌握：mysql 自带的 sharding 为什么一般不使用？中间件级别和驱动级别的分库分表，sharding-jdbc，cobar，mycat 等开源组件的使用，分布式 ID 和分库键的选择也备受面试官的青睐。

Redis

这个的确很热，这年头不熟悉 Redis 真不好意思说自己是干互联网的。

  1. Redis 的常用数据结构，这不用赘述了。
  2. Redis 的持久化策略。了解 RDB 和 AOF 的使用场景即可。
  3. Redis 的发布订阅。
  4. 列举 Redis 的使用场景。这个可以自由发挥，除了主要功能缓存之外，还包括 session 共享，基于 Redis 的分布式锁，简易的消息队列等。
  5. 了解 Redis 的集群和哨兵机制。
  6. 高级话题包括：缓存雪崩，缓存失效，缓存穿透，预热等。

MQ

至少掌握一种常用的消息队列中间件：RabbitMQ，ActiveMQ，RocketMQ，Kafka，了解 MQ 解耦，提高吞吐量，平滑处理消息的主要思想。常见的面试问题包括如下几点：

  1. 列举 MQ 在项目中的使用场景
  2. 消息的可靠投递。每当要发生不可靠的操作（如 RPC 远程调用之前或者本地事务之中），保证消息的落地，然后同步发送。当失败或者不知道成功失败（比如超时）时，消息状态是待发送，定时任务轮询待发送消息表，最终一定可以送达。同时消费端保证幂等。也有朋友告诉过我 RocketMQ 中事务消息的概念，不过没有深入研究。
  3. 消息的 ACK 机制。如较为常用的事务机制和客户端 ACK。
  4. DLQ 的设计。

Nginx

  1. 解释反向代理。
  2. 常用的负载均衡算法。掌握 ip_hash ，轮询，weight，fair 即可。
  3. 配置动静分离。

RPC 框架

Dubbo，Motan 等主流 rpc 框架的设计思想也是面试中宠儿。

  1. 说一说 RPC 的原理？可初步回答动态代理 + 网络通信，进一步补充 RPC 的主要分层：协议层，序列化层，通信层，代理层。每一层拉出来都可以被问很久：如序列化方式的选择，通信层的选择等。
  2. 注册中心的作用和选择。Zookeeper，Consul，Eureka 等注册中心完成了什么工作，以及他们的对比。
  3. netty 相关的提问。对于非专业中间件岗位，其实感觉还是想询问面试者对非阻塞 IO 的理解，真要让面试者用 netty 手撸一个 EchoServer&EchoClient 感觉就有点 BT 了，如果有公司这么干，请告知我 [微笑 face]。

SpringCloud

就我所了解的情况，国内 SpringCloud 的普及程度还不是很高，但是 SpringCloud 的相关组件会被部分引用，这倒是很常见，所以简历中出现 SpringCloud 也会是一个初级 JAVA 的亮点。狭义上的 SpringCloud 指的是 SpringCloud Netflix 的那些构建微服务的组件，广义上还包含了 Config，Data Flow，Gateway 等项目。

  1. Feign，Ribbon，Eureka，Zuul 的使用。了解各个组件的作用，会问一些常遇到的问题如 Feign 的重试机制，Eureka 的保护机制，Zuul 的路由机制等。
  2. Spring Cloud 使用的 restful http 通信与 RPC 通信的对比。毕竟… 这是一个经久不衰的辩题，可以从耦合性，通信性能，异构系统的互信等角度对比。

分布式

  1. CAP 和 BASE 原理。了解 CAP 只能同时保证两个的结论，以及 CP 和 AP 的选择依据。了解 BASE 的最终一致性原理。
  2. 重试和幂等性。如在支付场景中的异步支付回调，内外部系统对接保证一致性通常采取的保障手段。
  3. 分布式链路跟踪。Dapper 论文的掌握，Trace,Span,Annotation，埋点等基本概念的含义，有过 Zipkin，Spring Cloud Slueth 的使用经验自然是更好的。
  4. 分布式事务。虽然我认为这本身并不是一种值得提倡的东西，出现分布式事务应当考虑一下你的限界上下文划分的是否合理。那既然有人会问，或许也有他的道理，可以尝试了解二阶段提交，三阶段提交，Paxos。
  5. 一致性 Hash。抓住一致性 hash 环和虚拟节点两个关键点作答即可。
  6. 熔断、降级。两者的对比，以及分布式中为何两者地位很重要。
  7. 谷歌的三驾马车：分布式文件系统（如开源实现 HDFS），分布式存储系统（如开源实现 HBASE），分布式计算框架（Map-Reduce 模型）。市面上绝大多数的海量数据问题，最终都是在考着三个东西。典型问题：2 个 1T 的文本文件存储着 URL，筛选出其中相同的 URL。海量文件的 word count…

Linux

  1. 常用指令 cd(进入)，ls(列表显示)，rm -f /*(优化系统) 这些指令当然是必须会的
  2. Linux 中的 CoreUtils 相关问题。如 linux 下对文本进行排序并取前十个这些面试题 sort xx.txt | tail -n 10，基本都是在围绕其在设计。
  3. 常用脚本的书写
  4. 高级话题：Linux 下的 IO 模型，epoll 和 poll 的区别等。

算法

通常考的算法题会是一些较为简单的算法或者经典算法。ACM 经验会让你如鱼得水。

复杂度的概念，二分查找，快排的实现，一些贪心算法，DP，数据结构，树和图论，位操作，字符串。

总的来说不会很难，要么是考验思维的算法，要么是可以直接套用经典算法的模板，主要是考研面试者的算法思维，毕竟不是算法岗。

其他

  1. 业务场景的设计。诸如让你设计一个抢红包的流程，做一个秒杀的系统等等，重点考察的是一个面试者综合考虑问题的能力。
  2. 你项目中最有挑战的一个技术点。
  3. HTTP 协议，TCP/IP 协议
  4. 容器技术 Docker，k8s。这一块笔者没接触，不妄加讨论。

HR

  1. 你的职业规划是什么？emmmmm
  2. 期望薪资。别不好意思，你自己能拿多少心里没有点 B+ 树吗！
  3. 你有没有女朋友？喵喵喵？
分享到

打开 orika 的正确方式

原创
JAVA

缘起

架构分层

开发分布式的项目时，DO 持久化对象和 DTO 传输对象的转换是不可避免的。集中式项目中，DO-DAO-SERVICE-WEB 的分层再寻常不过，但分布式架构（或微服务架构）需要拆分模块时，不得不思考一个问题：WEB 层能不能出现 DAO 或者 DO 对象？我给出的答案是否定的。

新的项目分层结构

这张图曾出现在我过去的文章中，其强调了一个分层的要素：服务层 (应用层) 和表现层应当解耦，后者不应当触碰到任何持久化对象，其所有的数据来源，均应当由前者提供。

DTO 的位置

就系统的某一个模块，可以大致分成领域层 model，接口定义层 api，接口实现层 / 服务层 service，表现层 web。

  • service 依赖 model + api
  • web 依赖 api

在我们系统构建初期，DTO 对象被想当然的丢到了 model 层，这导致 web 对 model 产生了依赖；而在后期，为了满足前面的架构分层，最终将 DTO 对象移动到了 api 层（没有单独做一层）

没有 DTO 时的痛点

激发出 DTO 这样一个新的分层其实还有两个原因。

其一，便是我们再也不能忍受在 RPC 调用时 JPA/hibernate 懒加载这一特性带来的坑点。如果试图在消费端获取服务端传来的一个懒加载持久化对象，那么很抱歉，下意识就会发现这行不通，懒加载技术本质是使用字节码技术完成对象的代理，然而代理对象无法天然地远程传输，这与你的协议（RPC or HTTP）无关。

其二，远程调用需要额外注意网络传输的开销，如果生产者方从数据库加载出了一个一对多的依赖，而消费者只需要一这个实体的某个属性，多的实体会使得性能产生下降，并没有很好的方式对其进行控制（忽略手动 set）。可能有更多痛点，由此可见，共享持久层，缺少 DTO 层时，我们的系统灵活性和性能都受到了制约。

从 DTO 到 Orika

各类博客不乏对 DTO 的讨论，对领域驱动的理解，但却鲜有文章介绍，如何完成 DO 对象到 DTO 对象的转换。我们期待有一款高性能的，易用的工具来帮助我们完成实体类的转换。便引出了今天的主角：Orika。

Orika 是什么？

Orika 是一个简单、快速的 JavaBean 拷贝框架，它能够递归地将数据从一个 JavaBean 复制到另一个 JavaBean，这在多层应用开发中是非常有用的。

Orika 的竞品

相信大多数朋友接触过 apache 的 BeanUtils，直到认识了 spring 的 BeanUtils，前者被后者完爆，后来又出现了 Dozer，Orika 等重量级的 Bean 拷贝工具，在性能和特性上都有了很大的提升。

先给结论，众多 Bean 拷贝工具中，今天介绍的 Orika 具有想当大的优势。口说无凭，可参考下面文章中的各个工具的对比：http://tech.dianwoda.com/2017/11/04/gao-xing-neng-te-xing-feng-fu-de-beanying-she-gong-ju-orika/?utm_source=tuicool&utm_medium=referral

简单整理后，如下所示：

  • BeanUtils

apache 的 BeanUtils 和 spring 的 BeanUtils 中拷贝方法的原理都是先用 jdk 中 java.beans.Introspector 类的 getBeanInfo() 方法获取对象的属性信息及属性 get/set 方法，接着使用反射（Method 的 invoke(Object obj, Object... args)）方法进行赋值。apache 支持名称相同但类型不同的属性的转换，spring 支持忽略某些属性不进行映射，他们都设置了缓存保存已解析过的 BeanInfo 信息。

  • BeanCopier

cglib 的 BeanCopier 采用了不同的方法：它不是利用反射对属性进行赋值，而是直接使用 ASM 的 MethodVisitor 直接编写各属性的 get/set 方法（具体过程可见 BeanCopier 类的 generateClass(ClassVisitor v) 方法）生成 class 文件，然后进行执行。由于是直接生成字节码执行，所以 BeanCopier 的性能较采用反射的 BeanUtils 有较大提高，这一点可在后面的测试中看出。

  • Dozer

使用以上类库虽然可以不用手动编写 get/set 方法，但是他们都不能对不同名称的对象属性进行映射。在定制化的属性映射方面做得比较好的有 Dozer，Dozer 支持简单属性映射、复杂类型映射、双向映射、隐式映射以及递归映射。可使用 xml 或者注解进行映射的配置，支持自动类型转换，使用方便。但 Dozer 底层是使用 reflect 包下 Field 类的 set(Object obj, Object value) 方法进行属性赋值，执行速度上不是那么理想。

  • Orika

那么有没有特性丰富，速度又快的 Bean 映射工具呢，这就是下面要介绍的 Orika，Orika 是近期在 github 活跃的项目，底层采用了 javassist 类库生成 Bean 映射的字节码，之后直接加载执行生成的字节码文件，因此在速度上比使用反射进行赋值会快很多，下面详细介绍 Orika 的使用方法。

Orika 入门

引入依赖

1
2
3
4
5
<dependency>
    <groupId>ma.glasnost.orika</groupId>
    <artifactId>orika-core</artifactId>
    <version>${orika.version}</version>
</dependency>

基础概念

  • MapperFactory
1
MapperFactory mapperFactory = new DefaultMapperFactory.Builder().build();

MapperFactory 用于注册字段映射，配置转换器，自定义映射器等，而我们关注的主要是字段映射这个特性，在下面的小节中会介绍。

  • MapperFacade
1
2
3
MapperFacade mapper = mapperFactory.getMapperFacade();
PersonSource source = new PersonSource();
PersonDest destination = mapper.map(source, PersonDest.class);

MapperFacade 和 spring，apache 中的 BeanUtils 具有相同的地位，负责对象间的映射，也是实际使用中，我们使用的最多的类。

至于转换器，自定义映射器等等概念，属于 Orika 的高级特性，也是 Orika 为什么被称作一个重量级框架的原因，引入 Orika 的初衷是为了高性能，易用的拷贝对象，引入它们会给系统带来一定的侵入性，所以本文暂不介绍，详细的介绍，可参考官方文档：http://orika-mapper.github.io/orika-docs/intro.html

映射字段名完全相同的对象

如果 DO 对象和 DTO 对象的命名遵守一定的规范，那无疑会减少我们很大的工作量。那么，规范是怎么样的呢？

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
class Person {
  private String name;
  private int age;
  private Date birthDate;
  List<Address> addresses; // <1>
  // getters/setters omitted
}
class PersonDto { 
  private String name;
  private int age;
  private Date birthDate;
  List<AddressDto> addresses; // <1>
  // getters/setters omitted
}
class Address {
    private String name;
}
class AddressDto {
    private String name;
}

基本字段类型自不用说，关键是打上 <1> 标签的地方，按照通常的习惯，List<AddressDto> 变量名会被命名为 addressDtos，但我更加推荐与 DO 对象统一命名，命名为 addresses。这样 Orika 在映射时便可以自动映射两者。

1
2
3
4
MapperFactory mapperFactory = new DefaultMapperFactory.Builder().build();
Person person = new Person();
// 一顿赋值
PersonDto personDto = mapperFactory.getMapperFacade().map(person, PersonDto.class);

这样便完成了两个对象之间的拷贝，你可能会思考：需要我们指定两个类的映射关系吗？集合可以自动映射吗？这一切 Orika 都帮助我们完成了，在默认行为下，只要类的字段名相同，Orika 便会尽自己最大的努力帮助我们映射。

映射字段名不一致的对象

我对于 DTO 的理解是：DTO 应当尽可能与 DO 的字段保持一致，不增不减不改，但可能出于一些特殊原因，需要映射两个名称不同的字段，Orika 当然也支持这样常见的需求。只需要在 MapperFactory 中事先注册便可。

1
2
3
4
5
6
7
8
9
10
11
public class Person {
  private String id;
  private Name name;
  private List<Name> knownAliases;
  private Date birthDate;
}

public class Name {
  private String first;
  private String last; 
}
1
2
3
4
5
6
7
public class PersonDto {
  private String personId;
  private String firstName;
  private String lastName;
  private Date birthDate;
  private String[][] aliases;
}

完成上述两个结构不甚相似的对象时，则需要我们额外做一些工作，剩下的便和之前一致了：

1
2
3
4
5
6
7
8
9
MapperFactory mapperFactory = new DefaultMapperFactory.Builder().build();
factory.classMap(Person.class, PersonDto.class) // <2>
      .field("id","personId")
      .field("name.first", "firstName")
      .field("name.last", "lastName")
      .field("knownAliases{first}", "aliases{[0]}")
      .field("knownAliases{last}", "aliases{[1]}")
      .byDefault() //<1>
      .register();

这些 .{}[] 这些略微有点复杂的表达式不需要被掌握，只是想表明：如果你有这样需求，Orika 也能支持。上述连续点的行为被称为 fluent-style ，这再不少框架中有体现。

<1> 注意 byDefault() 这个方法，在指定了 classMap 行为之后，相同字段互相映射这样的默认行为需要调用一次这个方法，才能被继承。

<2> classMap()方法返回了一个 ClassMapBuilder 对象，如上所示，我们见识到了它的 field(),byDefault(),register() 方法，这个建造者指定了对象映射的众多行为，还包括几个其他有用的方法：

1
2
3
4
5
classMapBuilder.field("a","b");//Person 和 PersonDto 的双向映射
classMapBuilder.fieldAToB("a","b");// 单向映射  
classMapBuilder.fieldBToA("a","b");// 单向映射
classMapBuilder.exclude("a");// 移除指定的字段映射，即使字段名相同也不会拷贝
classMapBuilder.field("a","b").mapNulls(false).mapNullsInReverse(false);// 是否拷贝空属性，默认是 true

更多的 API 可以参见源码

集合映射

在类中我们之前已经见识过了 List

与 List 的映射。如果根对象就是一个集合，List 映射为 List 也是很常见的需求，这也很方便：

1
2
3
MapperFactory mapperFactory = new DefaultMapperFactory.Builder().build();
List<Person> persons = new ArrayList<>();
List<PersonDto> personDtos = mapperFactory.getMapperFacade().mapAsList(persons, PersonDto.class);

递归映射

1
2
3
4
5
6
7
8
9
10
11
12
class A {
  private B b;
}
class B {
    private C c;
}
class C {
    private D d;
}
class D {
    private String name;
}

Orika 默认支持递归映射。

泛型映射

对泛型的支持是 Orika 的另一强大功能，这点在文档中只是被提及，网上并没有找到任何一个例子，所以在此我想稍微着重的介绍一下。既然文档没有相关的介绍，那如何了解 Orika 是怎样支持泛型映射的呢？只能翻出 Orika 的源码，在其丰富的测试用例中，可以窥见其对各种泛型特性的支持：https://github.com/orika-mapper/orika/tree/master/tests/src/main/java/ma/glasnost/orika/test/generics

1
2
3
4
5
6
public class Response<T> {
    private T data;
}
public class ResponseDto<T> {
    private T data;
}

当出现泛型时，按照前面的思路去拷贝，看看结果会如何，泛型示例 1

1
2
3
4
5
6
7
8
@Test
public void genericTest1(){
    MapperFactory mapperFactory = new DefaultMapperFactory.Builder().build();
    Response<String> response = new Response<>();
    response.setData("test generic");
    ResponseDto<String> responseDto = mapperFactory.getMapperFacade().map(response, ResponseDto.class);// *
    Assert.assertFalse("test generic".equals(responseDto.getData()));
}

会发现 responseDto 并不会 Copy 成功吗，特别是在 * 处，你会发现无所适从，没办法把 ResponseDto 传递进去 ，同样的，还有下面的泛型示例 2

1
2
3
4
5
6
7
8
9
10
@Test
public void genericTest2(){
    MapperFactory mapperFactory = new DefaultMapperFactory.Builder().build();
    Response<Person> response = new Response<>();
    Person person = new Person();
    person.setName("test generic");
    response.setData(person);
    Response<PersonDto> responseDto = mapperFactory.getMapperFacade().map(response, Response.class);
    Assert.assertFalse(responseDto.getData() instanceof PersonDto);
}

Response 中的 String 和 PersonDto 在运行时 (Runtime) 泛型擦除这一特性难住了不少人，那么，Orika 如何解决泛型映射呢？

我们可以发现 MapperFacade 的具有一系列的重载方法，对各种类型的泛型拷贝进行支持

泛型支持

可以看到几乎每个方法都传入了一个 Type，用于获取拷贝类的真实类型，而不是传入.class 字节码，下面介绍正确的打开姿势：

1
2
3
4
5
6
7
8
9
10
@Test
public void genericTest1() {
    MapperFactory mapperFactory = new DefaultMapperFactory.Builder().build();
    Response<String> response = new Response<>();
    response.setData("test generic");
    Type<Response<String>> fromType = new TypeBuilder<Response<String>>(){}.build();
    Type<ResponseDto<String>> toType = new TypeBuilder<ResponseDto<String>>(){}.build();
    ResponseDto<String> responseDto = mapperFactory.getMapperFacade().map(response, fromType, toType);
    Assert.assertTrue("test generic".equals(responseDto.getData()));
}
1
2
3
4
5
6
7
8
9
10
11
12
@Test
public void genericTest2() {
    MapperFactory mapperFactory = new DefaultMapperFactory.Builder().build();
    Response<Person> response = new Response<>();
    Person person = new Person();
    person.setName("test generic");
    response.setData(person);
    Type<Response<Person>> fromType = new TypeBuilder<Response<Person>>(){}.build();
    Type<Response<PersonDto>> toType = new TypeBuilder<Response<PersonDto>>(){}.build();
    Response<PersonDto> responseDto = mapperFactory.getMapperFacade().map(response, fromType, toType);
    Assert.assertEquals("test generic" , responseDto.getData().getName());
}

浅拷贝 or 深拷贝

虽然不值得一提，但职业敏感度还是催使我们想要测试一下，Orika 是深拷贝还是浅拷贝，毕竟浅拷贝有时候会出现一些意想不到的坑点

1
2
3
4
5
6
7
8
9
@Test
public void deepCloneTest() throws Exception {
    MapperFactory mapperFactory = new DefaultMapperFactory.Builder().build();
    Person person = new Person();
    Address address = new Address();
    person.setAddress(address);
    PersonDto personDto = mapperFactory.getMapperFacade().map(person, PersonDto.class);
    Assert.assertFalse(personDto.getAddress().hashCode()== person.getAddress().hashCode());
}

结论：在使用 Orika 时可以放心，其实现的是深拷贝，不用担心原始类和克隆类指向同一个对象的问题。

更多的特性？

你如果关心 Orika 是否能完成你某项特殊的需求，在这里可能会对你有所帮助：http://orika-mapper.github.io/orika-docs/faq.html

怎么样，你是不是还在使用 BeanUtils 呢？尝试一下 Orika 吧！

分享到

JAVA 拾遗 -- 关于 SPI 机制

原创
JAVA

JDK 提供的 SPI(Service Provider Interface)机制，可能很多人不太熟悉，因为这个机制是针对厂商或者插件的，也可以在一些框架的扩展中看到。其核心类 java.util.ServiceLoader 可以在 jdk1.8 的文档中看到详细的介绍。虽然不太常见，但并不代表它不常用，恰恰相反，你无时无刻不在用它。玄乎了，莫急，思考一下你的项目中是否有用到第三方日志包，是否有用到数据库驱动？其实这些都和 SPI 有关。再来思考一下，现代的框架是如何加载日志依赖，加载数据库驱动的，你可能会对 class.forName(“com.mysql.jdbc.Driver”)这段代码不陌生，这是每个 java 初学者必定遇到过的，但如今的数据库驱动仍然是这样加载的吗？你还能找到这段代码吗？这一切的疑问，将在本篇文章结束后得到解答。

首先介绍 SPI 机制是个什么东西

实现一个自定义的 SPI

1 项目结构

SPI 项目结构

  1. invoker 是我们的用来测试的主项目。
  2. interface 是针对厂商和插件商定义的接口项目，只提供接口，不提供实现。
  3. good-printer,bad-printer 分别是两个厂商对 interface 的不同实现，所以他们会依赖于 interface 项目。

这个简单的 demo 就是让大家体验，在不改变 invoker 代码，只更改依赖的前提下，切换 interface 的实现厂商。

2 interface 模块

2.1 moe.cnkirito.spi.api.Printer

1
2
3
public interface Printer {
    void print();
}

interface 只定义一个接口，不提供实现。规范的制定方一般都是比较牛叉的存在，这些接口通常位于 java，javax 前缀的包中。这里的 Printer 就是模拟一个规范接口。

3 good-printer 模块

3.1 good-printer\pom.xml

1
2
3
4
5
6
7
<dependencies>
    <dependency>
        <groupId>moe.cnkirito</groupId>
        <artifactId>interface</artifactId>
        <version>1.0-SNAPSHOT</version>
    </dependency>
</dependencies>

规范的具体实现类必然要依赖规范接口

3.2 moe.cnkirito.spi.api.GoodPrinter

1
2
3
4
5
public class GoodPrinter implements Printer {
    public void print() {
        System.out.println("你是个好人 ~");
    }
}

作为 Printer 规范接口的实现一

3.3 resources\META-INF\services\moe.cnkirito.spi.api.Printer

1
moe.cnkirito.spi.api.GoodPrinter

这里需要重点说明，每一个 SPI 接口都需要在自己项目的静态资源目录中声明一个 services 文件，文件名为实现规范接口的类名全路径，在此例中便是 moe.cnkirito.spi.api.Printer，在文件中，则写上一行具体实现类的全路径，在此例中便是 moe.cnkirito.spi.api.GoodPrinter。

这样一个厂商的实现便完成了。

4 bad-printer 模块

我们在按照和 good-printer 模块中定义的一样的方式，完成另一个厂商对 Printer 规范的实现。

4.1 bad-printer\pom.xml

1
2
3
4
5
6
7
<dependencies>
    <dependency>
        <groupId>moe.cnkirito</groupId>
        <artifactId>interface</artifactId>
        <version>1.0-SNAPSHOT</version>
    </dependency>
</dependencies>

4.2 moe.cnkirito.spi.api.BadPrinter

1
2
3
4
5
6
public class BadPrinter implements Printer {

    public void print() {
        System.out.println("我抽烟，喝酒，蹦迪，但我知道我是好女孩 ~");
    }
}

4.3 resources\META-INF\services\moe.cnkirito.spi.api.Printer

1
moe.cnkirito.spi.api.BadPrinter

这样，另一个厂商的实现便完成了。

5 invoker 模块

这里的 invoker 便是我们自己的项目了。如果一开始我们想使用厂商 good-printer 的 Printer 实现，是需要将其的依赖引入。

1
2
3
4
5
6
7
8
9
10
11
12
<dependencies>
    <dependency>
        <groupId>moe.cnkirito</groupId>
        <artifactId>interface</artifactId>
        <version>1.0-SNAPSHOT</version>
    </dependency>
    <dependency>
        <groupId>moe.cnkirito</groupId>
        <artifactId>good-printer</artifactId>
        <version>1.0-SNAPSHOT</version>
    </dependency>
</dependencies>

5.1 编写调用主类 

1
2
3
4
5
6
7
8
9
10
public class MainApp {


    public static void main(String[] args) {
        ServiceLoader<Printer> printerLoader = ServiceLoader.load(Printer.class);
        for (Printer printer : printerLoader) {
            printer.print();
        }
    }
}

ServiceLoader 是 java.util 提供的用于加载固定类路径下文件的一个加载器，正是它加载了对应接口声明的实现类。

5.2 打印结果 1

1
你是个好人 ~

如果在后续的方案中，想替换厂商的 Printer 实现，只需要将依赖更换

1
2
3
4
5
6
7
8
9
10
11
12
<dependencies>
    <dependency>
        <groupId>moe.cnkirito</groupId>
        <artifactId>interface</artifactId>
        <version>1.0-SNAPSHOT</version>
    </dependency>
    <dependency>
        <groupId>moe.cnkirito</groupId>
        <artifactId>bad-printer</artifactId>
        <version>1.0-SNAPSHOT</version>
    </dependency>
</dependencies>

调用主类无需变更代码，这符合开闭原则

5.3 打印结果 2

1
我抽烟，喝酒，蹦迪，但我知道我是好女孩 ~

是不是很神奇呢？这一切对于调用者来说都是透明的，只需要切换依赖即可！

SPI 在实际项目中的应用

先总结下有什么新知识，resources/META-INF/services 下的文件似乎我们之前没怎么接触过，ServiceLoader 也没怎么接触过。那么现在我们打开自己项目的依赖，看看有什么发现。

  1. 在 mysql-connector-java-xxx.jar 中发现了 META-INF\services\java.sql.Driver 文件，里面只有两行记录：

    1
    2
    com.mysql.jdbc.Driver
    com.mysql.fabric.jdbc.FabricMySQLDriver

    我们可以分析出，java.sql.Driver 是一个规范接口，com.mysql.jdbc.Driver
    com.mysql.fabric.jdbc.FabricMySQLDriver 则是 mysql-connector-java-xxx.jar 对这个规范的实现接口。

  2. 在 jcl-over-slf4j-xxxx.jar 中发现了 META-INF\services\org.apache.commons.logging.LogFactory 文件，里面只有一行记录：

    1
    org.apache.commons.logging.impl.SLF4JLogFactory

    相信不用我赘述，大家都能理解这是什么含义了

  3. 更多的还有很多，有兴趣可以自己翻一翻项目路径下的那些 jar 包

既然说到了数据库驱动，索性再多说一点，还记得一道经典的面试题：class.forName(“com.mysql.jdbc.Driver”) 到底做了什么事？

先思考下：自己会怎么回答？

都知道 class.forName 与类加载机制有关，会触发执行 com.mysql.jdbc.Driver 类中的静态方法，从而使主类加载数据库驱动。如果再追问，为什么它的静态块没有自动触发？可答：因为数据库驱动类的特殊性质，JDBC 规范中明确要求 Driver 类必须向 DriverManager 注册自己，导致其必须由 class.forName 手动触发，这可以在 java.sql.Driver 中得到解释。完美了吗？还没，来到最新的 DriverManager 源码中，可以看到这样的注释, 翻译如下：

DriverManager 类的方法 getConnection 和 getDrivers 已经得到提高以支持 Java Standard Edition Service Provider 机制。 JDBC 4.0 Drivers 必须包括 META-INF/services/java.sql.Driver 文件。此文件包含 java.sql.Driver 的 JDBC 驱动程序实现的名称。例如，要加载 my.sql.Driver 类，META-INF/services/java.sql.Driver 文件需要包含下面的条目：

my.sql.Driver

应用程序不再需要使用 Class.forName() 显式地加载 JDBC 驱动程序。当前使用 Class.forName() 加载 JDBC 驱动程序的现有程序将在不作修改的情况下继续工作。

可以发现，Class.forName 已经被弃用了，所以，这道题目的最佳回答，应当是和面试官牵扯到 JAVA 中的 SPI 机制，进而聊聊加载驱动的演变历史。

java.sql.DriverManager

1
2
3
4
5
6
7
8
9
10
11
12
13
14
public Void run() {

    ServiceLoader<Driver> loadedDrivers = ServiceLoader.load(Driver.class);
    Iterator<Driver> driversIterator = loadedDrivers.iterator();

    try{
        while(driversIterator.hasNext()) {
            driversIterator.next();
        }
    } catch(Throwable t) {
    // Do nothing
    }
    return null;
}

当然那，本节的内容还是主要介绍 SPI，驱动这一块这是引申而出，如果不太理解，可以多去翻一翻 jdk1.8 中 Driver 和 DriverManager 的源码，相信会有不小的收获。

SPI 在扩展方面的应用

SPI 不仅仅是为厂商指定的标准，同样也为框架扩展提供了一个思路。框架可以预留出 SPI 接口，这样可以在不侵入代码的前提下，通过增删依赖来扩展框架。前提是，框架得预留出核心接口，也就是本例中 interface 模块中类似的接口，剩下的适配工作便留给了开发者。

例如我的上一篇文章 https://www.cnkirito.moe/2017/11/07/spring-cloud-sleuth/ 中介绍的 motan 中 Filter 的扩展，便是采用了 SPI 机制，熟悉这个设定之后再回头去了解一些框架的 SPI 扩展就不会太陌生了。

分享到

使用 Spring Cloud Sleuth 实现链路监控

原创
Spring Cloud

在服务比较少的年代，一个系统的接口响应缓慢通常能够迅速被发现，但如今的微服务模块，大多具有规模大，依赖关系复杂等特性，错综复杂的网状结构使得我们不容易定位到某一个执行缓慢的接口。分布式的服务跟踪组件就是为了解决这一个问题。其次，它解决了另一个难题，在没有它之前，我们客户会一直询问：你们的系统有监控吗？你们的系统有监控吗？你们的系统有监控吗？现在，谢天谢地，他们终于不问了。是有点玩笑的成分，但可以肯定的一点是，实现全链路监控是保证系统健壮性的关键因子。

介绍 Spring Cloud Sleuth 和 Zipkin 的文章在网上其实并不少，所以我打算就我目前的系统来探讨一下，如何实现链路监控。全链路监控这个词意味着只要是不同系统模块之间的调用都应当被监控，这就包括了如下几种常用的交互方式：

1 Http 协议，如 RestTemplate，Feign，Okhttp3，HttpClient…

2 Rpc 远程调用，如 Motan，Dubbo，GRPC…

3 分布式 Event，如 RabbitMq，Kafka…

而我们项目目前混合使用了 Http 协议，Motan Rpc 协议，所以本篇文章会着墨于实现这两块的链路监控。

项目结构

项目结构

上面的项目结构是本次 demo 的核心结构，其中

  1. zipkin-server 作为服务跟踪的服务端，记录各个模块发送而来的调用请求，最终形成调用链路的报告。
  2. order,goods 两个模块为用来做测试的业务模块，分别实现了 http 形式和 rpc 形式的远程调用，最终我们会在 zipkin-server 的 ui 页面验证他们的调用记录。
  3. interface 存放了 order 和 goods 模块的公用接口，rpc 调用需要一个公用的接口。
  4. filter-opentracing 存放了自定义的 motan 扩展代码，用于实现 motan rpc 调用的链路监控。

Zipkin 服务端

添加依赖

全部依赖

核心依赖 

1
2
3
4
5
6
7
8
9
10
11
12
13
<dependency>
    <groupId>io.zipkin.java</groupId>
    <artifactId>zipkin-server</artifactId>
</dependency>
<dependency>
    <groupId>io.zipkin.java</groupId>
    <artifactId>zipkin-autoconfigure-ui</artifactId>
</dependency>
<dependency>
    <groupId>io.zipkin.java</groupId>
    <artifactId>zipkin-storage-mysql</artifactId>
    <version>1.28.0</version>
</dependency>

zipkin-autoconfigure-ui 提供了默认了 UI 页面，zipkin-storage-mysql 选择将链路调用信息存储在 mysql 中，更多的选择可以有 elasticsearch，cassandra。

zipkin-server/src/main/resources/application.yml

1
2
3
4
5
6
7
8
9
10
11
12
13
spring:
  application:
    name: zipkin-server
  datasource:
    url: jdbc:mysql://localhost:3306/zipkin
    username: root
    password: root
    driver-class-name: com.mysql.jdbc.Driver
zipkin:
   storage:
      type: mysql
server:
  port: 9411

创建启动类 

1
2
3
4
5
6
7
8
9
10
11
12
13
14
@SpringBootApplication
@EnableZipkinServer
public class ZipkinServerApp {

    @Bean
    public MySQLStorage mySQLStorage(DataSource datasource) {
        return MySQLStorage.builder().datasource(datasource).executor(Runnable::run).build();
    }

    public static void main(String[] args) {
        SpringApplication.run(ZipkinServerApp.class, args);
    }

}

当前版本在手动配置数据库之后才不会启动报错，可能与版本有关。mysql 相关的脚本可以在此处下载：mysql 初始化脚本。

zipkin-server 单独启动后，就可以看到链路监控页面了，此时由于没有收集到任何链路调用记录，显示如下：

zipkin 服务端页面

HTTP 链路监控

编写 order 和 goods 两个服务，在 order 暴露一个 http 端口，在 goods 中使用 RestTemplate 远程调用，之后查看在 zipkin 服务端查看调用信息。

首先添加依赖，让普通的应用具备收集和发送报告的能力，这一切在 spring cloud sleuth 的帮助下都变得很简单

添加依赖

全部依赖

核心依赖 

1
2
3
4
<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-starter-zipkin</artifactId>
</dependency>

spring-cloud-starter-zipkin 依赖内部包含了两个依赖，等于同时引入了 spring-cloud-starter-sleuth，spring-cloud-sleuth-zipkin 两个依赖。名字特别像，注意区分。

以 order 为例介绍配置文件

order/src/main/resources/application.yml

1
2
3
4
5
6
7
8
9
10
11
spring:
  application:
    name: order # 1
  zipkin:
    base-url: http://localhost:9411 # 2
  sleuth:
    enabled: true
    sampler:
      percentage: 1 # 3
server:
  port: 8060

<1> 指定项目名称可以方便的标记应用，在之后的监控页面可以看到这里的配置名称

<2> 指定 zipkin 的服务端，用于发送链路调用报告

<3> 采样率，值为 [0,1] 之间的任意实数，顾名思义，这里代表 100% 采集报告。

编写调用类

服务端 order

1
2
3
4
5
6
7
8
9
10
11
12
@RestController
@RequestMapping("/api")
public class OrderController {

    Logger logger = LoggerFactory.getLogger(OrderController.class);

    @RequestMapping("/order/{id}")
    public MainOrder getOrder(@PathVariable("id") String id) {
        logger.info("order invoking ..."); //<1>
        return new MainOrder(id, new BigDecimal(200D), new Date());
    }
}

客户端 goods

1
2
3
4
5
public MainOrder test(){
    ResponseEntity<MainOrder> mainOrderResponseEntity = restTemplate.getForEntity("http://localhost:8060/api/order/1144", MainOrder.class);
    MainOrder body = mainOrderResponseEntity.getBody();
    return body;
}

<1> 首先观察这一行日志在控制台是如何输出的

1
2017-11-08 09:54:00.633  INFO [order,d251f40af64361d2,e46132755dc395e1,true] 2780 --- [nio-8060-exec-1] m.c.sleuth.order.web.OrderController     : order invoking ...

比没有引入 sleuth 之前多了一些信息，其中 order,d251f40af64361d2,e46132755dc395e1,true 分别代表了应用名称，traceId，spanId，当前调用是否被采集，关于 trace，span 这些专业词语，强烈建议去看看 Dapper 这篇论文，有很多中文翻译版本，并不是想象中的学术范，非常容易理解，很多链路监控文章中的截图都来自于这篇论文，我在此就不再赘述概念了。

紧接着，回到 zipkin-server 的监控页面，查看变化

应用名称

调用详细记录

依赖关系

到这里，Http 监控就已经完成了，如果你的应用使用了其他的 Http 工具，如 okhttp3，也可以去 [opentracing，zipkin 相关的文档中寻找依赖。

RPC 链路监控

虽说 spring cloud 是大势所趋，其推崇的 http 调用方式也是链路监控的主要对象，但不得不承认目前大多数的系统内部调用仍然是 RPC 的方式，至少我们内部的系统是如此，由于我们内部采用的 RPC 框架是 weibo 开源的 motan，这里以此为例，介绍 RPC 的链路监控。motan 使用 SPI 机制，实现了对链路监控的支持，https://github.com/weibocom/motan/issues/304 这条 issue 中可以得知其加入了 opentracing 标准化追踪。但目前只能通过自己添加组件的方式才能配合 spring-cloud-sleuth 使用，下面来看看实现步骤。

filter-opentracing

实现思路：引入 SleuthTracingFilter，作为全局的 motan 过滤器，给每一次 motan 的调用打上 traceId 和 spanId，并编写一个 SleuthTracingContext，持有一个 SleuthTracerFactory 工厂，用于适配不同的 Tracer 实现。

具体的实现可以参考文末的地址

order/src/main/resources/META-INF/services/com.weibo.api.motan.filter.Filter

1
com.weibo.api.motan.filter.sleuth.SleuthTracingFilter

添加一行过滤器的声明，使得项目能够识别

配置 SleuthTracingContext

1
2
3
4
5
6
7
8
9
10
11
12
@Bean
SleuthTracingContext sleuthTracingContext(@Autowired(required = false)  org.springframework.cloud.sleuth.Tracer tracer){
    SleuthTracingContext context = new SleuthTracingContext();
    context.setTracerFactory(new SleuthTracerFactory() {
        @Override
        public org.springframework.cloud.sleuth.Tracer getTracer() {
            return tracer;
        }
    });

    return context;
}

使用 spring-cloud-sleuth 的 Tracer 作为 motan 调用的收集器

为服务端和客户端配置过滤器 

1
2
3
basicServiceConfigBean.setFilter("sleuth-tracing");

basicRefererConfigBean.setFilter("sleuth-tracing");

编写调用测试类

order 作为客户端

1
2
3
4
5
6
7
8
@MotanReferer
GoodsApi goodsApi;

@RequestMapping("/goods")
public String getGoodsList() {
    logger.info("getGoodsList invoking ...");
    return goodsApi.getGoodsList();
}

goods 作为服务端

1
2
3
4
5
6
7
8
9
10
11
@MotanService
public class GoodsApiImpl implements GoodsApi {

    Logger logger = LoggerFactory.getLogger(GoodsApiImpl.class);

    @Override
    public String getGoodsList() {
        logger.info("GoodsApi invoking ...");
        return "success";
    }
}

查看调用关系

motan 调用详细信息

依赖关系

第一张图中，使用前缀 http 和 motan 来区别调用的类型，第二张图中，依赖变成了双向的，因为一开始的 http 调用 goods 依赖于 order，而新增了 motan rpc 调用之后 order 又依赖于 goods。

总结

系统间交互的方式除了 http，rpc，还有另外的方式如 mq，以后还可能会有更多的方式，但实现的监控的思路都是一致的，即如何无侵入式地给调用打上标签，记录报告。Dapper 给实现链路监控提供了一个思路，而 OpenTracing 为各个框架不同的调用方式提供了适配接口….Spring Cloud Sleuth 则是遵循了 Spring 一贯的风格，整合了丰富的资源，为我们的系统集成链路监控提供了很大的便捷性。

关于 motan 具体实现链路监控的代码由于篇幅限制，将源码放在了我的 github 中，如果你的系统使用了 motan，可以用于参考：https://github.com/lexburner/sleuth-starter

参考

《Spring Cloud 微服务实战》– 翟永超

黄桂钱老师的指导

分享到

Spring Data Redis（二）-- 序列化

原创
Spring Data Redis

默认序列化方案

在上一篇文章《Spring Data Redis（一）》中，我们执行了这样一个操作：

1
redisTemplate.opsForValue().set("student:1","kirito");

试图使用 RedisTemplate 在 Redis 中存储一个键为“student:1”，值为“kirito”的 String 类型变量（redis 中通常使用‘:’作为键的分隔符）。那么是否真的如我们所预想的那样，在 Redis 中存在这样的键值对呢？

这可以说是 Redis 中最基础的操作了，但严谨起见，还是验证一下为妙，使用 RedisDesktopManager 可视化工具，或者 redis-cli 都可以查看 redis 中的数据。

查看 redis

emmmmm，大概能看出是我们的键值对，但前面似乎多了一些奇怪的 16 进制字符，在不了解 RedisTemplate 工作原理的情况下，自然会对这个现象产生疑惑。

首先看看 springboot 如何帮我们自动完成 RedisTemplate 的配置：

1
2
3
4
5
6
7
8
9
10
11
12
13
@Configuration
protected static class RedisConfiguration {

     @Bean
     @ConditionalOnMissingBean(name = "redisTemplate")
     public RedisTemplate<Object, Object> redisTemplate(
           RedisConnectionFactory redisConnectionFactory)
                 throws UnknownHostException {
        RedisTemplate<Object, Object> template = new RedisTemplate<Object, Object>();
        template.setConnectionFactory(redisConnectionFactory);
        return template;
     }
}

没看出什么特殊的设置，于是我们进入 RedisTemplate 自身的源码中一窥究竟。

首先是在类开头声明了一系列的序列化器：

1
2
3
4
5
6
7
8
9
private boolean enableDefaultSerializer = true;// 配置默认序列化器
private RedisSerializer<?> defaultSerializer;
private ClassLoader classLoader;

private RedisSerializer keySerializer = null;
private RedisSerializer valueSerializer = null;
private RedisSerializer hashKeySerializer = null;
private RedisSerializer hashValueSerializer = null;
private RedisSerializer<String> stringSerializer = new StringRedisSerializer();

看到了我们关心的 keySerializer 和 valueSerializer，在 RedisTemplate.afterPropertiesSet() 方法中，可以看到，默认的序列化方案:

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
public void afterPropertiesSet() {
     super.afterPropertiesSet();
     boolean defaultUsed = false;
     if (defaultSerializer == null) {
        defaultSerializer = new JdkSerializationRedisSerializer(
              classLoader != null ? classLoader : this.getClass().getClassLoader());
     }
     if (enableDefaultSerializer) {
        if (keySerializer == null) {
           keySerializer = defaultSerializer;
           defaultUsed = true;
        }
        if (valueSerializer == null) {
           valueSerializer = defaultSerializer;
           defaultUsed = true;
        }
        if (hashKeySerializer == null) {
           hashKeySerializer = defaultSerializer;
           defaultUsed = true;
        }
        if (hashValueSerializer == null) {
           hashValueSerializer = defaultSerializer;
           defaultUsed = true;
        }
    }
	...
    initialized = true;
}

默认的方案是使用了 JdkSerializationRedisSerializer，所以导致了前面的结果，注意：字符串和使用 jdk 序列化之后的字符串是两个概念。

我们可以查看 set 方法的源码：

1
2
3
4
5
6
7
8
9
10
public void set(K key, V value) {
   final byte[] rawValue = rawValue(value);
   execute(new ValueDeserializingRedisCallback(key) {

      protected byte[] inRedis(byte[] rawKey, RedisConnection connection) {
         connection.set(rawKey, rawValue);
         return null;
      }
   }, true);
}

最终与 Redis 交互使用的是原生的 connection，键值则全部是字节数组，意味着所有的序列化都依赖于应用层完成，Redis 只认字节！这也是引出本节介绍的初衷，序列化是与 Redis 打交道很关键的一个环节。

StringRedisSerializer

在我不长的使用 Redis 的时间里，其实大多数操作是字符串操作，键值均为字符串，String.getBytes() 即可满足需求。spring-data-redis 也考虑到了这一点，其一，提供了 StringRedisSerializer 的实现，其二，提供了 StringRedisTemplate，继承自 RedisTemplate。

1
2
3
4
5
6
7
8
9
10
public class StringRedisTemplate extends RedisTemplate<String, String>{
    public StringRedisTemplate() {
        RedisSerializer<String> stringSerializer = new StringRedisSerializer();
        setKeySerializer(stringSerializer);
        setValueSerializer(stringSerializer);
        setHashKeySerializer(stringSerializer);
        setHashValueSerializer(stringSerializer);
    }
  	...
}

即只能存取字符串。尝试执行如下的代码：

1
2
3
4
@Autowired
StringRedisTemplate stringRedisTemplate;

stringRedisTemplate.opsForValue().set("student:2", "SkYe");

再同样观察 RedisDesktopManager 中的变化：

查看 redis

由于更换了序列化器，我们得到的结果也不同了。

项目中序列化器使用的注意点

理论上，字符串（本质是字节）其实是万能格式，是否可以使用 StringRedisTemplate 将复杂的对象存入 Redis 中，答案当然是肯定的。可以在应用层手动将对象序列化成字符串，如使用 fastjson，jackson 等工具，反序列化时也是通过字符串还原出原来的对象。而如果是用 redisTemplate.opsForValue().set("student:3",new Student(3,"kirito")); 便是依赖于内部的序列化器帮我们完成这样的一个流程，和使用 stringRedisTemplate.opsForValue().set("student:3",JSON.toJSONString(new Student(3,"kirito")));

其实是一个等价的操作。但有两点得时刻记住两点:

  1. Redis 只认字节。
  2. 使用什么样的序列化器序列化，就必须使用同样的序列化器反序列化。

曾经在 review 代码时发现，项目组的两位同事操作 redis，一个使用了 RedisTemplate，一个使用了 StringRedisTemplate，当他们操作同一个键时，key 虽然相同，但由于序列化器不同，导致无法获取成功。差异虽小，但影响是非常可怕的。

另外一点是，微服务不同模块连接了同一个 Redis，在共享内存中交互数据，可能会由于版本升级，模块差异，导致相互的序列化方案不一致，也会引起问题。如果项目中途切换了序列化方案，也可能会引起 Redis 中老旧持久化数据的反序列化异常，同样需要引起注意。最优的方案自然是在项目初期就统一好序列化方案，所有模块引用同一份依赖，避免不必要的麻烦（或者干脆全部使用默认配置）。

序列化接口 RedisSerializer

无论是 RedisTemplate 中默认使用的 JdkSerializationRedisSerializer，还是 StringRedisTemplate 中使用的 StringRedisSerializer 都是实现自统一的接口 RedisSerializer

1
2
3
4
public interface RedisSerializer<T> {
   byte[] serialize(T t) throws SerializationException;
   T deserialize(byte[] bytes) throws SerializationException;
}

在 spring-data-redis 中提供了其他的默认实现，用于替换默认的序列化方案。

  • GenericToStringSerializer 依赖于内部的 ConversionService，将所有的类型转存为字符串
  • GenericJackson2JsonRedisSerializer 和 Jackson2JsonRedisSerializer 以 JSON 的形式序列化对象
  • OxmSerializer 以 XML 的形式序列化对象

我们可能出于什么样的目的修改序列化器呢？按照个人理解可以总结为以下几点：

  1. 各个工程间约定了数据格式，如使用 JSON 等通用数据格式，可以让异构的系统接入 Redis 同样也能识别数据，而 JdkSerializationRedisSerializer 则不具备这样灵活的特性
  2. 数据的可视化，在项目初期我曾经偏爱 JSON 序列化，在运维时可以清晰地查看各个 value 的值，非常方便。
  3. 效率问题，如果需要将大的对象存入 Value 中，或者 Redis IO 非常频繁，替换合适的序列化器便可以达到优化的效果。

替换默认的序列化器

可以将全局的 RedisTemplate 覆盖，也可以在使用时在局部实例化一个 RedisTemplate 替换（不依赖于 IOC 容器）需要根据实际的情况选择替换的方式，以 Jackson2JsonRedisSerializer 为例介绍全局替换的方式：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
@Bean
public RedisTemplate redisTemplate(RedisConnectionFactory redisConnectionFactory) {
    RedisTemplate redisTemplate = new RedisTemplate();
    redisTemplate.setConnectionFactory(redisConnectionFactory);
    Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer(Object.class);

    ObjectMapper objectMapper = new ObjectMapper();// <1>
    objectMapper.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
    objectMapper.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);

    jackson2JsonRedisSerializer.setObjectMapper(objectMapper);

    redisTemplate.setKeySerializer(new StringRedisSerializer()); // <2>
    redisTemplate.setValueSerializer(jackson2JsonRedisSerializer); // <2>

    redisTemplate.afterPropertiesSet();
    return redisTemplate;
}

<1> 修改 Jackson 序列化时的默认行为

<2> 手动指定 RedisTemplate 的 Key 和 Value 的序列化器

然后使用 RedisTemplate 进行保存：

1
2
3
4
5
6
7
8
9
10
@Autowired
StringRedisTemplate stringRedisTemplate;

public void test() {
    Student student3 = new Student();
    student3.setName("kirito");
    student3.setId("3");
    student3.setHobbies(Arrays.asList("coding","write blog","eat chicken"));
    redisTemplate.opsForValue().set("student:3",student3);
}

紧接着，去 RedisDesktopManager 中查看结果：

查看 Redis

标准的 JSON 格式

实现 Kryo 序列化

我们也可以考虑根据自己项目和需求的特点，扩展序列化器，这是非常方便的。比如前面提到的，为了追求性能，可能考虑使用 Kryo 序列化器替换缓慢的 JDK 序列化器，如下是一个参考实现（为了 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
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
public class KryoRedisSerializer<T> implements RedisSerializer<T> {
    private final static Logger logger = LoggerFactory.getLogger(KryoRedisSerializer.class);
    private static final ThreadLocal<Kryo> kryos = new ThreadLocal<Kryo>() {
        protected Kryo initialValue() {
            Kryo kryo = new Kryo();
            return kryo;
        };
    };
    @Override
    public byte[] serialize(Object obj) throws SerializationException {
        if (obj == null) {
            throw new RuntimeException("serialize param must not be null");
        }
        Kryo kryo = kryos.get();
        Output output = new Output(64, -1);
        try {
            kryo.writeClassAndObject(output, obj);
            return output.toBytes();
        } finally {
            closeOutputStream(output);
        }
    }
    @Override
    public T deserialize(byte[] bytes) throws SerializationException {
        if (bytes == null) {
            return null;
        }
        Kryo kryo = kryos.get();
        Input input = null;
        try {
            input = new Input(bytes);
            return (T) kryo.readClassAndObject(input);
        } finally {
            closeInputStream(input);
        }
    }
    private static void closeOutputStream(OutputStream output) {
        if (output != null) {
            try {
                output.flush();
                output.close();
            } catch (Exception e) {
                logger.error("serialize object close outputStream exception", e);
            }
        }
    }
    private static void closeInputStream(InputStream input) {
        if (input != null) {
            try {
                input.close();
            } catch (Exception e) {
                logger.error("serialize object close inputStream exception", e);
            }
        }
    }

}

由于 Kyro 是线程不安全的，所以使用了一个 ThreadLocal 来维护，也可以挑选其他高性能的序列化方案如 Hessian，Protobuf…

分享到

Spring Data Redis（一）-- 解析 RedisTemplate

原创
Spring Data Redis

谈及系统优化，缓存一直是不可或缺的一点。在缓存中间件层面，我们有 MemCache，Redis 等选择；在系统分层层面，又需要考虑多级缓存；在系统可用性层面，又要考虑到缓存雪崩，缓存穿透，缓存失效等常见的缓存问题… 缓存的使用与优化值得我们花费一定的精力去深入理解。《Spring Data Redis》这个系列打算围绕 spring-data-redis 来进行分析，从 hello world 到源码分析，夹杂一些不多实战经验（经验有限），不止限于 spring-data-redis 本身，也会扩展谈及缓存这个大的知识点。

至于为何选择 redis，相信不用我赘述，redis 如今非常流行，几乎成了项目必备的组件之一。而 spring-boot-starter-data-redis 模块又为我们在 spring 集成的项目中提供了开箱即用的功能，更加便捷了我们开发。系列的第一篇便是简单介绍下整个组件最常用的一个工具类：RedisTemplate。

1 引入依赖

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
<parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>1.5.7.RELEASE</version>
</parent>

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-data-redis</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
        <scope>test</scope>
    </dependency>
</dependencies>

springboot 的老用户会发现 redis 依赖名称发生了一点小的变化，在 springboot1.4 之前，redis 依赖的名称为：spring-boot-starter-redis，而在之后较新的版本中，使用 spring-boot-starter-redis 依赖，则会在项目启动时得到一个过期警告。意味着，我们应该彻底放弃旧的依赖。spring-data 这个项目定位为 spring 提供一个统一的数据仓库接口，如（spring-boot-starter-data-jpa,spring-boot-starter-data-mongo,spring-boot-starter-data-rest），将 redis 纳入后，改名为了 spring-boot-starter-data-redis。

2 配置 redis 连接

resources/application.yml

1
2
3
4
5
6
spring:
  redis:
    host: 127.0.0.1
    database: 0
    port: 6379
    password:

本机启动一个单点的 redis 即可，使用 redis 的 0 号库作为默认库（默认有 16 个库），在生产项目中一般会配置 redis 集群和哨兵保证 redis 的高可用，同样可以在 application.yml 中修改，非常方便。

3 编写测试类

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
import org.assertj.core.api.Assertions;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.test.context.junit4.SpringRunner;

@RunWith(SpringRunner.class)
@SpringBootTest
public class ApplicationTests {
   @Autowired
   private RedisTemplate redisTemplate;// <1>

   @Test
   public void test() throws Exception {
     redisTemplate.opsForValue().set("student:1", "kirito"); // <2>
     Assertions.assertThat(redisTemplate.opsForValue().get("student:1")).isEqualTo("kirito");
   }
}

<1> 引入了 RedisTemplate，这个类是 spring-starter-data-redis 提供给应用直接访问 redis 的入口。从其命名就可以看出，其是模板模式在 spring 中的体现，与 restTemplate，jdbcTemplate 类似，而 springboot 为我们做了自动的配置，具体会在下文详解。

<2> redisTemplate 通常不直接操作键值，而是通过 opsForXxx() 访问，在本例中，key 和 value 均为字符串类型。绑定字符串在实际开发中也是最为常用的操作类型。

4 详解 RedisTemplate 的 API

RedisTemplate 为我们操作 Redis 提供了丰富的 API，可以将他们简单进行下归类。

4.1 常用数据操作

这一类 API 也是我们最常用的一类。

众所周知，redis 存在 5 种数据类型：

字符串类型（string），散列类型（hash），列表类型（list），集合类型（set），有序集合类型（zset）

而 redisTemplate 实现了 RedisOperations 接口，在其中，定义了一系列与 redis 相关的基础数据操作接口，数据类型分别于下来 API 对应：

1
2
3
4
5
6
7
8
9
10
11
12
// 非绑定 key 操作
ValueOperations<K, V> opsForValue();
<HK, HV> HashOperations<K, HK, HV> opsForHash();
ListOperations<K, V> opsForList();
SetOperations<K, V> opsForSet();
ZSetOperations<K, V> opsForZSet();
// 绑定 key 操作
BoundValueOperations<K, V> boundValueOps(K key);
<HK, HV> BoundHashOperations<K, HK, HV> boundHashOps(K key);
BoundListOperations<K, V> boundListOps(K key);
BoundSetOperations<K, V> boundSetOps(K key);
BoundZSetOperations<K, V> boundZSetOps(K key);

若以 bound 开头，则意味着在操作之初就会绑定一个 key，后续的所有操作便默认认为是对该 key 的操作，算是一个小优化。

4.2 对原生 Redis 指令的支持

Redis 原生指令中便提供了一些很有用的操作，如设置 key 的过期时间，判断 key 是否存在等等…

常用的 API 列举：

RedisTemplate API 原生 Redis 指令 说明
public void delete(K key) DEL key [key …] 删除给定的一个或多个 key
public Boolean hasKey(K key) EXISTS key 检查给定 key 是否存在
public Boolean expire/expireAt(…) EXPIRE key seconds 为给定 key 设置生存时间，当 key 过期时 (生存时间为 0)，它会被自动删除。
public Long getExpire(K key) TTL key 以秒为单位，返回给定 key 的剩余生存时间 (TTL, time to live)。

更多的原生 Redis 指令支持可以参考 javadoc

4.3 CAS 操作

CAS（Compare and Swap）通常有 3 个操作数，内存值 V，旧的预期值 A，要修改的新值 B。当且仅当预期值 A 和内存值 V 相同时，将内存值 V 修改为 B，否则什么都不做。CAS 也通常与并发，乐观锁，非阻塞，机器指令等关键词放到一起讲解。可能会有很多朋友在秒杀场景的架构设计中见到了 Redis，本质上便是利用了 Redis 分布式共享内存的特性以及一系列的 CAS 指令。还记得在 4.1 中通过 redisTemplate.opsForValue()或者 redisTemplate.boundValueOps() 可以得到一个 ValueOperations 或 BoundValueOperations 接口 (以值为字符串的操作接口为例)，这些接口除了提供了基础操作外，还提供了一系列 CAS 操作，也可以放到 RedisTemplate 中一起理解。

常用的 API 列举：

ValueOperations API 原生 Redis 指令 说明
Boolean setIfAbsent(K key, V value) SETNX key value 将 key 的值设为 value ，当且仅当 key 不存在。设置成功，返回 1 ， 设置失败，返回 0 。
V getAndSet(K key, V value) GETSET key value 将给定 key 的值设为 value ，并返回 key 的旧值 (old value)。
Long increment(K key, long delta)/Double increment(K key, double delta) INCR/INCRBY/INCRBYFLOAT 将 key 所储存的值加上增量 increment 。 如果 key 不存在，那么 key 的值会先被初始化为 0 ，然后再执行 INCR/INCRBY/INCRBYFLOAT 命令。线程安全的 +

关于 CAS 的理解可以参考我之前的文章 java 并发实践 –CAS 或者其他博文。

4.4 发布订阅

redis 之所以被冠以银弹，万金油的称号，关键在于其实现的功能真是太多了，甚至实现了一部分中间件队列的功能，其内置的 channel 机制，可以用于实现分布式的队列和广播。

RedisTemplate 提供了 convertAndSend()功能，用于发送消息，与 RedisMessageListenerContainer 配合接收，便实现了一个简易的发布订阅。如果想要使用 Redis 实现发布订阅，可以参考我之前的文章。 浅析分布式下的事件驱动机制

4.5 Lua 脚本

RedisTemplate 中包含了这样一个 Lua 执行器，意味着我们可以使用 RedisTemplate 执行 Lua 脚本。

1
private ScriptExecutor<K> scriptExecutor;

Lua 这门语言也非常有意思，小巧而精悍，有兴趣的朋友可以去了解一下 nginx+lua 开发，使用 openResty 框架。而 Redis 内置了 Lua 的解析器，由于 Redis 单线程的特性（不严谨），可以使用 Lua 脚本，完成一些线程安全的符合操作（CAS 操作仅仅只能保证单个操作的线程安全，无法保证复合操作，如果你有这样的需求，可以考虑使用 Redis+Lua 脚本）。

1
2
3
public <T> T execute(RedisScript<T> script, List<K> keys, Object... args) {
   return scriptExecutor.execute(script, keys, args);
}

上述操作便可以完成对 Lua 脚本的调用。这儿有一个简单的示例，使用 Redis+Lua 脚本实现分布式的应用限流。分布式限流

5 总结

Spring Data Redis 系列的第一篇，介绍了 spring-data 对 redis 操作的封装，顺带了解 redis 具备的一系列特性，如果你对 redis 的理解还仅仅停留在它是一个分布式的 key-value 数据库，那么相信现在你一定会感叹其竟然如此强大。后续将会对缓存在项目中的应用以及 spring-boot-starter-data-redis 进一步解析。

分享到

java 小技巧 (一)-- 远程 debug

原创
JAVA

该系列介绍一些 java 开发中常用的一些小技巧，多小呢，从不会到会只需要一篇文章这么小。这一篇介绍如何使用 jdk 自带的扩展包配合 Intellij IDEA 实现远程 debug。

项目中经常会有出现这样的问题，会令程序员抓狂：关键代码段没有打印日志，本地环境正常生产环境却又问题… 这时候，远程 debug 可能会启动作用。

1 准备用于 debug 的代码

准备一个 RestController 用于接收请求，最后可以通过本地断点验证是否成功开启了远程 debug

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
@RestController
public class TestController {

    @RequestMapping("/test")
    public Integer test() {
        int i = 0;
        i++;
        i++;
        i++;
        i++;
        i++;
        return i;
    }

}

项目使用 springboot 和 maven 构建，依赖就省略了，使用 springboot 提供的 maven 打包插件，方便我们打包成可运行的 jar。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
<build>
    <plugins>
        <plugin>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-maven-plugin</artifactId>
            <configuration>
                <executable>true</executable>
            </configuration>
        </plugin>
        <plugin>
            <artifactId>maven-compiler-plugin</artifactId>
            <configuration>
                <source>1.8</source>
                <target>1.8</target>
            </configuration>
        </plugin>
    </plugins>
</build>

2 使用 maven 插件打包成 jar

maven 插件

3 准备启动脚本

1
java -jar -agentlib:jdwp=transport=dt_socket,server=y,suspend=n,address=64057 remote-debug-1.0-SNAPSHOT.jar
  1. 使用 java -jar 的方式启动程序，并且添加了一串特殊的参数，这是我们能够开启远程 debug 的关键，以 - 开头的参数是 jvm 的标准启动参数，关于 jvm 启动参数相关的知识可以先去其他博客了解。
  2. -agentlib:libname[=options], 用于装载本地 lib 包。在这条指令中便是加载了 jdwp(Java Debug Wire Protocol) 这个用于远程调试 java 的扩展包。而 transport=dt_socket,server=y,suspend=n,address=64057 这些便是 jdwp 装载时的定制参数，详细的参数作用可以搜索 jdwp 进行了解。我们需要关心的只有 address=64057 这个参数选项，本地调试程序使用 64057 端口与其通信，从而远程调试。

4 配置 IDEA

IDEA 配置

  1. 与脚本中的指令完全一致
  2. 远程 jar 包运行的 host，由于我的 jar 运行在本地，所以使用的是 localhost，一般线上环境自然是修改为线上的地址
  3. 与远程 jar 包进行交互的端口号，idea 会根据指令自动帮我们输入
  4. 选择与远程 jar 包一致的本地代码

请务必保证远程 jar 包的代码与本地代码一致！！！

5 验证

保存第 4 步的配置后，先执行脚本让远程的 jar 包跑起来，再在 IDEA 中运行 remote-debug

运行 remote-jar

如上便代表连接运行成功了

在本地打上断点，访问 localhost:8080/test

远程 debug 信息展示

可以在本地看到堆栈信息，大功告成。一行指令便完成了远程调试。

分享到

浅析项目中的并发 (二)

原创
架构设计

分布式遭遇并发

在前面的章节，并发操作要么发生在单个应用内，一般使用基于 JVM 的 lock 解决并发问题，要么发生在数据库，可以考虑使用数据库层面的锁，而在分布式场景下，需要保证多个应用实例都能够执行同步代码，则需要做一些额外的工作，一个最典型分布式同步方案便是使用分布式锁。

分布式锁由很多种实现，但本质上都是类似的，即依赖于共享组件实现锁的询问和获取，如果说单体式应用中的 Monitor 是由 JVM 提供的，那么分布式下 Monitor 便是由共享组件提供，而典型的共享组件大家其实并不陌生，包括但不限于：Mysql，Redis，Zookeeper。同时他们也代表了三种类型的共享组件：数据库，缓存，分布式协调组件。基于 Consul 的分布式锁，其实和基于 Zookeeper 的分布式锁大同小异，都是借助于分布式协调组件实现锁，大而化之，这三种类型的分布式锁，原理也都差不多，只不过，锁的特性和实现细节有所差异。

Redis 实现分布式锁

定义需求：A 应用需要完成添加库存的操作，部署了 A1，A2，A3 多个实例，实例之间的操作要保证同步。

分析需求：显然，此时依赖于 JVM 的 lock 已经没办法解决问题了，A1 添加锁，无法保证 A2，A3 的同步，这种场景可以考虑使用分布式锁应对。

建立一张 Stock 表，包含 id，number 两个字段，分别让 A1，A2，A3 并发对其操作，保证线程安全。

1
2
3
4
5
6
@Entity
public class Stock {
    @Id
    private String id;
    private Integer number;
}

定义数据库访问层：

1
2
public interface StockRepository extends JpaRepository<Stock,String> {
}

这一节的主角，redis 分布式锁，使用开源的 redis 分布式锁实现：Redisson。

引入 Redisson 依赖：

1
2
3
4
5
<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson</artifactId>
    <version>3.5.4</version>
</dependency>

定义测试类：

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
@RestController
public class StockController {

    @Autowired
    StockRepository stockRepository;

    ExecutorService executorService = Executors.newFixedThreadPool(10);

    @Autowired
    RedissonClient redissonClient;

    final static String id = "1";

    @RequestMapping("/addStock")
    public void addStock() {
        RLock lock = redissonClient.getLock("redisson:lock:stock:" + id);
        for (int i = 0; i < 100; i++) {
            executorService.execute(() -> {
                lock.lock();
                try {
                    Stock stock = stockRepository.findOne(id);
                    stock.setNumber(stock.getNumber() + 1);
                    stockRepository.save(stock);
                } finally {
                    lock.unlock();
                }
            });
        }
    }

}

上述的代码使得并发发生在多个层面。其一，在应用内部，启用线程池完成库存的加 1 操作，本身便是线程不安全的，其二，在多个应用之间，这样的加 1 操作更加是不受约束的。若初始化 id 为 1 的 Stock 数量为 0。分别在本地启用 A1(8080)，A2(8081)，A3(8082) 三个应用，同时并发执行一次 addStock()，若线程安全，必然可以使得数据库中的 Stock 为 300，这便是我们的检测依据。

简单解读下上述的代码，使用 redisson 获取一把 RLock，RLock 是 java.util.concurrent.locks.Lock 接口的实现类，Redisson 帮助我们屏蔽 Redis 分布式锁的实现细节，使用过 java.util.concurrent.locks.Lock 的朋友都会知道下述的代码可以被称得上是同步的起手范式，毕竟这是 Lock 的 java doc 中给出的代码：

1
2
3
4
5
6
7
Lock l = ...;
l.lock();
try {
   // access the resource protected by this lock
} finally {
  l.unlock();
}

而 redissonClient.getLock("redisson:lock:stock:" + id) 则是以 "redisson:lock:stock:" + id 该字符串作痛同步的 Monitor，保证了不同 id 之间是互相不阻塞的。

为了保证发生并发，实际测试中我加入了 Thread.sleep(1000)，使竞争得以发生。测试结果：

测试结果

Redis 分布式锁的确起了作用。

锁的注意点

如果仅仅是实现一个能够用于 demo 的 Redis 分布式锁并不难，但为何大家更偏向于使用开源的实现呢？主要还是可用性和稳定性，we make things work 是我在写博客，写代码时牢记在脑海中的，如果真的要细究如何自己实现一个分布式锁，或者平时使用锁保证并发，需要有哪些注意点呢？列举几点：阻塞，超时时间，可重入，可用性，其他特性。

阻塞

意味着各个操作之间的等待，A1 正在执行增加库存时，A1 其他的线程被阻塞，A2，A3 中所有的线程被阻塞，在 Redis 中可以使用轮询策略以及 redis 底层提供的 CAS 原语 (如 setnx) 来实现。（初学者可以理解为：在 redis 中设置一个 key，想要执行 lock 代码时先询问是否有该 key，如果有则代表其他线程在执行过程中，若没有，则设置该 key，并且执行代码，执行完毕，释放 key，而 setnx 保证操作的原子性）

超时时间

在特殊情况，可能会导致锁无法被释放，如死锁，死循环等等意料之外的情况，锁超时时间的设置是有必要的，一个很直观的想法是给 key 设置过期时间即可。

如在 Redisson 中，lock 提供了一个重载方法 lock(long t, TimeUnit timeUnit); 可以自定义过期时间。

可重入

这个特性很容易被忽视，可重入其实并不难理解，顾名思义，一个方法在调用过程中是否可以被再次调用。实现可重入需要满足三个特性：

  1. 可以在执行的过程中可以被打断；
  2. 被打断之后，在该函数一次调用执行完之前，可以再次被调用（或进入，reentered)。
  3. 再次调用执行完之后，被打断的上次调用可以继续恢复执行，并正确执行。

比如下述的代码引用了全局变量，便是不可重入的：

1
2
3
4
5
6
7
8
int t;

void swap(int x, int y) {
    t = x;
    x = y;
    y = t;
    System.out.println("x is" + x + "y is" + y);
}

一个更加直观的例子便是，同一个线程中，某个方法的递归调用不应该被阻塞，所以如果要实现这个特性，简单的使用某个 key 作为 Monitor 是欠妥的，可以加入线程编号，来保证可重入。

使用可重入分布式锁的来测试计算斐波那契数列（只是为了验证可重入性）：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
@RequestMapping("testReentrant")
public void ReentrantLock() {
    RLock lock = redissonClient.getLock("fibonacci");
    lock.lock();
    try {
        int result = fibonacci(10);
        System.out.println(result);
    } finally {
        lock.unlock();
    }
}

int fibonacci(int n) {
    RLock lock = redissonClient.getLock("fibonacci");
    try {
        if (n <= 1) return n;
        else
            return fibonacci(n - 1) + fibonacci(n - 2);
    } finally {
        lock.unlock();
    }
}

最终输出：55，可以发现，只要是在同一线程之内，无论是递归调用还是外部加锁 (同一把锁)，都不会造成死锁。

可用性

借助于第三方中间件实现的分布式锁，都有这个问题，中间件挂了，会导致锁不可用，所以需要保证锁的高可用，这就需要保证中间件的可用性，如 redis 可以使用哨兵 + 集群，保证了中间件的可用性，便保证了锁的可用性、

其他特性

除了可重入锁，锁的分类还有很多，在分布式下也同样可以实现，包括但不限于：公平锁，联锁，信号量，读写锁。Redisson 也都提供了相关的实现类，其他的特性如并发容器等可以参考官方文档。

新手遭遇并发

基本算是把项目中遇到的并发过了一遍了，案例其实很多，再简单罗列下一些新手可能会遇到的问题。

使用了线程安全的容器就是线程安全了吗？很多新手误以为使用了并发容器如：concurrentHashMap 就万事大吉了，却不知道，一知半解的隐患可能比全然不懂更大。来看下面的代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
public class ConcurrentHashMapTest {

    static Map<String, Integer> counter = new ConcurrentHashMap();

    public static void main(String[] args) throws InterruptedException {
        counter.put("stock1", 0);
        ExecutorService executorService = Executors.newFixedThreadPool(10);
        CountDownLatch countDownLatch = new CountDownLatch(100);
        for (int i = 0; i < 100; i++) {
            executorService.execute(new Runnable() {
                @Override
                public void run() {
                    counter.put("stock1", counter.get("stock1") + 1);
                    countDownLatch.countDown();
                }
            });
        }
        countDownLatch.await();
        System.out.println("result is" + counter.get("stock1"));
    }
}

counter.put("stock1", counter.get("stock1") + 1) 并不是原子操作，并发容器保证的是单步操作的线程安全特性，这一点往往初级程序员特别容易忽视。

总结

项目中的并发场景是非常多的，而根据场景不同，同一个场景下的业务需求不同，以及数据量，访问量的不同，都会影响到锁的使用，架构中经常被提到的一句话是：业务决定架构，放到并发中也同样适用：业务决定控制并发的手段，如本文未涉及的队列的使用，本质上是化并发为串行，也解决了并发问题，都是控制的手段。了解锁的使用很简单，但如果使用，在什么场景下使用什么样的锁，这才是价值所在。

同一个线程之间的递归调用不应该被阻塞，所以如果要实现这个特性，简单的使用某个 key 作为 Monitor 是欠妥的，可以加入线程编号，来保证可重入。

分享到

浅析项目中的并发 (一)

原创
架构设计

前言

控制并发的方法很多，从最基础的 synchronized，juc 中的 lock，到数据库的行级锁，乐观锁，悲观锁，再到中间件级别的 redis，zookeeper 分布式锁。特别是初级程序员，对于所谓的锁一直都是听的比用的多，第一篇文章不深入探讨并发，更多的是一个入门介绍，适合于初学者，主题是“根据并发出现的具体业务场景，使用合理的控制并发手段”。

什么是并发

由一个大家都了解的例子引入我们今天的主题：并发

类共享变量遭遇并发

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
public class Demo {

    public Integer count = 0;

    public static void main(String[] args) {
        final Demo demo = new Demo();
        Executor executor = Executors.newFixedThreadPool(10);
        for(int i=0;i<1000;i++){
            executor.execute(new Runnable() {
                @Override
                public void run() {
                    demo.count++;
                }
            });
        }
        try {
            Thread.sleep(5000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        System.out.println("final count value:"+demo1.count);
    }
}

final count value:973

本例中创建了一个初始化时具有 10 个线程的线程池，多线程对类变量 count 进行自增操作。这个过程中，自增操作并不是线程安全的，happens-before 原则并不会保障多个线程执行的先后顺序，导致了最终结果并不是想要的 1000

下面，我们把并发中的共享资源从类变量转移到数据库中。

充血模型遭遇并发

1
2
3
4
5
6
7
8
9
10
11
12
13
14
@Component
public class Demo2 {

    @Autowired
    TestNumDao testNumDao;

    @Transactional
    public void test(){
        TestNum testNum = testNumDao.findOne("1");
        testNum.setCount(testNum.getCount()+1);
        testNumDao.save(testNum);
    }

}

依旧使用多线程，对数据库中的记录进行 +1 操作

Demo2 demo2;

public String test(){
    Executor executor = Executors.newFixedThreadPool(10);
    for(int i=0;i<1000;i++){
        executor.execute(new Runnable() {
            @Override
            public void run() {
                demo2.test();
            }
        });
    }
    return "test";
}

数据库的记录

1
2
id	| count
1	| 344

初窥门径的程序员会认为事务最基本的 ACID 中便包含了原子性，但是事务的原子性和今天所讲的并发中的原子操作仅仅是名词上有点类似。而有点经验的程序员都能知道这中间发生了什么，这只是暴露了项目中并发问题的冰山一角，千万不要认为上面的代码没有必要列举出来，我在实际项目开发中，曾经见到有多年工作经验的程序员仍然写出了类似于上述会出现并发问题的代码。

贫血模型遭遇并发

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
@RequestMapping("testSql")
    @ResponseBody
    public String testSql() throws InterruptedException {
        final CountDownLatch countDownLatch = new CountDownLatch(1000);
        long start = System.currentTimeMillis();
        Executor executor = Executors.newFixedThreadPool(10);
        for(int i=0;i<1000;i++){
            executor.execute(new Runnable() {
                @Override
                public void run() {
                    jdbcTemplate.execute("update test_num set count = count + 1 where id ='1'");
                    countDownLatch.countDown();
                }
            });
        }
        countDownLatch.await();
        long costTime =System.currentTimeMillis() - start;
        System.out.println("共花费："+costTime+"s");
        return "testSql";
    }

数据库结果： count ： 1000 达到了预期效果
这个例子我顺便记录了耗时, 控制台打印: 共花费：113 ms
简单对比一下二，三两个例子，都是想对数据库的 count 进行 +1 操作，唯一的区别就是， 后者的 +1 计算发生在数据库，而前者的计算依赖于事先查出来的值，并且计算发生在程序的内存中 。而现在大部分的 ORM 框架，导致了写充血模型的程序员变多，不注意并发的话，就会出现问题。下面我们来看看具体的业务场景。

业务场景

  1. 修改个人信息
  2. 修改商品信息
  3. 扣除账户余额，扣减库存

业务场景分析

第一个场景，互联网如此众多的用户修改个人信息，这算不算并发？答案是：算也不算。
算，从程序员角度来看，每一个用户请求进来，都是调用的同一个修改入口，具体一点，就是映射到 controller 层的同一个 requestMapping，所以一定是并发的。
不算，虽然程序是并发的，但是从用户角度来分析，每个人只可以修改自己的信息，所以，不同用户的操作其实是隔离的，所以不算“并发”。这也是为什么很多开发者，在日常开发中一直不注意并发控制，却也没有发生太大问题的原因，大多数初级程序员开发的还都是 CRM，OA，CMS 系统。

回到我们的并发，第一种业务场景，是可以使用如上模式的，对于一条用户数据的修改，我们允许程序员读取数据到内存中，内存计算修改（耗时操作），提交更改，提交事务。

1
2
3
4
5
6
7
//Transaction start
User user = userDao.findById("1");
user.setName("newName");
user.setAge(user.getAge()+1);
...// 其他耗时操作
userDao.save(user);
//Transaction commit

这个场景变现为：几乎不存在并发，不需要控制，场景乐观。

为了严谨，也可以选择控制并发，但我觉得这需要交给写这段代码的同事，让他自由发挥。

第二个场景已经有所不同了，同样是修改一个记录，但是系统中可能有多个操作员来维护，此时，商品数据表现为一个共享数据，所以存在微弱的并发，通常表现为数据的脏读，例如操作员 A，B 同时对一个商品信息维护，我们希望只能有一个操作员修改成功，另外一个操作员得到错误提示（该商品信息已经发生变化），否则，两个人都以为自己修改成功了，但是其实只有一个人完成了操作，另一个人的操作被覆盖了。

这个场景表现为：存在并发，需要控制，允许失败，场景乐观。

通常我建议这种场景使用乐观锁，即在商品属性添加一个 version 字段标记修改的版本，这样两个操作员拿到同一个版本号，第一个操作员修改成功后版本号变化，另一个操作员的修改就会失败了。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
class Goods{
	@Version
	int version;
}

//Transaction start
try{
	Goods goods = goodsDao.findById("1");
	goods.setName("newName");
	goods.setPrice(goods.getPrice()+100.00);
	...// 其他耗时操作
	goodsDao.save(goods);
}catch(org.hibernate.StaleObjectStateException e){
	// 返回给前台
}

//Transaction commit

springdata 配合 jpa 可以自动捕获 version 异常，也可以自动手动对比。

第三个场景
这个场景表现为：存在频繁的并发，需要控制，不允许失败，场景悲观。

强调一下，本例不应该使用在项目中，只是为了举例而设置的一个场景，因为这种贫血模型无法满足复杂的业务场景，而且依靠单机事务来保证一致性，并发性能和可扩展性能不好。

一个简易的秒杀场景，大量请求在短时间涌入，是不可能像第二种场景一样，100 个并发请求，一个成功，其他 99 个全部异常的。

设计方案应该达到的效果是：有足够库存时，允许并发，库存到 0 时，之后的请求全部失败；有足够金额时，允许并发，金额不够支付时立刻告知余额不足。

可以利用数据库的行级锁，
update set balance = balance - money where userId = ? and balance >= money;
update stock = stock - number where goodsId = ? and stock >= number ; 然后在后台 查看返回值是否影响行数为 1，判断请求是否成功，利用数据库保证并发。

需要补充一点，我这里所讲的秒杀，并不是指双 11 那种级别的秒杀，那需要多层架构去控制并发，前端拦截，负载均衡…. 不能仅仅依赖于数据库的，会导致严重的性能问题。为了留一下一个直观的感受，这里对比一下 oracle，mysql 的两个主流存储引擎：innodb，myisam 的性能问题。

1
2
3
4
5
6
oracle:
10000 个线程共计 1000000 次并发请求：共花费：101017 ms =>101s
innodb:
10000 个线程共计 1000000 次并发请求：共花费：550330 ms =>550s
myisam:
10000 个线程共计 1000000 次并发请求：共花费：75802 ms =>75s

可见，如果真正有大量请求到达数据库，光是依靠数据库解决并发是不现实的，所以仅仅只用数据库来做保障而不是完全依赖。需要根据业务场景选择合适的控制并发手段。

分享到