Spring Boot 中各种缓存的使用_replicateasynchronously

进程内缓存的使用与Cache注解详解

User实体的定义

@Entity
@Data
@NoArgsConstructor
public class User {
    @Id
    @GeneratedValue
    private Long id;
    private String name;
    private Integer age;
    public User(String name, Integer age) {
        this.name = name;
        this.age = age;
    }
 }
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

• @Entity: 代表此类映射为数据库的表结构
• @Id: 指定一个主键
• @GeneratedValue: 配置主键相关信息
  • Table: 使用一个特定的数据库表来保存主键
  • IDENTITY: 数据库自动生成
  • AUTO: 主键由程序控制，默认值
  • SEQUENCE: 通过数据库的序列产生主键, MYSQL不支持，部分数据库(Oracle,PostgreSQL,DB2)支持序列对象

User实体的数据访问实现

public interface UserRepository extends JpaRepository<User, Long> {
    User findByName(String name);
    User findByNameAndAge(String name, Integer age);
    @Query("from User u where u.name=:name")
    User findUser(@Param("name") String name);
}
1
2
3
4
5
6

为了更好的理解缓存，我们先对该工程做一些简单的改造。

• application.properties文件中新增spring.jpa.show-sql=true，开启hibernate对sql语句的打印。如果是1.x版本，使用spring.jpa.properties.hibernate.show_sql=true参数。
• 修改单元测试类，插入User表一条用户名为AAA，年龄为10的数据。并通过findByName函数完成两次查询，具体代码如下：

@RunWith(SpringRunner.class)
@SpringBootTest
public class Chapter51ApplicationTests {
    @Autowired
    private UserRepository userRepository;
    @Test
    public void test() throws Exception {
        // 创建1条记录
        userRepository.save(new User("AAA", 10));
        User u1 = userRepository.findByName("AAA");
        System.out.println("第一次查询：" + u1.getAge());
        User u2 = userRepository.findByName("AAA");
        System.out.println("第二次查询：" + u2.getAge());
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

在没有加入缓存之前，我们可以先执行一下这个案例，可以看到如下的日志：

Hibernate: select user0_.id as id1_0_, user0_.age as age2_0_, user0_.name as name3_0_ from user user0_ where user0_.name=?
第一次查询：10
Hibernate: select user0_.id as id1_0_, user0_.age as age2_0_, user0_.name as name3_0_ from user user0_ where user0_.name=?
第二次查询：10
1
2
3
4

两次findByName查询都执行了两次SQL，都是对MySQL数据库的查询。

引入缓存

第一步：在pom.xml中引入cache依赖，添加如下内容：

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

第二步：在Spring Boot主类中增加@EnableCaching注解开启缓存功能，如下：

@EnableCaching
@SpringBootApplication
public class Chapter51Application {
	public static void main(String[] args) {
		SpringApplication.run(Chapter51Application.class, args);
	}
}
1
2
3
4
5
6
7

第三步：在数据访问接口中，增加缓存配置注解，如：

@CacheConfig(cacheNames = "users")
public interface UserRepository extends JpaRepository<User, Long> {
    @Cacheable
    User findByName(String name);
}
1
2
3
4
5

第四步：再来执行以下单元测试，可以在控制台中输出了下面的内容

Hibernate: insert into user (age, name, id) values (?, ?, ?)
Hibernate: select user0_.id as id1_0_, user0_.age as age2_0_, user0_.name as name3_0_ from user user0_ where user0_.name=?
第一次查询：10
第二次查询：10
1
2
3
4

到这里，我们可以看到，在调用第二次findByName函数时，没有再执行select语句，也就直接减少了一次数据库的读取操作。
为了可以更好的观察，缓存的存储，我们可以在单元测试中注入CacheManager。

@Autowired
private CacheManager cacheManager;
1
2

使用debug模式运行单元测试，观察CacheManager中的缓存集users以及其中的User对象的缓存加深理解。

可以看到，在第一次调用findByName函数之后，CacheManager将这个查询结果保存了下来，所以在第二次访问的时候，就能匹配上而不需要再访问数据库了。

Cache配置注解详解

回过头来我们再来看这里使用到的两个注解分别作了什么事情：

• @CacheConfig：主要用于配置该类中会用到的一些共用的缓存配置。在这里@CacheConfig(cacheNames = "users")：配置了该数据访问对象中返回的内容将存储于名为users的缓存对象中，我们也可以不使用该注解，直接通过@Cacheable自己配置缓存集的名字来定义。
• @Cacheable：配置了findByName函数的返回值将被加入缓存。同时在查询时，会先从缓存中获取，若不存在才再发起对数据库的访问。该注解主要有下面几个参数：
  • value、cacheNames：两个等同的参数（cacheNames为Spring 4新增，作为value的别名），用于指定缓存存储的集合名。由于Spring 4中新增了@CacheConfig，因此在Spring 3中原本必须有的value属性，也成为非必需项了
  • key：缓存对象存储在Map集合中的key值，非必需，缺省按照函数的所有参数组合作为key值，若自己配置需使用SpEL表达式，比如：@Cacheable(key = “#p0”)：使用函数第一个参数作为缓存的key值，更多关于SpEL表达式的详细内容可参考官方文档
  • condition：缓存对象的条件，非必需，也需使用SpEL表达式，只有满足表达式条件的内容才会被缓存，比如：@Cacheable(key = "#p0", condition = "#p0.length() < 3")，表示只有当第一个参数的长度小于3的时候才会被缓存，若做此配置上面的AAA用户就不会被缓存，读者可自行实验尝试。
  • unless：另外一个缓存条件参数，非必需，需使用SpEL表达式。它不同于condition参数的地方在于它的判断时机，该条件是在函数被调用之后才做判断的，所以它可以通过对result进行判断。
  • keyGenerator：用于指定key生成器，非必需。若需要指定一个自定义的key生成器，我们需要去实现org.springframework.cache.interceptor.KeyGenerator接口，并使用该参数来指定。需要注意的是：该参数与key是互斥的
  • cacheManager：用于指定使用哪个缓存管理器，非必需。只有当有多个时才需要使用
  • cacheResolver：用于指定使用那个缓存解析器，非必需。需通过org.springframework.cache.interceptor.CacheResolver接口来实现自己的缓存解析器，并用该参数指定。

除了这里用到的两个注解之外，还有下面几个核心注解：

• @CachePut：配置于函数上，能够根据参数定义条件来进行缓存，它与@Cacheable不同的是，它每次都会真实调用函数，所以主要用于数据新增和修改操作上。它的参数与@Cacheable类似，具体功能可参考上面对@Cacheable参数的解析
• @CacheEvict：配置于函数上，通常用在删除方法上，用来从缓存中移除相应数据。除了同@Cacheable一样的参数之外，它还有下面两个参数：
  • allEntries：非必需，默认为false。当为true时，会移除所有数据
  • beforeInvocation：非必需，默认为false，会在调用方法之后移除数据。当为true时，会在调用方法之前移除数据。

EhCache缓存的使用

Spring Boot中整合EhCache

在Spring Boot中通过@EnableCaching注解自动化配置合适的缓存管理器（CacheManager），Spring Boot根据下面的顺序去侦测缓存提供者：

• Generic、JCache (JSR-107) (EhCache 3, Hazelcast, Infinispan, and others)、EhCache 2.x、Hazelcast、Infinispan、Couchbase、Redis、Caffeine、Simple
• 除了按顺序侦测外，我们也可以通过配置属性spring.cache.type来强制指定。
• 当我们不指定具体其他第三方实现的时候，Spring Boot的Cache模块会使用ConcurrentHashMap来存储。

使用EhCache

User实体的定义

@Entity
@Data
@NoArgsConstructor
public class User {
    @Id
    @GeneratedValue
    private Long id;
    private String name;
    private Integer age;
    public User(String name, Integer age) {
        this.name = name;
        this.age = age;
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14

User实体的数据访问实现

@CacheConfig(cacheNames = "users")
public interface UserRepository extends JpaRepository<User, Long> {
    @Cacheable
    User findByName(String name);
}
1
2
3
4
5

测试验证用例：

@Slf4j
@RunWith(SpringRunner.class)
@SpringBootTest
public class Chapter51ApplicationTests {
    @Autowired
    private UserRepository userRepository;
    @Autowired
    private CacheManager cacheManager;
    @Test
    public void test() throws Exception {
        // 创建1条记录
        userRepository.save(new User("AAA", 10));
        User u1 = userRepository.findByName("AAA");
        System.out.println("第一次查询：" + u1.getAge());
        User u2 = userRepository.findByName("AAA");
        System.out.println("第二次查询：" + u2.getAge());
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

把上面的缓存应用改成使用ehcache缓存管理
第一步：在pom.xml中引入ehcache依赖

<dependency>
    <groupId>net.sf.ehcache</groupId>
    <artifactId>ehcache</artifactId>
</dependency>
1
2
3
4

在Spring Boot的parent管理下，不需要指定具体版本，会自动采用Spring Boot中指定的版本号。

第二步：在src/main/resources目录下创建：ehcache.xml

<ehcache xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:noNamespaceSchemaLocation="ehcache.xsd">
    <cache name="users"
           maxEntriesLocalHeap="200"
           timeToLiveSeconds="600">
    </cache>
</ehcache>
1
2
3
4
5
6
7

完成上面的配置之后，再通过debug模式运行单元测试，观察此时CacheManager已经是EhCacheManager实例，说明EhCache开启成功了。或者在测试用例中加一句CacheManager的输出，比如：

@Autowired
private CacheManager cacheManager;
@Test
public void test() throws Exception {
    System.out.println("CacheManager type : " + cacheManager.getClass());
    userRepository.save(new User("AAA", 10));
    User u1 = userRepository.findByName("AAA");
    System.out.println("第一次查询：" + u1.getAge());
    User u2 = userRepository.findByName("AAA");
    System.out.println("第二次查询：" + u2.getAge());
}
1
2
3
4
5
6
7
8
9
10
11

执行测试输出可以得到：

CacheManager type : class org.springframework.cache.ehcache.EhCacheCacheManager
Hibernate: select next_val as id_val from hibernate_sequence for update
Hibernate: update hibernate_sequence set next_val= ? where next_val=?
Hibernate: insert into user (age, name, id) values (?, ?, ?)
2020-07-14 18:09:28.465  INFO 58538 --- [           main] o.h.h.i.QueryTranslatorFactoryInitiator  : HHH000397: Using ASTQueryTranslatorFactory
Hibernate: select user0_.id as id1_0_, user0_.age as age2_0_, user0_.name as name3_0_ from user user0_ where user0_.name=?
第一次查询：10
第二次查询：10
 
1
2
3
4
5
6
7
8
9

使用EhCache缓存集群

如何来组建进程内缓存EnCache的集群以及配置他们的同步策略。
基于上面的demo进行改造。
第一步：为需要同步的缓存对象实现Serializable接口

@Entity
@Data
@NoArgsConstructor
public class User implements Serializable {
    @Id
    @GeneratedValue
    private Long id;
    private String name;
    private Integer age;
    public User(String name, Integer age) {
        this.name = name;
        this.age = age;
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14

注意：如果没有做这一步，后续缓存集群通过过程中，因为要传输User对象，会导致序列化与反序列化相关的异常

第二步：重新组织ehcache的配置文件。我们尝试手工组建集群的方式，不同实例在网络相关配置上会产生不同的配置信息，所以我们建立不同的配置文件给不同的实例使用。比如下面这样：

实例1，使用ehcache-1.xml

<ehcache xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:noNamespaceSchemaLocation="ehcache.xsd">
    <cache name="users"
           maxEntriesLocalHeap="200"
           timeToLiveSeconds="600">
        <cacheEventListenerFactory
                class="net.sf.ehcache.distribution.RMICacheReplicatorFactory"
                properties="replicateAsynchronously=true,
            replicatePuts=true,
            replicateUpdates=true,
            replicateUpdatesViaCopy=false,
            replicateRemovals=true "/>
    </cache>
    <cacheManagerPeerProviderFactory
            class="net.sf.ehcache.distribution.RMICacheManagerPeerProviderFactory"
            properties="hostName=10.10.0.100,
                        port=40001,
                        socketTimeoutMillis=2000,
                        peerDiscovery=manual,
                        rmiUrls=//10.10.0.101:40001/users" />
</ehcache>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

实例2，使用ehcache-2.xml

<ehcache xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:noNamespaceSchemaLocation="ehcache.xsd">
    <cache name="users"
           maxEntriesLocalHeap="200"
           timeToLiveSeconds="600">
        <cacheEventListenerFactory
                class="net.sf.ehcache.distribution.RMICacheReplicatorFactory"
                properties="replicateAsynchronously=true,
            replicatePuts=true,
            replicateUpdates=true,
            replicateUpdatesViaCopy=false,
            replicateRemovals=true "/>
    </cache>
    <cacheManagerPeerProviderFactory
            class="net.sf.ehcache.distribution.RMICacheManagerPeerProviderFactory"
            properties="hostName=10.10.0.101,
                        port=40001,
                        socketTimeoutMillis=2000,
                        peerDiscovery=manual,
                        rmiUrls=//10.10.0.100:40001/users" />
</ehcache>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

配置说明：

• cache标签中定义名为users的缓存，这里我们增加了一个子标签定义cacheEventListenerFactory，这个标签主要用来定义缓存事件监听的处理策略，它有以下这些参数用来设置缓存的同步策略：

  • replicatePuts：当一个新元素增加到缓存中的时候是否要复制到其他的peers。默认是true。
  • replicateUpdates：当一个已经在缓存中存在的元素被覆盖时是否要进行复制。默认是true。
  • replicateRemovals：当元素移除的时候是否进行复制。默认是true。
  • replicateAsynchronously：复制方式是异步的指定为true时;同步的指定为false时。默认是true。
  • replicatePutsViaCopy：当一个新增元素被拷贝到其他的cache中时是否进行复制;指定为true时为复制，默认是true。
  • replicateUpdatesViaCopy：当一个元素被拷贝到其他的cache中时是否进行复制;指定为true时为复制，默认是true。

• 新增了一个cacheManagerPeerProviderFactory标签的配置，用来指定组建的集群信息和要同步的缓存信息，其中：

  • hostName：是当前实例的主机名
  • port：当前实例用来同步缓存的端口号
  • socketTimeoutMillis：同步缓存的Socket超时时间
  • peerDiscovery：集群节点的发现模式，有手工与自动两种，这里采用了手工指定的方式
  • rmiUrls：当peerDiscovery设置为manual的时候，用来指定需要同步的缓存节点，如果存在多个用|连接

第三步：打包部署与启动。打包没啥大问题，主要缓存配置内容存在一定差异，所以在指定节点的模式下，需要单独拿出来，然后使用启动参数来控制读取不同的配置文件。比如这样：

-Dspring.cache.ehcache.config=classpath:ehcache-1.xml
-Dspring.cache.ehcache.config=classpath:ehcache-2.xml
1
2

第四步：实现几个接口用来验证缓存的同步效果

@RestController
static class HelloController {
    @Autowired
    private UserRepository userRepository;
    @GetMapping("/create")    
    public void create() {
        userRepository.save(new User("AAA", 10));
    }
    @GetMapping("/find")
    public User find() {
        User u1 = userRepository.findByName("AAA");
        System.out.println("查询AAA用户：" + u1.getAge());
        return u1;
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

验证逻辑：

• 启动通过第三步说的命令参数，启动两个实例
• 调用实例1的/create接口，创建一条数据
• 调用实例1的/find接口，实例1缓存User，同时同步缓存信息给实例2，在实例1中会存在SQL查询语句
• 调用实例2的/find接口，由于缓存集群同步了User的信息，所以在实例2中的这次查询也不会出现SQL语句

使用集中式缓存Redis

上述两种进程内缓存的用法，包括Spring Boot默认使用的ConcurrentMap缓存以及缓存框架EhCache。虽然EhCache已经能够适用很多应用场景，但是由于EhCache是进程内的缓存框架，在集群模式下时，各应用服务器之间的缓存都是独立的，因此在不同服务器的进程间会存在缓存不一致的情况。即使EhCache提供了集群环境下的缓存同步策略，但是同步依然是需要一定的时间，短暂的缓存不一致依然存在。

在一些要求高一致性（任何数据变化都能及时的被查询到）的系统和应用中，就不能再使用EhCache来解决了，这个时候使用集中式缓存就可以很好的解决缓存数据的一致性问题。如何在Spring Boot的缓存支持中使用Redis实现数据缓存。

基于上面的demo进行改造。
第一步：增加相关依赖：

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<dependency>
    <groupId>org.apache.commons</groupId>
    <artifactId>commons-pool2</artifactId>
</dependency>
1
2
3
4
5
6
7
8

需要注意的是：在Spring Boot 1.x的早期版本中，该依赖的名称为spring-boot-starter-redis

第二步：配置文件中增加配置信息，以本地运行为例，比如：

spring.redis.host=localhost
spring.redis.port=6379
spring.redis.lettuce.pool.max-idle=8
spring.redis.lettuce.pool.max-active=8
spring.redis.lettuce.pool.max-wait=-1ms
spring.redis.lettuce.pool.min-idle=0
spring.redis.lettuce.shutdown-timeout=100ms
1
2
3
4
5
6
7

关于连接池的配置，注意几点：

Redis的连接池配置在1.x版本中前缀为spring.redis.pool与Spring Boot 2.x有所不同。
在1.x版本中采用jedis作为连接池，而在2.x版本中采用了lettuce作为连接池
以上配置均为默认值，实际上生产需进一步根据部署情况与业务要求做适当修改.

再来试试单元测试：

@Slf4j
@RunWith(SpringRunner.class)
@SpringBootTest
public class Chapter54ApplicationTests {
    @Autowired
    private UserRepository userRepository;
    @Autowired
    private CacheManager cacheManager;
    @Test
    public void test() throws Exception {
        System.out.println("CacheManager type : " + cacheManager.getClass());
        // 创建1条记录
        userRepository.save(new User("AAA", 10));
        User u1 = userRepository.findByName("AAA");
        System.out.println("第一次查询：" + u1.getAge());
        User u2 = userRepository.findByName("AAA");
        System.out.println("第二次查询：" + u2.getAge());
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

测试结果：

CacheManager type : class org.springframework.data.redis.cache.RedisCacheManager
Hibernate: select next_val as id_val from hibernate_sequence for update
Hibernate: update hibernate_sequence set next_val= ? where next_val=?
Hibernate: insert into user (age, name, id) values (?, ?, ?)
2020-08-12 16:25:26.954  INFO 68282 --- [           main] io.lettuce.core.EpollProvider            : Starting without optional epoll library
2020-08-12 16:25:26.955  INFO 68282 --- [           main] io.lettuce.core.KqueueProvider           : Starting without optional kqueue library
Hibernate: select user0_.id as id1_0_, user0_.age as age2_0_, user0_.name as name3_0_ from user user0_ where user0_.name=?
第一次查询：10
第二次查询：10
1
2
3
4
5
6
7
8
9

可以看到：

• 第一行输出的CacheManager type为org.springframework.data.redis.cache.RedisCacheManager，而不是EhCacheCacheManager了
• 第二次查询的时候，没有输出SQL语句，所以是走的缓存获取
  整合成功！

使用Redis的发布订阅功能

观察者模式

发布订阅模式
发布订阅模式在两个角色中间是一个中间角色来过渡的，发布者并不直接与订阅者产生交互

在Spring Boot应用中，通过接口的方式实现一个消息发布者的角色，然后再写一个Service来实现消息的订阅（把接口传过来的消息内容打印处理）。

第一步：创建一个基础的Spring Boot应用

第二步：pom.xml中加入必须的几个依赖：

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<dependency>
    <groupId>org.apache.commons</groupId>
    <artifactId>commons-pool2</artifactId>
</dependency>
1
2
3
4
5
6
7
8
9
10
11
12

第三步：创建一个接口，用来发送消息。

@SpringBootApplication
public class Chapter55Application {
    private static String CHANNEL = "didispace";
    public static void main(String[] args) {
        SpringApplication.run(Chapter55Application.class, args);
    }
    @RestController
    static class RedisController {
        private RedisTemplate<String, String> redisTemplate;
        public RedisController(RedisTemplate<String, String> redisTemplate) {
            this.redisTemplate = redisTemplate;
        }
        @GetMapping("/publish")
        public void publish(@RequestParam String message) {
            // 发送消息
            redisTemplate.convertAndSend(CHANNEL, message);
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

这里为了简单实现，公用CHANNEL名称字段，我都写在了应用主类里。

第四步：继续应用主类里实现消息订阅，打接收到的消息打印处理

@Slf4j
@Service
static class MessageSubscriber {
    public MessageSubscriber(RedisTemplate redisTemplate) {
        RedisConnection redisConnection = redisTemplate.getConnectionFactory().getConnection();
        redisConnection.subscribe(new MessageListener() {
            @Override
            public void onMessage(Message message, byte[] bytes) {
                // 收到消息的处理逻辑
                log.info("Receive message : " + message);
            }
        }, CHANNEL.getBytes(StandardCharsets.UTF_8));
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14

第五步：验证结果
启动应用Spring Boot主类
通过curl或其他工具调用接口curl localhost:8080/publish?message=hello
观察控制台，可以看到打印了收到的message参数

INFO 34351 --- [ioEventLoop-4-2] .c.Chapter55Application$MessageSubscribe : Receive message : hello
1