spring boot: 用redis的消息訂閱功能更新應用內的caffeine本地緩存(spring boot 2.3.2)

一，為什么要更新caffeine緩存？

1，caffeine緩存的優點和缺點

生產環境中，caffeine緩存是我們在應用中使用的本地緩存，

它的優勢在於存在於應用內，訪問速度最快，通常都不到1ms就能做出響應，

缺點在於不方便管理，因為存在於多台負載均衡的web服務器上，

很難象管理redis緩存一樣對它做出更新、刪除。

 

2,通常我們會把caffeine緩存的時間設置為5分鍾或10分鍾，

   但當有大型促銷活動開始時，如果緩存還沒過期，

   則web服務顯示的數據不會立刻得到更新，

   我們如何更新多台web服務器的的應用內緩存？

   使用redis的消息訂閱是解決方法之一，

   我們從后台發送一條消息到redis,

   訂閱了redis的web服務收到消息可以對緩存進行處理，

   這樣實現對多台web服務器上的緩存的更新

 

3, 生產環境中通常會使用多級緩存，

    我們在更新caffeine緩存時，

    也不要去訪問數據庫，避免導致對數據庫的並發訪問，

    而是更新完redis后，

    本地緩存從redis獲取數據，

    而幾百幾千數量級的並發訪問對於redis來說壓力很小

 

說明：劉宏締的架構森林是一個專注架構的博客，地址：https://www.cnblogs.com/architectforest

         對應的源碼可以訪問這里獲取： https://github.com/liuhongdi/

說明：作者:劉宏締 郵箱: [e-mail]

 

二，演示項目的相關信息

1，項目地址

https://github.com/liuhongdi/redispubsub

 

2,項目功能說明：

   web服務通過訂閱redis的消息，

   實現對緩存的更新/刪除/清除

 

3，項目結構：如圖:

 

 

三，配置文件說明

1,pom.xml

        <!--redis begin-->
        <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>
        <dependency>
            <groupId>com.fasterxml.jackson.core</groupId>
            <artifactId>jackson-core</artifactId>
            <version>2.11.1</version>
        </dependency>
        <dependency>
            <groupId>com.fasterxml.jackson.core</groupId>
            <artifactId>jackson-databind</artifactId>
            <version>2.11.1</version>
        </dependency>
        <!--redis   end-->

        <!-- fastjson begin-->
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.2.73</version>
        </dependency>
        <!-- fastjson   end-->

        <!--local cache begin-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-cache</artifactId>
        </dependency>
        <dependency>
            <groupId>com.github.ben-manes.caffeine</groupId>
            <artifactId>caffeine</artifactId>
            <version>2.8.5</version>
        </dependency>
        <!--local cache   end-->

        <!--mybatis begin-->
        <dependency>
            <groupId>org.mybatis.spring.boot</groupId>
            <artifactId>mybatis-spring-boot-starter</artifactId>
            <version>2.1.3</version>
        </dependency>
        <!--mybatis end-->

        <!--mysql begin-->
        <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
            <scope>runtime</scope>
        </dependency>
        <!--mysql end-->

2,application.properties

#error
server.error.include-stacktrace=always
#errorlog
logging.level.org.springframework.web=trace

#redis1
spring.redis1.host=127.0.0.1
spring.redis1.port=6379
spring.redis1.password=lhddemo
spring.redis1.database=0

spring.redis1.lettuce.pool.max-active=32
spring.redis1.lettuce.pool.max-wait=300
spring.redis1.lettuce.pool.max-idle=16
spring.redis1.lettuce.pool.min-idle=8

spring.redis1.enabled=1

#mysql
spring.datasource.url=jdbc:mysql://localhost:3306/store?characterEncoding=utf8&useSSL=false
spring.datasource.username=root
spring.datasource.password=lhddemo
spring.datasource.driver-class-name=com.mysql.cj.jdbc.Driver

#mybatis
mybatis.mapper-locations=classpath:/mapper/*Mapper.xml
mybatis.type-aliases-package=com.example.demo.mapper
mybatis.configuration.log-impl=org.apache.ibatis.logging.stdout.StdOutImpl

#profile
spring.profiles.active=cacheenable

3,goods數據表的建表sql:

CREATE TABLE `goods` (
 `goodsId` int(11) NOT NULL AUTO_INCREMENT COMMENT 'id',
 `goodsName` varchar(500) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci NOT NULL DEFAULT '' COMMENT 'name',
 `subject` varchar(200) NOT NULL DEFAULT '' COMMENT '標題',
 `price` decimal(15,2) NOT NULL DEFAULT '0.00' COMMENT '價格',
 `stock` int(11) NOT NULL DEFAULT '0' COMMENT 'stock',
 PRIMARY KEY (`goodsId`)
) ENGINE=InnoDB AUTO_INCREMENT=0 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci COMMENT='商品表'

 

四，java代碼說明

1,CacheConfig.java

@Profile("cacheenable")   //prod這個profile時緩存才生效
@Configuration
@EnableCaching //開啟緩存
public class CacheConfig {
    public static final int DEFAULT_MAXSIZE = 10000;
    public static final int DEFAULT_TTL = 600;
    private SimpleCacheManager cacheManager = new SimpleCacheManager();

    //定義cache名稱、超時時長（秒）、最大容量
    public enum CacheEnum{
        goods(60,1000),       //有效期600秒, 最大容量1000
        homePage(7200,1000),  //有效期2個小時 , 最大容量1000
        ;
        CacheEnum(int ttl, int maxSize) {
            this.ttl = ttl;
            this.maxSize = maxSize;
        }
        private int maxSize=DEFAULT_MAXSIZE;    //最大數量
        private int ttl=DEFAULT_TTL;        //過期時間（秒）
        public int getMaxSize() {
            return maxSize;
        }
        public int getTtl() {
            return ttl;
        }
    }

    //創建基於Caffeine的Cache Manager
    @Bean
    @Primary
    public CacheManager caffeineCacheManager() {
        ArrayList<CaffeineCache> caches = new ArrayList<CaffeineCache>();
        for(CacheEnum c : CacheEnum.values()){
            caches.add(new CaffeineCache(c.name(),
                    Caffeine.newBuilder().recordStats()
                            .expireAfterWrite(c.getTtl(), TimeUnit.SECONDS)
                            .maximumSize(c.getMaxSize()).build())
            );
        }
        cacheManager.setCaches(caches);
        return cacheManager;
    }
    @Bean
    public CacheManager getCacheManager() {
        return cacheManager;
    }
}

說明：創建了兩個緩存 goods,homePage

 

2,RedisConfig.java

@Configuration
public class RedisConfig {

    @Bean
    @Primary
    public LettuceConnectionFactory redis1LettuceConnectionFactory(RedisStandaloneConfiguration redis1RedisConfig,
                                                                    GenericObjectPoolConfig redis1PoolConfig) {
        LettuceClientConfiguration clientConfig =
                LettucePoolingClientConfiguration.builder().commandTimeout(Duration.ofMillis(100))
                        .poolConfig(redis1PoolConfig).build();
        return new LettuceConnectionFactory(redis1RedisConfig, clientConfig);
    }

    @Bean
    public RedisTemplate redis1Template(
            @Qualifier("redis1LettuceConnectionFactory") LettuceConnectionFactory redis1LettuceConnectionFactory) {
        StringRedisTemplate redisTemplate = new StringRedisTemplate();

        //使用Jackson2JsonRedisSerializer來序列化和反序列化redis的value值
        redisTemplate.setKeySerializer(new StringRedisSerializer());
        redisTemplate.setValueSerializer(new GenericJackson2JsonRedisSerializer());

        //使用StringRedisSerializer來序列化和反序列化redis的key值
        redisTemplate.setHashKeySerializer(new StringRedisSerializer());
        redisTemplate.setHashValueSerializer(new GenericJackson2JsonRedisSerializer());

        //開啟事務
        redisTemplate.setEnableTransactionSupport(true);
        redisTemplate.setConnectionFactory(redis1LettuceConnectionFactory);
        redisTemplate.afterPropertiesSet();
        return redisTemplate;
    }

    @Configuration
    public static class Redis1Config {
        @Value("${spring.redis1.host}")
        private String host;
        @Value("${spring.redis1.port}")
        private Integer port;
        @Value("${spring.redis1.password}")
        private String password;
        @Value("${spring.redis1.database}")
        private Integer database;

        @Value("${spring.redis1.lettuce.pool.max-active}")
        private Integer maxActive;
        @Value("${spring.redis1.lettuce.pool.max-idle}")
        private Integer maxIdle;
        @Value("${spring.redis1.lettuce.pool.max-wait}")
        private Long maxWait;
        @Value("${spring.redis1.lettuce.pool.min-idle}")
        private Integer minIdle;

        @Bean
        public GenericObjectPoolConfig redis1PoolConfig() {
            GenericObjectPoolConfig config = new GenericObjectPoolConfig();
            config.setMaxTotal(maxActive);
            config.setMaxIdle(maxIdle);
            config.setMinIdle(minIdle);
            config.setMaxWaitMillis(maxWait);
            return config;
        }

        @Bean
        public RedisStandaloneConfiguration redis1RedisConfig() {
            RedisStandaloneConfiguration config = new RedisStandaloneConfiguration();
            config.setHostName(host);
            config.setPassword(RedisPassword.of(password));
            config.setPort(port);
            config.setDatabase(database);
            return config;
        }
    }
}

實現到redis的訪問連接配置

 

3,RedisListenerConfig.java

@Configuration
public class RedisListenerConfig {

    //創建兩個消息監聽器MessageListener
    @Bean
    RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory,
                                            MessageListenerAdapter listenerAdapter) {
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        container.addMessageListener(listenerAdapter, new PatternTopic(Constants.CHANNEL_GOODS));
        container.addMessageListener(listenerAdapter, new PatternTopic(Constants.CHANNEL_HOME));
        return container;
    }

    //指定接收消息的類名和方法名
    @Bean
    MessageListenerAdapter listenerAdapter(RedisMessageReceiver messageReceiver) {
        System.out.println("listenerAdapter");
        return new MessageListenerAdapter(messageReceiver, "onReceiveMessage");
    }

    //指定StringRedisTemplate的生成
    @Bean
    StringRedisTemplate stringRedisTemplate(RedisConnectionFactory connectionFactory) {
        return new StringRedisTemplate(connectionFactory);
    }
}

創建RedisMessageListenerContainer，創建兩個消息隊列的監聽

 

4,RedisMessageReceiver.java

@Component
public class RedisMessageReceiver {
    @Resource
    private LocalCacheService localCacheService;
    //收到消息后進行處理
    public void onReceiveMessage(String message,String channel) {
        message=message.replace("\\\"","\"");
        message=message.replace("\"{","{");
        message=message.replace("}\"","}");

        Msg msg = JSON.parseObject(message, Msg.class);
        System.out.println(channel+":消息："+msg.getMsgType()+";content:"+msg.getContent());
        if (channel.equals(Constants.CHANNEL_GOODS)) {
            if (msg.getMsgType().equals("deleteall")) {
                localCacheService.deleteGoodsCacheAll();
            } else if (msg.getMsgType().equals("delete") || msg.getMsgType().equals("update")) {
                String goodslist = msg.getContent();
                String[] strArr = goodslist.split(",");
                System.out.println(strArr);

                for (int i = 0; i < strArr.length; ++i){
                    Long goodsId = Long.parseLong(strArr[i]);
                    if (msg.getMsgType().equals("update")) {
                        localCacheService.updateGoodsCache(goodsId);
                    } else if (msg.getMsgType().equals("delete")) {
                        localCacheService.deleteGoodsCache(goodsId);
                    }
                }
            }
        }
    }
}

說明：收到消息后，根據消息內容進行處理，

        我們收到的針對商品緩存的消息有三類：deleteall,update,delete

        分別調用三個不同的處理方法

 

5,LocalCacheServiceImpl.java

@Service
public class LocalCacheServiceImpl implements LocalCacheService {

    @Resource
    private RedisTemplate redis1Template;

    //更新緩存
    @CachePut(value = "goods", key="#goodsId")
    @Override
    public Goods updateGoodsCache(Long goodsId){
        System.out.println("get data from redis");
        Goods goodsr = (Goods) redis1Template.opsForValue().get("goods_"+String.valueOf(goodsId));
        return goodsr;
    }

    //刪除緩存
    @CacheEvict(value = "goods" ,key = "#goodsId")
    @Override
    public void deleteGoodsCache(Long goodsId) {
        System.out.println("刪除緩存 ");
    }

    //清除緩存
    @CacheEvict(value = "goods", allEntries=true)
    @Override
    public void deleteGoodsCacheAll() {
        System.out.println("已刪除全部緩存 ");
    }
}

說明：實現了對緩存的處理

 

6,HomeController.java

@RestController
@RequestMapping("/home")
public class HomeController {
    @Resource
    private RedisTemplate redis1Template;
    @Resource
    private GoodsService goodsService;
    @Resource
    private CacheManager getCacheManager;

    //發清空緩存的消息
    @GetMapping("/deleteall")
    public String deleteall(){
        String ret = "清除緩存的消息已發出";
        //刪除id為4的商品的緩存
        Msg msg_del = new Msg();
        msg_del.setMsgType("deleteall");
        msg_del.setContent("");
        redis1Template.convertAndSend("goodsCache",JSON.toJSONString(msg_del));
        return ret;
    }

    //發更新緩存和刪除緩存的消息
    @GetMapping("/update")
    public String update(){
           String ret = "";
           int goodsId = 3;
            //更新redis
           System.out.println("get data from redis");
           String key = "goods_"+String.valueOf(goodsId);
           Goods goodsr = (Goods)redis1Template.opsForValue().get(key);
           ret = "更新前:<br/>"+goodsr.toString()+"<br/>";
           String now = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.sss").format(System.currentTimeMillis());
           goodsr.setGoodsName("更新后的商品名，更新時間:"+now);
           redis1Template.opsForValue().set(key,goodsr);
           Goods goodsr2 = (Goods)redis1Template.opsForValue().get(key);
           ret += "更新后:<br/>"+goodsr2.toString()+"<br/>";

            //發布消息，接收者更新本地cache
            Msg msg_up = new Msg();
            msg_up.setMsgType("update");
            msg_up.setContent("3,5");
        redis1Template.convertAndSend("goodsCache",JSON.toJSONString(msg_up));

        //刪除id為4的商品的緩存
            Msg msg_del = new Msg();
            msg_del.setMsgType("delete");
            msg_del.setContent("4");
        redis1Template.convertAndSend("goodsCache",JSON.toJSONString(msg_del));
        return ret;
    }


    //商品詳情 參數:商品id
    @Cacheable(value = "goods", key="#goodsId",sync = true)
    @GetMapping("/goodsget")
    @ResponseBody
    public Goods goodsInfo(@RequestParam(value="goodsid",required = true,defaultValue = "0") Long goodsId) {
        Goods goods = goodsService.getOneGoodsById(goodsId);
        return goods;
    }

    //統計，如果是生產環境，需要加密才允許訪問
    @GetMapping("/stats")
    @ResponseBody
    public Object stats() {
        CaffeineCache caffeine = (CaffeineCache)getCacheManager.getCache("goods");
        Cache goods = caffeine.getNativeCache();
        String statsInfo="cache名字:goods<br/>";
        Long size = goods.estimatedSize();
        statsInfo += "size:"+size+"<br/>";
        ConcurrentMap map= goods.asMap();
        statsInfo += "map keys:<br/>";
        for(Object key : map.keySet()) {
            statsInfo += "key:"+key.toString()+";value:"+map.get(key)+"<br/>";
        }
        statsInfo += "統計信息:"+goods.stats().toString();
        return statsInfo;
    }
}

說明：更新/刪除/清空緩存的操作 我們都是通過發送redis消息實現，

         在生產環境中，這些功能需要放到管理后台

 

五，測試效果

1,生成緩存:

分別訪問:

http://127.0.0.1:8080/home/goodsget?goodsid=3

http://127.0.0.1:8080/home/goodsget?goodsid=4

http://127.0.0.1:8080/home/goodsget?goodsid=5

使商品id分別為 3/4/5的這三件商品生成caffeine緩存

查看效果:訪問:

http://127.0.0.1:8080/home/stats

可以看到緩存的數據:

cache名字:goods
size:3
map keys:
key:3;value: Goods:goodsId=3 goodsName=100分電動牙刷 subject=好用到讓你愛上刷牙 price=59.00 stock=15
key:4;value: Goods:goodsId=4 goodsName=蜂蜜牛奶手工皂 subject=深入滋養，肌膚細膩嫩滑 price=70.00 stock=33
key:5;value: Goods:goodsId=5 goodsName=紫光筷子筒 subject=紫光智護，干爽防潮更健康 price=189.00 stock=20
統計信息:CacheStats{hitCount=3, missCount=6, loadSuccessCount=6, loadFailureCount=0, totalLoadTime=624491686, evictionCount=3, evictionWeight=3}

 

2,更新緩存:訪問:

http://127.0.0.1:8080/home/update

我們在這個update方法中實現了兩項功能：

更新了緩存中商品id為3的商品的名字

刪除了緩存中商品id為4的對象

查看效果:,訪問:

http://127.0.0.1:8080/home/stats

返回：

cache名字:goods
size:2
map keys:
key:3;value: Goods:goodsId=3 goodsName=更新后的商品名，更新時間:2020-08-06 15:21:49.049 subject=好用到讓你愛上刷牙 price=59.00 stock=15
key:5;value: Goods:goodsId=5 goodsName=紫光筷子筒 subject=紫光智護，干爽防潮更健康 price=189.00 stock=20
統計信息:CacheStats{hitCount=1, missCount=3, loadSuccessCount=3, loadFailureCount=0, totalLoadTime=169516569, evictionCount=0, evictionWeight=0}

可以看到緩存中商品id為3的對象商品名被更新，

商品id為4的對象已被刪除

 

3,清除緩存:

訪問:

http://127.0.0.1:8080/home/deleteall

查看效果:訪問:

http://127.0.0.1:8080/home/stats

返回：

cache名字:goods
size:0
map keys:
統計信息:CacheStats{hitCount=1, missCount=3, loadSuccessCount=3, loadFailureCount=0, totalLoadTime=169516569, evictionCount=0, evictionWeight=0}

可以看到緩存名為goods的緩存中的對象已被清空

 

六，查看spring boot版本

  .   ____          _            __ _ _
 /\\ / ___'_ __ _ _(_)_ __  __ _ \ \ \ \
( ( )\___ | '_ | '_| | '_ \/ _` | \ \ \ \
 \\/  ___)| |_)| | | | | || (_| |  ) ) ) )
  '  |____| .__|_| |_|_| |_\__, | / / / /
 =========|_|==============|___/=/_/_/_/
 :: Spring Boot ::        (v2.3.2.RELEASE)