redis 缓存雪崩、缓存击穿、缓存穿透的解决方案

系统管理员 2021-09-18 05:36 459阅读 0赞

#### 缓存流程图 ####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dqYXZhZG9n_size_16_color_FFFFFF_t_70]

#### 缓存雪崩 ####

缓存雪崩，由于原有缓存失效，例如设置缓存时采用了相同的过期时间，在同一时刻出现大面积的缓存过期，所有原本应该访问缓存的请求都去查询数据库了，而对数据库CPU和内存造成巨大压力，严重的会造成数据库宕机。  
从而形成一系列连锁反应，造成整个系统崩溃。

模拟缓存雪崩示例：

@RequestMapping(value = "redis")
    @RestController
    @Slf4j
    public class CacheController { 
    
        @Autowired
        private StringRedisTemplate stringRedisTemplate;
        private AtomicInteger atomicInteger = new AtomicInteger();
    
    	@PostConstruct
        public void wrongInit() { 
            //初始化1000个城市数据到Redis，所有缓存数据有效期30秒
            IntStream.rangeClosed(1, 1000).forEach(i -> stringRedisTemplate.opsForValue()
                    .set("city" + i, getCityFromDb(i), 30, TimeUnit.SECONDS));
            log.info("Cache init finished");
    
            //每秒一次，输出数据库访问的QPS
            Executors.newSingleThreadScheduledExecutor().scheduleAtFixedRate(() -> { 
                log.info("DB QPS : {}", atomicInteger.getAndSet(0));
            }, 0, 1, TimeUnit.SECONDS);
        }
    
        @GetMapping("city")
        public String city() { 
            //随机查询一个城市
            int id = ThreadLocalRandom.current().nextInt(1000) + 1;
            String key = "city" + id;
            String data = stringRedisTemplate.opsForValue().get(key);
            // 模拟 redis 雪崩
            if (data == null) { 
                //回源到数据库查询
                data = getCityFromDb(id);
                if (!StringUtils.isEmpty(data))
                    //缓存30秒过期
                    stringRedisTemplate.opsForValue().set(key, data, 30, TimeUnit.SECONDS);
            }
            return data;
        }
    
        private String getCityFromDb(int cityId) { 
            //模拟查询数据库，查一次增加计数器加一
            atomicInteger.incrementAndGet();
            return "citydata" + System.currentTimeMillis();
        }
    }

使用 wrk 进行压测

wrk -c10 -t10 -d 100s http://localhost:8080/redis/city

结果：

[2020-07-18 19:51:15.848] [pool-1-thread-1] [INFO ] [c.o.controller.redis.CacheController:50] - DB QPS : 0
    [2020-07-18 19:51:16.848] [pool-1-thread-1] [INFO ] [c.o.controller.redis.CacheController:50] - DB QPS : 0
    [2020-07-18 19:51:17.846] [pool-1-thread-1] [INFO ] [c.o.controller.redis.CacheController:50] - DB QPS : 0
    [2020-07-18 19:51:18.848] [pool-1-thread-1] [INFO ] [c.o.controller.redis.CacheController:50] - DB QPS : 0
    [2020-07-18 19:51:19.848] [pool-1-thread-1] [INFO ] [c.o.controller.redis.CacheController:50] - DB QPS : 759
    [2020-07-18 19:51:20.848] [pool-1-thread-1] [INFO ] [c.o.controller.redis.CacheController:50] - DB QPS : 245
    [2020-07-18 19:51:21.848] [pool-1-thread-1] [INFO ] [c.o.controller.redis.CacheController:50] - DB QPS : 2
    [2020-07-18 19:51:22.848] [pool-1-thread-1] [INFO ] [c.o.controller.redis.CacheController:50] - DB QPS : 0

可以看到在 key 同一时间失效时，DB 的 QPS 增到了 749。

解决的方式有一下几种：

1.  key 设置差异化的过期时间

@PostConstruct
    public void rightInit1() { 
        //这次缓存的过期时间是30秒+10秒内的随机延迟
        IntStream.rangeClosed(1, 1000).forEach(i -> stringRedisTemplate.opsForValue().set("city" + i, getCityFromDb(i), 30 + ThreadLocalRandom.current().nextInt(10), TimeUnit.SECONDS));
        log.info("Cache init finished");
        //同样1秒一次输出数据库QPS：
       Executors.newSingleThreadScheduledExecutor().scheduleAtFixedRate(() -> { 
            log.info("DB QPS : {}", atomicInteger.getAndSet(0));
        }, 0, 1, TimeUnit.SECONDS);
    }

继续压测，QPS 得到明显的降低

[2020-07-18 19:55:23.312] [pool-1-thread-1] [INFO ] [c.o.controller.redis.CacheController:63] - DB QPS : 0
    [2020-07-18 19:55:24.313] [pool-1-thread-1] [INFO ] [c.o.controller.redis.CacheController:63] - DB QPS : 0
    [2020-07-18 19:55:25.312] [pool-1-thread-1] [INFO ] [c.o.controller.redis.CacheController:63] - DB QPS : 0
    [2020-07-18 19:55:26.313] [pool-1-thread-1] [INFO ] [c.o.controller.redis.CacheController:63] - DB QPS : 0
    [2020-07-18 19:55:27.312] [pool-1-thread-1] [INFO ] [c.o.controller.redis.CacheController:63] - DB QPS : 0
    [2020-07-18 19:55:28.312] [pool-1-thread-1] [INFO ] [c.o.controller.redis.CacheController:63] - DB QPS : 0
    [2020-07-18 19:55:29.313] [pool-1-thread-1] [INFO ] [c.o.controller.redis.CacheController:63] - DB QPS : 72
    [2020-07-18 19:55:30.313] [pool-1-thread-1] [INFO ] [c.o.controller.redis.CacheController:63] - DB QPS : 87
    [2020-07-18 19:55:31.313] [pool-1-thread-1] [INFO ] [c.o.controller.redis.CacheController:63] - DB QPS : 107
    [2020-07-18 19:55:32.312] [pool-1-thread-1] [INFO ] [c.o.controller.redis.CacheController:63] - DB QPS : 40
    [2020-07-18 19:55:33.312] [pool-1-thread-1] [INFO ] [c.o.controller.redis.CacheController:63] - DB QPS : 154
    [2020-07-18 19:55:34.312] [pool-1-thread-1] [INFO ] [c.o.controller.redis.CacheController:63] - DB QPS : 110
    [2020-07-18 19:55:35.311] [pool-1-thread-1] [INFO ] [c.o.controller.redis.CacheController:63] - DB QPS : 113
    [2020-07-18 19:55:36.311] [pool-1-thread-1] [INFO ] [c.o.controller.redis.CacheController:63] - DB QPS : 93
    [2020-07-18 19:55:37.311] [pool-1-thread-1] [INFO ] [c.o.controller.redis.CacheController:63] - DB QPS : 110
    [2020-07-18 19:55:38.311] [pool-1-thread-1] [INFO ] [c.o.controller.redis.CacheController:63] - DB QPS : 108
    [2020-07-18 19:55:39.311] [pool-1-thread-1] [INFO ] [c.o.controller.redis.CacheController:63] - DB QPS : 16
    [2020-07-18 19:55:40.310] [pool-1-thread-1] [INFO ] [c.o.controller.redis.CacheController:63] - DB QPS : 0
    [2020-07-18 19:55:41.310] [pool-1-thread-1] [INFO ] [c.o.controller.redis.CacheController:63] - DB QPS : 0

1.  让缓存不主动过期，定期更新缓存

@PostConstruct
    public void rightInit2() throws InterruptedException { 
        CountDownLatch countDownLatch = new CountDownLatch(1);
        //每隔30秒全量更新一次缓存 
        Executors.newSingleThreadScheduledExecutor().scheduleAtFixedRate(() -> { 
            IntStream.rangeClosed(1, 1000).forEach(i -> { 
                String data = getCityFromDb(i);
                //模拟更新缓存需要一定的时间
                try { 
                    TimeUnit.MILLISECONDS.sleep(20);
                } catch (InterruptedException e) {  }
                if (!StringUtils.isEmpty(data)) { 
                    //缓存永不过期，被动更新
                    stringRedisTemplate.opsForValue().set("city" + i, data);
                }
            });
            log.info("Cache update finished");
            //启动程序的时候需要等待首次更新缓存完成
            countDownLatch.countDown();
        }, 0, 30, TimeUnit.SECONDS);
    
        Executors.newSingleThreadScheduledExecutor().scheduleAtFixedRate(() -> { 
            log.info("DB QPS : {}", atomicInteger.getAndSet(0));
        }, 0, 1, TimeUnit.SECONDS);
    
        countDownLatch.await();
    }

继续压测，QPS 很稳定

[2020-07-18 19:58:25.462] [pool-2-thread-1] [INFO ] [c.o.controller.redis.CacheController:92] - DB QPS : 29
    [2020-07-18 19:58:26.460] [pool-2-thread-1] [INFO ] [c.o.controller.redis.CacheController:92] - DB QPS : 38
    [2020-07-18 19:58:27.461] [pool-2-thread-1] [INFO ] [c.o.controller.redis.CacheController:92] - DB QPS : 38
    [2020-07-18 19:58:28.459] [pool-2-thread-1] [INFO ] [c.o.controller.redis.CacheController:92] - DB QPS : 37
    [2020-07-18 19:58:29.459] [pool-2-thread-1] [INFO ] [c.o.controller.redis.CacheController:92] - DB QPS : 39
    [2020-07-18 19:58:30.458] [pool-2-thread-1] [INFO ] [c.o.controller.redis.CacheController:92] - DB QPS : 37
    [2020-07-18 19:58:31.461] [pool-2-thread-1] [INFO ] [c.o.controller.redis.CacheController:92] - DB QPS : 39
    [2020-07-18 19:58:32.459] [pool-2-thread-1] [INFO ] [c.o.controller.redis.CacheController:92] - DB QPS : 38
    [2020-07-18 19:58:33.462] [pool-2-thread-1] [INFO ] [c.o.controller.redis.CacheController:92] - DB QPS : 38

1.  使用加锁的方式  
    但是在高并发的情况下，会导致大量的线程去竞争锁，导致用户等待时间过长；如果应用是分布式部署，还要考虑使用分布式锁，增加的系统的复杂性，一般不采用这种方式。

@GetMapping("city")
        public String city() { 
            //随机查询一个城市
            int id = ThreadLocalRandom.current().nextInt(1000) + 1;
            String key = "city" + id;
            String data = stringRedisTemplate.opsForValue().get(key);
            // 模拟 redis 雪崩
            if (data == null) { 
            	// 加锁
            	lock(key);
                //回源到数据库查询
                data = getCityFromDb(id);
                if (!StringUtils.isEmpty(data))
                    //缓存30秒过期
                    stringRedisTemplate.opsForValue().set(key, data, 30, TimeUnit.SECONDS);
            }
            // 注意释放锁
            lock.unlock();
            return data;
        }

1.  做二级缓存，或者双缓存策略。A1为原始缓存，A2为拷贝缓存，A1失效时，可以访问A2，A1缓存失效时间设置为短期，A2设置为长期。

那么如何选择解决方案呢？

*  如果无法全量缓存所有数据，那么只能使用方案一
 *  即使使用了方案二，缓存永不过期，同样需要在查询的时候，确保有回源的逻辑。正如之前所说，我们无法确保缓存系统中的数据永不丢失
 *  在把数据从数据库加入缓存的时候，都需要判断来自数据库的数据是否合法，比如进行最基本的判空检查

#### 缓存穿透 ####

缓存穿透是指用户查询数据，在数据库没有，在缓存中也没有。这样就导致用户查询的时候，在缓存中找不到，每次都要去数据库再查询一遍，然后返回空（相当于进行了两次无用的查询）。

示例代码：

@GetMapping("wrong2")
        public String wrong(@RequestParam("id") int id) { 
            String key = "user" + id;
            String data = stringRedisTemplate.opsForValue().get(key);
            //无法区分是无效用户还是缓存失效
            if (StringUtils.isEmpty(data)) { 
                data = getCityFromDb2(id);
                stringRedisTemplate.opsForValue().set(key, data, 30, TimeUnit.SECONDS);
            }
            return data;
        }
    
        private String getCityFromDb2(int id) { 
            atomicInteger.incrementAndGet();
            //注意，只有ID介于0（不含）和10000（包含）之间的用户才是有效用户，可以查询到用户信息
            if (id > 0 && id <= 10000) return "userdata";
            //否则返回空字符串
            return "";
        }

压测一下：

wrk -c10 -t10 -d 100s http://localhost:8080/redis/wrong2\?id\=0

结果：

[2020-07-18 21:48:50.440] [pool-1-thread-1] [INFO ] [c.o.controller.redis.CacheController:128] - DB QPS : 200
    [2020-07-18 21:48:51.440] [pool-1-thread-1] [INFO ] [c.o.controller.redis.CacheController:128] - DB QPS : 1656
    [2020-07-18 21:48:52.440] [pool-1-thread-1] [INFO ] [c.o.controller.redis.CacheController:128] - DB QPS : 2318
    [2020-07-18 21:48:53.441] [pool-1-thread-1] [INFO ] [c.o.controller.redis.CacheController:128] - DB QPS : 2789
    [2020-07-18 21:48:54.440] [pool-1-thread-1] [INFO ] [c.o.controller.redis.CacheController:128] - DB QPS : 2311

缓存穿透是指，缓存没有起到压力缓冲的作用；而缓存击穿是指，缓存失效时瞬时的并发打到数据库。  
解决方案：

1.  使用布隆过滤器，提前将所有的数据 hash 到 bitmap 中  
    可以使用 Google 的 guava 包中包含的 bloomfilter

<dependencies>
       <dependency>
           <groupId>com.google.guava</groupId>
           <artifactId>guava</artifactId>
           <version>22.0</version>
       </dependency>
    </dependencies>

示例代码

private BloomFilter<Integer> bloomFilter;
    
    @PostConstruct
    public void init() { 
        //创建布隆过滤器，元素数量10000，期望误判率1%
        bloomFilter = BloomFilter.create(Funnels.integerFunnel(), 10000, 0.01);
        //填充布隆过滤器
        IntStream.rangeClosed(1, 10000).forEach(bloomFilter::put);
    }
    
    @GetMapping("right2")
    public String right2(@RequestParam("id") int id) { 
        String data = "";
        //通过布隆过滤器先判断
        if (bloomFilter.mightContain(id)) { 
            String key = "user" + id;
            //走缓存查询
            data = stringRedisTemplate.opsForValue().get(key);
            if (StringUtils.isEmpty(data)) { 
                //走数据库查询
                data = getCityFromDb(id);
                stringRedisTemplate.opsForValue().set(key, data, 30, TimeUnit.SECONDS);
            }
        }
        return data;
    }

需要同步所有可能存在的值并加入布隆过滤器，这是比较麻烦的地方。如果业务规则明确的话，也可以考虑直接根据业务规则判断值是否存在。  
其实，方案一可以和方案二同时使用，即将布隆过滤器前置，对于误判的情况再保存特殊值到缓存，双重保险避免无效数据查询请求打到数据库。

1.  将反会的 null 值也缓存到 redis 中，并设置过期时间  
    但，这种方式可能会把大量无效的数据加入缓存中，如果担心大量无效数据占满缓存的话还可以考虑方案一，即使用布隆过滤器做前置过滤。  
    示例代码：

@GetMapping("right")
    public String right(@RequestParam("id") int id) { 
        String key = "user" + id;
        String data = stringRedisTemplate.opsForValue().get(key);
        if (StringUtils.isEmpty(data)) { 
            data = getCityFromDb(id);
            //校验从数据库返回的数据是否有效
            if (!StringUtils.isEmpty(data)) { 
                stringRedisTemplate.opsForValue().set(key, data, 30, TimeUnit.SECONDS);
            }
            else { 
                //如果无效，直接在缓存中设置一个NODATA，这样下次查询时即使是无效用户还是可以命中缓存
                stringRedisTemplate.opsForValue().set(key, "NODATA", 30, TimeUnit.SECONDS);
            }
        }
        return data;
    }

#### 缓存击穿（热点Key） ####

缓存击穿是指数据库中有数据，缓存中的数据突然过期，同时该数据为热点数据，突然大量的并发访问，导致直接去查询数据库。  
示例如下，在redis中存储 hotKey，5s 过期时间，使用wrk进行压测

@PostConstruct
        public void init() { 
            //初始化一个热点数据到Redis中，过期时间设置为5秒
            stringRedisTemplate.opsForValue().set("hotsopt", getExpensiveData(), 5, TimeUnit.SECONDS);
            //每隔1秒输出一下回源的QPS
    
            Executors.newSingleThreadScheduledExecutor().scheduleAtFixedRate(() -> { 
                log.info("DB QPS : {}", atomicInteger.getAndSet(0));
            }, 0, 1, TimeUnit.SECONDS);
        }
    
        @GetMapping("wrong")
        public String wrong() { 
            String data = stringRedisTemplate.opsForValue().get("hotKey");
            if (StringUtils.isEmpty(data)) { 
                data = getExpensiveData();
                //重新加入缓存，过期时间还是5秒
                stringRedisTemplate.opsForValue().set("hotKey", data, 5, TimeUnit.SECONDS);
            }
            return data;
        }
    
        private String getExpensiveData() { 
            //模拟查询数据库，查一次增加计数器加一
            atomicInteger.incrementAndGet();
            return "hotValue";
        }

结果显示，每过 5s 就会有一个QPS高的时刻

[2020-07-18 21:38:27.815] [pool-1-thread-1] [INFO ] [c.o.controller.redis.CacheController:125] - DB QPS : 0
    [2020-07-18 21:38:28.813] [pool-1-thread-1] [INFO ] [c.o.controller.redis.CacheController:125] - DB QPS : 0
    [2020-07-18 21:38:29.815] [pool-1-thread-1] [INFO ] [c.o.controller.redis.CacheController:125] - DB QPS : 10
    [2020-07-18 21:38:30.813] [pool-1-thread-1] [INFO ] [c.o.controller.redis.CacheController:125] - DB QPS : 0
    [2020-07-18 21:38:31.815] [pool-1-thread-1] [INFO ] [c.o.controller.redis.CacheController:125] - DB QPS : 0
    [2020-07-18 21:38:32.814] [pool-1-thread-1] [INFO ] [c.o.controller.redis.CacheController:125] - DB QPS : 0
    [2020-07-18 21:38:33.815] [pool-1-thread-1] [INFO ] [c.o.controller.redis.CacheController:125] - DB QPS : 0
    [2020-07-18 21:38:34.815] [pool-1-thread-1] [INFO ] [c.o.controller.redis.CacheController:125] - DB QPS : 10
    [2020-07-18 21:38:35.815] [pool-1-thread-1] [INFO ] [c.o.controller.redis.CacheController:125] - DB QPS : 0
    [2020-07-18 21:38:36.814] [pool-1-thread-1] [INFO ] [c.o.controller.redis.CacheController:125] - DB QPS : 0
    [2020-07-18 21:38:37.815] [pool-1-thread-1] [INFO ] [c.o.controller.redis.CacheController:125] - DB QPS : 0
    [2020-07-18 21:38:38.814] [pool-1-thread-1] [INFO ] [c.o.controller.redis.CacheController:125] - DB QPS : 0
    [2020-07-18 21:38:39.815] [pool-1-thread-1] [INFO ] [c.o.controller.redis.CacheController:125] - DB QPS : 9
    [2020-07-18 21:38:40.814] [pool-1-thread-1] [INFO ] [c.o.controller.redis.CacheController:125] - DB QPS : 0
    [2020-07-18 21:38:41.815] [pool-1-thread-1] [INFO ] [c.o.controller.redis.CacheController:125] - DB QPS : 0

解决方案：

1.  使用互斥锁，和缓存雪崩的解决方式一一样

@Autowired
    private RedissonClient redissonClient;
    @GetMapping("right")
    public String right() {
        String data = stringRedisTemplate.opsForValue().get("hotKey");
        if (StringUtils.isEmpty(data)) {
            RLock locker = redissonClient.getLock("locker");
            //获取分布式锁
            if (locker.tryLock()) {
                try {
                    data = stringRedisTemplate.opsForValue().get("hotsopt");
                    //双重检查，因为可能已经有一个B线程过了第一次判断，在等锁，然后A线程已经把数据写入了Redis中
                    if (StringUtils.isEmpty(data)) {
                        //回源到数据库查询
                        data = getExpensiveData();
                        stringRedisTemplate.opsForValue().set("hotsopt", data, 5, TimeUnit.SECONDS);
                    }
                } finally {
                    //别忘记释放，另外注意写法，获取锁后整段代码try+finally，确保unlock万无一失
                    locker.unlock();
                }
            }
        }
        return data;
    }

压测结果：

[2020-07-18 21:42:14.438] [pool-1-thread-1] [INFO ] [c.o.controller.redis.CacheController:127] - DB QPS : 0
    [2020-07-18 21:42:15.438] [pool-1-thread-1] [INFO ] [c.o.controller.redis.CacheController:127] - DB QPS : 0
    [2020-07-18 21:42:16.437] [pool-1-thread-1] [INFO ] [c.o.controller.redis.CacheController:127] - DB QPS : 1
    [2020-07-18 21:42:17.438] [pool-1-thread-1] [INFO ] [c.o.controller.redis.CacheController:127] - DB QPS : 0
    [2020-07-18 21:42:18.439] [pool-1-thread-1] [INFO ] [c.o.controller.redis.CacheController:127] - DB QPS : 0
    [2020-07-18 21:42:19.438] [pool-1-thread-1] [INFO ] [c.o.controller.redis.CacheController:127] - DB QPS : 0
    [2020-07-18 21:42:20.438] [pool-1-thread-1] [INFO ] [c.o.controller.redis.CacheController:127] - DB QPS : 0
    [2020-07-18 21:42:21.438] [pool-1-thread-1] [INFO ] [c.o.controller.redis.CacheController:127] - DB QPS : 1
    [2020-07-18 21:42:22.438] [pool-1-thread-1] [INFO ] [c.o.controller.redis.CacheController:127] - DB QPS : 0

1.  缓存永不过期  
    真正的缓存过期时间不有Redis控制，而是由程序代码控制，有可能会造成缓存不一致的现象。  
    和缓存雪崩相同的方案。

在真实的业务场景下，不一定要这么严格地使用双重检查分布式锁进行全局的并发限制，因为这样虽然可以把数据库回源并发降到最低，但也限制了缓存失效时的并发。可以考虑的方式是：

*  方案一，使用进程内的锁进行限制，这样每一个节点都可以以一个并发回源数据库；
 *  方案二，不使用锁进行限制，而是使用类似 Semaphore 的工具限制并发数，比如限制为 10，这样既限制了回源并发数不至于太大，又能使得一定量的线程可以同时回源。

#### 缓存数据同步策略 ####

*  先更新缓存，再更新数据库；  
    “先更新缓存再更新数据库”策略不可行。数据库设计复杂，压力集中，数据库因为超时等原因更新操作失败的可能性较大，此外还会涉及事务，很可能因为数据库更新失败，导致缓存和数据库的数据不一致。
 *  先更新数据库，再更新缓存；  
    “先更新数据库再更新缓存”策略不可行。一是，如果线程 A 和 B 先后完成数据库更新，但更新缓存时却是 B 和 A 的顺序，那很可能会把旧数据更新到缓存中引起数据不一致；二是，我们不确定缓存中的数据是否会被访问，不一定要把所有数据都更新到缓存中去。
 *  先删除缓存，再更新数据库，访问的时候按需加载数据到缓存；  
    “先删除缓存再更新数据库，访问的时候按需加载数据到缓存”策略也不可行。在并发的情况下，很可能删除缓存后还没来得及更新数据库，就有另一个线程先读取了旧值到缓存中，如果并发量很大的话这个概率也会很大。
 *  先更新数据库，再删除缓存，访问的时候按需加载数据到缓存。  
    “先更新数据库再删除缓存，访问的时候按需加载数据到缓存”策略是最好的。虽然在极端情况下，这种策略也可能出现数据不一致的问题，但概率非常低，基本可以忽略。举一个“极端情况”的例子，比如更新数据的时间节点恰好是缓存失效的瞬间，这时 A 先读取到了旧值，随后在 B 操作数据库完成更新并且删除了缓存之后，A 再把旧值加入缓存。

因此，针对缓存更新更推荐的方式是，缓存中的数据不由数据更新操作主动触发，统一在需要使用的时候按需加载，数据更新后及时删除缓存中的数据即可。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dqYXZhZG9n_size_16_color_FFFFFF_t_70]: /images/20210918/008eb61c98e34457a654eae23affe433.png