当前位置：首页 > news >正文

redis高级案列case

news 来源：原创 2024/5/20 11:38:37

案列一双写一致性

案例二双锁策略

package com.redis.redis01.service;import com.redis.redis01.bean.RedisBs;
import com.redis.redis01.mapper.RedisBsMapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;import javax.annotation.Resource;
import java.beans.Transient;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;@Slf4j
@Service
public class RedisBsService {//定义key前缀/命名空间public static final String CACHE_KEY_USER = "user:";@Autowiredprivate RedisBsMapper mapper;@Resourceprivate RedisTemplate<String, Object> redisTemplate;private static ReentrantLock lock = new ReentrantLock();/*** 业务逻辑没有写错，对于中小长(qps<=1000)可以使用，但是大厂不行：大长需要采用双检加锁策略** @param id* @return*/@Transactionalpublic RedisBs findUserById(Integer id,int type,int qps) {//qps<=1000if(qps<=1000){return qpsSmall1000(id);}//qps>1000return qpsBig1000(id, type);}/*** 加强补充，避免突然key失效了，或者不存在的key穿透redis打爆mysql，做一下预防，尽量不出现缓存击穿的情况，进行排队等候* @param id* @param type 0使用synchronized重锁，1ReentrantLock轻量锁* @return*/private RedisBs qpsBig1000(Integer id, int type) {RedisBs redisBs = null;String key = CACHE_KEY_USER + id;//1先从redis里面查询，如果有直接返回，没有再去查mysqlredisBs = (RedisBs) redisTemplate.opsForValue().get(key);if (null == redisBs) {switch (type) {case 0://加锁，假设请求量很大，缓存过期，大厂用，对于高qps的优化，进行加锁保证一个请求操作，让外面的redis等待一下，避免击穿mysqlsynchronized (RedisBsService.class) {//第二次查询缓存目的防止加锁之前刚好被其他线程缓存了redisBs = (RedisBs) redisTemplate.opsForValue().get(key);if (null != redisBs) {//查询到数据直接返回return redisBs;} else {//数据缓存//查询mysql，回写到redis中redisBs = mapper.findUserById(id);if (null == redisBs) {// 3 redis+mysql都没有数据，防止多次穿透(redis为防弹衣，mysql为人，穿透直接伤人，就是直接访问mysql)，优化：记录这个null值的key，列入黑名单或者记录或者异常return new RedisBs(-1, "当前值已经列入黑名单");}//4 mysql有，回写保证数据一致性//setifabsentredisTemplate.opsForValue().setIfAbsent(key, redisBs,7l, TimeUnit.DAYS);}}break;case 1://加锁，大厂用，对于高qps的优化，进行加锁保证一个请求操作，让外面的redis等待一下，避免击穿mysqllock.lock();try {//第二次查询缓存目的防止加锁之前刚好被其他线程缓存了redisBs = (RedisBs) redisTemplate.opsForValue().get(key);if (null != redisBs) {//查询到数据直接返回return redisBs;} else {//数据缓存//查询mysql，回写到redis中redisBs = mapper.findUserById(id);if (null == redisBs) {// 3 redis+mysql都没有数据，防止多次穿透(redis为防弹衣，mysql为人，穿透直接伤人，就是直接访问mysql)，优化：记录这个null值的key，列入黑名单或者记录或者异常return new RedisBs(-1, "当前值已经列入黑名单");}//4 mysql有，回写保证数据一致性redisTemplate.opsForValue().set(key, redisBs);}} catch (Exception e) {e.printStackTrace();} finally {//解锁lock.unlock();}}}return redisBs;}private RedisBs qpsSmall1000(Integer id) {RedisBs redisBs = null;String key = CACHE_KEY_USER + id;//1先从redis里面查询，如果有直接返回，没有再去查mysqlredisBs = (RedisBs) redisTemplate.opsForValue().get(key);if (null == redisBs) {//2查询mysql，回写到redis中redisBs = mapper.findUserById(id);if (null == redisBs) {// 3 redis+mysql都没有数据，防止多次穿透(redis为防弹衣，mysql为人，穿透直接伤人，就是直接访问mysql)，优化：记录这个null值的key，列入黑名单或者记录或者异常return new RedisBs(-1, "当前值已经列入黑名单");}//4 mysql有，回写保证数据一致性redisTemplate.opsForValue().set(key, redisBs);}return redisBs;}}

案列三 mysql+redis实时同步

下载canal监控端admin和服务端deployer

https://github.com/alibaba/canal/releases/tag/canal-1.1.7

登录mysql授权canal连接mysql账户

DROP USER IF EXISTS 'canal'@'%';
CREATE USER 'canal'@'%' IDENTIFIED BY 'canal';  
GRANT ALL PRIVILEGES ON *.* TO 'canal'@'%' IDENTIFIED BY 'canal';  
FLUSH PRIVILEGES;

配置canal

修改mysql ip

启动

./startup.bat

Canal客户端（Java编写）

非springboot项目

<dependency><groupId>com.alibaba.otter</groupId><artifactId>canal.client</artifactId><version>1.1.0</version>
</dependency>

server.port=8002
#连接数据源
spring.datasource.druid.username=root
spring.datasource.druid.password=xgm@2023..
spring.datasource.druid.url=jdbc:mysql://172.16.204.51:3306/redis?serverTimezone=GMT%2B8
spring.datasource.druid.driver-class-name=com.mysql.cj.jdbc.Driver
spring.datasource.druid.initial-size=5##指定缓存类型redis
#spring.cache.type=redis
##一个小时，以毫秒为单位
#spring.cache.redis.time-to-live=3600000
##给缓存的建都起一个前缀。  如果指定了前缀就用我们指定的，如果没有就默认使用缓存的名字作为前缀,一般不指定
#spring.cache.redis.key-prefix=CACHE_
##指定是否使用前缀
#spring.cache.redis.use-key-prefix=true
##是否缓存空值，防止缓存穿透
#spring.cache.redis.cache-null-values=true#redis
spring.redis.host=172.16.204.51
spring.redis.port=6379
spring.redis.password=123456
spring.redis.database=1# mybatis配置
mybatis:
check-config-location: true
#  mybatis框架配置文件，对mybatis的生命周期起作用
config-location: "classpath:mybatis/mybatis-config.xml"
#  配置xml路径
mapper-locations: "classpath:mybatis/mapper/*Mapper.xml"
#  配置model包路径
type-aliases-package: "com.redis.redis01.bean.*"#日志
logging.level.root=info
#logging.level.io.lettuce.core=debug
#logging.level.org.springframework.data.redis=debug#canal安装地址
canal.server=172.16.204.51:11111
canal.destination=example
#控制台刷新时间，每隔5秒检查一下数据库数据是否更新 根据需求设置其他时间
canal.timeout=5
spring.datasource.driver-class-name=com.mysql.cj.jdbc.Driver
spring.datasource.username=canal
spring.datasource.password=canal
spring.datasource.url=jdbc:mysql://172.16.204.51:3306/redis?autoReconnect=true&useUnicode=true&characterEncoding=utf8&zeroDateTimeBehavior=CONVERT_TO_NULL&useSSL=false&serverTimezone=CTT&allowMultiQueries=true

package com.redis.redis01;import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.common.utils.AddressUtils;
import com.alibaba.otter.canal.protocol.Message;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.common.utils.AddressUtils;
import com.alibaba.otter.canal.protocol.Message;
import com.alibaba.otter.canal.protocol.CanalEntry.Column;
import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
import com.alibaba.otter.canal.protocol.CanalEntry.EntryType;
import com.alibaba.otter.canal.protocol.CanalEntry.EventType;
import com.alibaba.otter.canal.protocol.CanalEntry.RowChange;
import com.alibaba.otter.canal.protocol.CanalEntry.RowData;
import com.google.gson.Gson;
import lombok.extern.slf4j.Slf4j;
import org.json.JSONException;
import org.json.JSONObject;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;import java.net.InetSocketAddress;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
@Slf4j
public class CanalTest {public static final Integer _60SECONDS = 60;public static final String REDIS_IP_ADDR = "172.16.204.51";private  void redisInsert(List<Column> columns) throws JSONException {JSONObject jsonObject = new JSONObject();for (Column column : columns) {System.out.println(column.getName() + " : " + column.getValue() + "    update=" + column.getUpdated());jsonObject.put(column.getName(), column.getValue());}if (columns.size() > 0) {try (Jedis jedis = new RedisUtils().getJedis()){jedis.set(columns.get(0).getValue(), jsonObject.toString());} catch (Exception e) {e.printStackTrace();}}}public class RedisUtils {public  final String REDIS_IP_ADDR = "172.16.204.51";public  final String REDIS_pwd = "123456";public JedisPool jedisPool;public  Jedis getJedis() throws Exception {JedisPoolConfig jedisPoolConfig = new JedisPoolConfig();jedisPoolConfig.setMaxTotal(20);jedisPoolConfig.setMaxIdle(10);jedisPool=new JedisPool(jedisPoolConfig, REDIS_IP_ADDR,6379,10000,REDIS_pwd);if (null != jedisPool) {return jedisPool.getResource();}throw new Exception("Jedispool is not ok");}}private void redisDelete(List<Column> columns) throws JSONException {JSONObject jsonObject = new JSONObject();for (Column column : columns) {jsonObject.put(column.getName(), column.getValue());}if (columns.size() > 0) {try (Jedis jedis = new RedisUtils().getJedis()) {jedis.del(columns.get(0).getValue());} catch (Exception e) {e.printStackTrace();}}}private void redisUpdate(List<Column> columns) throws JSONException {JSONObject jsonObject = new JSONObject();for (Column column : columns) {System.out.println(column.getName() + " : " + column.getValue() + "    update=" + column.getUpdated());jsonObject.put(column.getName(), column.getValue());}if (columns.size() > 0) {try (Jedis jedis =new RedisUtils().getJedis()){jedis.set(columns.get(0).getValue(), jsonObject.toString());System.out.println("---------update after: " + jedis.get(columns.get(0).getValue()));} catch (Exception e) {e.printStackTrace();}}}public  void printEntry(List<Entry> entrys) throws JSONException {for (Entry entry : entrys) {if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN || entry.getEntryType() == EntryType.TRANSACTIONEND) {continue;}RowChange rowChage = null;try {//获取变更的row数据rowChage = RowChange.parseFrom(entry.getStoreValue());} catch (Exception e) {throw new RuntimeException("ERROR ## parser of eromanga-event has an error,data:" + entry.toString(), e);}//获取变动类型EventType eventType = rowChage.getEventType();System.out.println(String.format("================&gt; binlog[%s:%s] , name[%s,%s] , eventType : %s",entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),entry.getHeader().getSchemaName(), entry.getHeader().getTableName(), eventType));for (RowData rowData : rowChage.getRowDatasList()) {if (eventType == EventType.INSERT) {redisInsert(rowData.getAfterColumnsList());} else if (eventType == EventType.DELETE) {redisDelete(rowData.getBeforeColumnsList());} else {//EventType.UPDATEredisUpdate(rowData.getAfterColumnsList());}}}}public static void main(String[] args) {System.out.println("---------O(∩_∩)O哈哈~ initCanal() main方法-----------");//=================================// 创建链接canal服务端CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(REDIS_IP_ADDR,11111), "example", "", "");  // 这里用户名和密码如果在这写了，会覆盖canal配置文件的账号密码，如果不填从配置文件中读int batchSize = 1000;//空闲空转计数器int emptyCount = 0;System.out.println("---------------------canal init OK，开始监听mysql变化------");try {connector.connect();//connector.subscribe(".*\\..*");connector.subscribe("redis.redis_syc");   // 设置监听哪个表connector.rollback();int totalEmptyCount = 10 * _60SECONDS;while (emptyCount < totalEmptyCount) {System.out.println("我是canal，每秒一次正在监听:" + UUID.randomUUID().toString());Message message = connector.getWithoutAck(batchSize); // 获取指定数量的数据long batchId = message.getId();int size = message.getEntries().size();if (batchId == -1 || size == 0) {emptyCount++;try {TimeUnit.SECONDS.sleep(1);} catch (InterruptedException e) {e.printStackTrace();}} else {//计数器重新置零emptyCount = 0;new CanalTest().printEntry(message.getEntries());}connector.ack(batchId); // 提交确认// connector.rollback(batchId); // 处理失败, 回滚数据}System.out.println("已经监听了" + totalEmptyCount + "秒，无任何消息，请重启重试......");} catch (JSONException e) {throw new RuntimeException(e);} finally {connector.disconnect();}}}

截图

spingboot项目

        <dependency><groupId>top.javatool</groupId><artifactId>canal-spring-boot-starter</artifactId><version>1.2.1-RELEASE</version></dependency

# canal starter配置信息
canal.server=127.0.0.1:11111
canal.destination=examplelogging.level.root=info
logging.level.top.javatool.canal.client.client.AbstractCanalClient=error

package com.redis.redis01.canal;import com.redis.redis01.bean.RedisSyc;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import top.javatool.canal.client.annotation.CanalTable;
import top.javatool.canal.client.handler.EntryHandler;@Component
@CanalTable(value = "redis_syc")
@Slf4j
public class RedisCanalClientExample implements EntryHandler<RedisSyc> {@Overridepublic void insert(RedisSyc redisSyc) {EntryHandler.super.insert(redisSyc);log.info("新增 ---> {}",redisSyc);}@Overridepublic void update(RedisSyc before, RedisSyc after) {EntryHandler.super.update(before, after);log.info("更新前 --->{} , 更新后 --->{} ", before, after);}@Overridepublic void delete(RedisSyc redisSyc) {EntryHandler.super.delete(redisSyc);log.info("删除 --->{} " , redisSyc);}
}

私服监听

注意：canal依赖stater在中央仓库是不存在的，需要手动放进本地仓库或者你公司里面的nexus

	<!--canal依赖--><dependency><groupId>com.xpand</groupId><artifactId>starter-canal</artifactId><version>0.0.1-SNAPSHOT</version></dependency>

@SpringBootApplication
@EnableCanalClient
public class CanalApplication {public static void main(String[] args) {SpringApplication.run(CanalApplication.class,args);}
}

@CanalEventListener
public class CanalDataEventListener {/**** 增加数据监听* @param eventType* @param rowData*/@InsertListenPointpublic void onEventInsert(CanalEntry.EventType eventType, CanalEntry.RowData rowData) {rowData.getAfterColumnsList().forEach((c) -> System.out.println("By--Annotation: " + c.getName() + " ::   " + c.getValue()));}/**** 修改数据监听* @param rowData*/@UpdateListenPointpublic void onEventUpdate(CanalEntry.RowData rowData) {System.out.println("UpdateListenPoint");rowData.getAfterColumnsList().forEach((c) -> System.out.println("By--Annotation: " + c.getName() + " ::   " + c.getValue()));}/**** 删除数据监听* @param eventType*/@DeleteListenPointpublic void onEventDelete(CanalEntry.EventType eventType) {System.out.println("DeleteListenPoint");}/**** 自定义数据修改监听* @param eventType* @param rowData*/@ListenPoint(destination = "example", schema = "torlesse_test", table = {"tb_user", "tb_order"}, eventType = CanalEntry.EventType.UPDATE)public void onEventCustomUpdate(CanalEntry.EventType eventType, CanalEntry.RowData rowData) {System.err.println("DeleteListenPoint");rowData.getAfterColumnsList().forEach((c) -> System.out.println("By--Annotation: " + c.getName() + " ::   " + c.getValue()));}@ListenPoint(destination = "example",schema = "test_canal", //所要监听的数据库名table = {"tb_user"}, //所要监听的数据库表名eventType = {CanalEntry.EventType.UPDATE, CanalEntry.EventType.INSERT, CanalEntry.EventType.DELETE})public void onEventCustomUpdateForTbUser(CanalEntry.EventType eventType, CanalEntry.RowData rowData){getChangeValue(eventType,rowData);}public static void getChangeValue(CanalEntry.EventType eventType, CanalEntry.RowData rowData){if(eventType == CanalEntry.EventType.DELETE){rowData.getBeforeColumnsList().forEach(column -> {//获取删除前的数据System.out.println(column.getName() + " == " + column.getValue());});}else {rowData.getBeforeColumnsList().forEach(column -> {//打印改变前的字段名和值System.out.println(column.getName() + " == " + column.getValue());});rowData.getAfterColumnsList().forEach(column -> {//打印改变后的字段名和值System.out.println(column.getName() + " == " + column.getValue());});}}
}

案列四统计千亿级别PV

UV: Unique Visitor ,独立访客数，是指在一个统计周期内，访问网站的人数之和。一般理解客户ip，需要去重
PV : Page View，浏览量，是指在一个统计周期内，浏览页面的数之和。不需要去重
DAU: Daily Active User 日活跃用户数量；去重
DNU：Daily New User，日新增用户数
MAU：Monthly New User，月活跃用户；去重
需要使用redis hyperloglog基数统计数据结构来实现
基数统计：数据集中不重复的元素的个数

模拟后台1万用户点击首页

package com.redis.redis01.service;import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.util.StopWatch;
import javax.annotation.Resource;
import java.util.Random;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;@Slf4j
@Service
public class HyperLogService {@Resourceprivate RedisTemplate<String, Object> redisTemplate;/*** 模拟后台1万用户点击首页，每个用户来自不同的ip地址*/public void hyperloglogUvTest() {StopWatch stopWatch=new StopWatch();stopWatch.start();CountDownLatch countDownLatch=new CountDownLatch(10000);//主子线程传递共享连接资源redisTemplateExecutorService executorService = Executors.newFixedThreadPool(200);executorService.execute(new Runnable() {@Overridepublic void run() {//模拟1万用户for (int i = 0; i < 10000; i++) {countDownLatch.countDown();Random random = new Random();String ipAddress = random.nextInt(256)+ "." + random.nextInt(256)+ "." + random.nextInt(256)+ "." + random.nextInt(256);redisTemplate.opsForHyperLogLog().add("uv_click", ipAddress);System.out.println("countDownLatch=" + countDownLatch.getCount());}}});try {countDownLatch.await();stopWatch.stop();Long uvClick1 = redisTemplate.opsForHyperLogLog().size("uv_click");//用户访问首页次数uv=10059System.out.println("用户访问首页次数uv=" + uvClick1);//共耗时=3:秒System.out.println("共耗时=" + stopWatch.getTotalTimeMillis()/1000+":秒");} catch (InterruptedException e) {throw new RuntimeException(e);}}
}

案列五布隆过滤器方案实现

利用bitmap实现，一个bitmap=2^32bit最大能存512M，一个用户一天签到用1个bit，一年365个bit就可以实现，1千万个用户一年只需要435MB还不到一个bitmap最大存储能力
优点

高效地插入和查询，内存占用 bit 空间少

缺点

不能删除元素
因为删除元素会导致误判率增加，因为hash冲突同一个位置可能存的东西是多个共有的，你删除一个元素的同时可能也把其他的删除了

存在误判，不能精准过滤
有，可能有
无，绝对无

package com.redis.redis01.service;import com.google.common.collect.Lists;
import com.redis.redis01.bean.RedisBs;
import com.redis.redis01.mapper.RedisBsMapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;import javax.annotation.Resource;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.locks.ReentrantLock;@Slf4j
@Service
public class BitmapService {@Resourceprivate RedisTemplate<String, Object> redisTemplate;private static ReentrantLock lock = new ReentrantLock();@Autowiredprivate RedisBsMapper redisBsMapper;/*** 场景一：布隆过滤器解决缓存穿透问题(null/黑客攻击)；利用redis+bitmap实现* 有可能有，没有一定没有*                                                    无-------------》mysql查询*                     有--------》redis查询----------》有-----------》返回* 请求-----》布隆过滤器-----------》*                      无-------终止** @param type:0初始化，1常规查询*/public void booleanFilterBitmap(int type, Integer id) {switch (type) {case 0://初始化数据for (int i = 0; i < 10; i++) {RedisBs initBs = RedisBs.builder().id(i).name("赵三" + i).phone("1580080569" + i).build();//1 插入数据库redisBsMapper.insert(initBs);//2 插入redisredisTemplate.opsForValue().set("customer:info" + i, initBs);}//3 将用户id插入布隆过滤器中,作为白名单for (int i = 0; i < 10; i++) {String booleanKey = "customer:booleanFilter:" + i;//3.1 计算hashvalueint abs = Math.abs(booleanKey.hashCode());//3.2 通过abs和2的32次方取余，获得布隆过滤器/bitmap对应的下标坑位/indexlong index = (long) (abs % Math.pow(2, 32));log.info("坑位:{}", index);//3.3 设置redis里面的bitmap对应类型的白名单redisTemplate.opsForValue().setBit("whiteListCustomer", index, true);}break;case 1://常规查询//1 获取当前传过来的id对应的哈希值String inputBooleanKey = "customer:booleanFilter:" + id;int abs = Math.abs(inputBooleanKey.hashCode());long index = (long) (abs % Math.pow(2, 32));Boolean whiteListCustomer = redisTemplate.opsForValue().getBit("whiteListCustomer", index);//加入双检锁//加锁，大厂用，对于高qps的优化，进行加锁保证一个请求操作，让外面的redis等待一下，避免击穿mysqllock.lock();try {if (null == whiteListCustomer) {whiteListCustomer = redisTemplate.opsForValue().getBit("whiteListCustomer", index);if (null != whiteListCustomer && whiteListCustomer) {//布隆过滤器中存在，则可能存在//2 查找redisObject queryCustomer = redisTemplate.opsForValue().get("customer:info" + id);if (null != queryCustomer) {log.info("返回客户信息:{}", queryCustomer);break;} else {//3 redis没有查找mysqlRedisBs userById = redisBsMapper.findUserById(id);if (null != userById) {log.info("返回客户信息:{}", queryCustomer);redisTemplate.opsForValue().set("customer:info" + id, userById);break;} else {log.info("当前客户信息不存在:{}", id);break;}}} else {//redis没有，去mysql中查询//3 redis没有查找mysqlRedisBs userById = redisBsMapper.findUserById(id);if (null != userById) {log.info("返回客户信息:{}", userById);redisTemplate.opsForValue().set("customer:info" + id, userById);break;} else {log.info("当前客户信息不存在:{}", id);break;}}}} finally {lock.unlock();}log.info("当前客户信息不存在:{}", id);break;default:break;}}
}