序
本文主要是对第二章的购物车服务的代码从jredis改为SpringBoot的redis template版本。
主要功能
- 存储登录的用户
- 存储最近登录的用户列表
- 存储用户最近浏览的项目
- 存储用户的购物车
- 缓存请求内容/数据行
数据结构选择
- 用map存储登陆用户
- 用zset存储最近登陆的用户
- 用zset存储最近被浏览的item
- 用zset存储用户最近浏览的item
- 用map存储用户的购物车
常量声明
/**
* 登录用户
* 数据结构 -- map
* key -- loginMap
* value -- k:token v:user
*/
public static final String KEY_LOGIN_USER = "loginMap";
/**
* 最近登录用户
* 数据结构 -- zset
* key -- recentSet
* value -- v:token score:timestamp
*/
public static final String KEY_RECENT_USER = "recentSet";
/**
* 项目浏览计数
*/
public static final String KEY_ITEM_VIEW_COUNT = "itemViewedZSet";
/**
* 存储用户最近浏览的项目
* 数据结构 -- zset
* key -- viewZset:token
* value -- v:item score:timestamp
*/
public static final String KEY_USER_VIEW_PREFIX = "viewZset:";
/**
* 用户购物车
* 数据结构 -- map
* key -- cartMap:session
* value -- k:item v:count
*/
public static final String KEY_USER_CART_PREFIX = "cartMap:";
/**
* 请求的缓存
* 数据结构 -- string
* key -- cache:hashcode
* value -- string
*/
public static final String KEY_CACHE_PREFIX = "cache:";
/**
* 缓存库存信息
* key -- inventory:rowId
* value -- json
*/
public static final String KEY_INVENTORY_PREFIX = "inventory:";
/**
* 调度ZSet
* key -- scheduleZSet
* value -- k:rowId v:timestamp
*/
public static final String KEY_SCHEDULE = "scheduleZSet";
/**
* 到期ZSet
* key -- delayZSet
* value -- k:rowId v:timestamp
*/
public static final String KEY_DELAY = "delayZSet";主要功能
用户浏览item
/**
* 用户浏览项目
* @param token
* @param user
* @param item
*/
public void viewItem(String token,String user,String item){
long timestamp = System.currentTimeMillis()/1000;
//模拟登录下
redisTemplate.opsForHash().put(KEY_LOGIN_USER, token, user);
//更新最近登录
redisTemplate.boundZSetOps(KEY_RECENT_USER).add(token,timestamp);
if(item == null){
return ;
}
String userViewKey = formUserViewKey(token);
//添加最近浏览记录
redisTemplate.boundZSetOps(userViewKey).add(item,timestamp);
//缩减下最近浏览记录,保持在25条
redisTemplate.boundZSetOps(userViewKey).removeRange(0,-26);
//对项目浏览得分-1,最后升序排
redisTemplate.boundZSetOps(KEY_ITEM_VIEW_COUNT).incrementScore(item,-1);
}用户添加购物车
/**
* 添加到购物车
* @param session
* @param item
* @param count
*/
public void addToCart(String session,String item,int count){
String cartKey = formUserCartKey(session);
if(count <= 0){
redisTemplate.opsForHash().delete(cartKey,item);
}else{
redisTemplate.opsForHash().put(cartKey, item, String.valueOf(count));
}
}缓存请求
/**
* 缓存请求
* @param request
* @param supplier
* @return
*/
public String cacheRequest(String request,Supplier<String> supplier){
if(!canCache(request)){
//不走缓存
return supplier.get();
}
String pageKey = formCacheKey(request);
ValueOperations<String,String> ops = redisTemplate.opsForValue();
String content = ops.get(pageKey);
if(content == null && supplier != null){
//缓存不存在
content = supplier.get();
redisTemplate.opsForValue().setIfAbsent(pageKey,content);
}
return content;
}
/**
* 判断需不需要缓存该请求
* 浏览量上w的才请求
* @param request
* @return
*/
public boolean canCache(String request){
try {
URL url = new URL(request);
Map<String,String> params = new HashMap<String,String>();
if (url.getQuery() != null){
for (String param : url.getQuery().split("&")){
String[] pair = param.split("=", 2);
params.put(pair[0], pair.length == 2 ? pair[1] : null);
}
}
String itemId = params.get("item");
if (itemId == null || params.containsKey("_")) {
return false;
}
Long rank = redisTemplate.boundZSetOps(KEY_ITEM_VIEW_COUNT).rank(itemId);
return rank != null && rank < 10000;
}catch(MalformedURLException mue){
return false;
}
}缓存请求行
/**
* 缓存数据行
* 1,取出schedule到期的数据项
* 2,取出该数据项的过期时间
* 3,更新该数据项的过期时间
*/
class CacheRowsTask extends Thread{
private volatile boolean stop = false;
public void quit(){
stop = true;
}
@Override
public void run() {
Gson gson = new Gson();
while (!stop){
//取第一个出来
Set<ZSetOperations.TypedTuple> range = redisTemplate.boundZSetOps(KEY_SCHEDULE).rangeWithScores(0,0);
ZSetOperations.TypedTuple next = range.size() > 0 ? range.iterator().next() : null;
long now = System.currentTimeMillis() / 1000;
if (next == null || next.getScore() > now){
try {
sleep(50);
}catch(InterruptedException ie){
Thread.currentThread().interrupt();
}
continue;
}
String rowId = (String) next.getValue();
double delay = redisTemplate.boundZSetOps(KEY_DELAY).score(rowId);
if (delay <= 0) {
redisTemplate.boundZSetOps(KEY_DELAY).remove(rowId);
redisTemplate.boundZSetOps(KEY_SCHEDULE).remove(rowId);
redisTemplate.delete(formInventoryKey(rowId));
continue;
}
Inventory row = Inventory.get(rowId);
redisTemplate.opsForZSet().add(KEY_SCHEDULE, rowId, now + delay);
redisTemplate.opsForValue().set(formInventoryKey(rowId), gson.toJson(row));
}
}
}
/**
* 被缓存的项
*/
static class Inventory {
private String id;
private String data;
private long time;
private Inventory (String id) {
this.id = id;
this.data = "data to cache...";
this.time = System.currentTimeMillis() / 1000;
}
public static Inventory get(String id) {
return new Inventory(id);
}
}缓存调度
/**
* 初始化缓存调度
* @param rowId
* @param delay
*/
public void scheduleRowCache(String rowId, int delay) {
redisTemplate.opsForZSet().add(KEY_DELAY,rowId,delay);
redisTemplate.opsForZSet().add(KEY_SCHEDULE,rowId,System.currentTimeMillis() / 1000);
}单元测试
public class ShoppingServiceTest extends RedisdemoApplicationTests{
@Autowired
ShoppingService shoppingService;
@Autowired
RedisTemplate redisTemplate;
@Test
public void loginCookies() throws InterruptedException {
System.out.println("\n----- testLoginCookies -----");
String token = UUID.randomUUID().toString();
shoppingService.viewItem(token, "username", "itemX");
System.out.println("We just logged-in/updated token: " + token);
System.out.println("For user: 'username'");
System.out.println();
System.out.println("What username do we get when we look-up that token?");
String r = shoppingService.getLgoinUserByToken(token);
System.out.println(r);
System.out.println();
Assert.assertNotNull(r);
System.out.println("Let's drop the maximum number of cookies to 0 to clean them out");
System.out.println("We will start a thread to do the cleaning, while we stop it later");
shoppingService.startCleanSessionTask();
long s = redisTemplate.opsForHash().size(ShoppingService.KEY_LOGIN_USER);
System.out.println("The current number of sessions still available is: " + s);
Assert.assertTrue(s == 0);
}
@Test
public void shoppingCartCookies() throws InterruptedException {
System.out.println("\n----- testShopppingCartCookies -----");
String token = UUID.randomUUID().toString();
System.out.println("We'll refresh our session...");
shoppingService.viewItem(token, "username", "itemX");
System.out.println("And add an item to the shopping cart");
shoppingService.addToCart(token, "itemY", 3);
Map<String,String> r = redisTemplate.opsForHash().entries(shoppingService.formUserCartKey(token));
System.out.println("Our shopping cart currently has:");
for (Map.Entry<String,String> entry : r.entrySet()){
System.out.println(" " + entry.getKey() + ": " + entry.getValue());
}
System.out.println();
Assert.assertTrue(r.size() >= 1);
System.out.println("Let's clean out our sessions and carts");
shoppingService.startCleanSessionTask();
r = redisTemplate.opsForHash().entries(shoppingService.formUserCartKey(token));
System.out.println("Our shopping cart now contains:");
for (Map.Entry<String,String> entry : r.entrySet()){
System.out.println(" " + entry.getKey() + ": " + entry.getValue());
}
Assert.assertTrue(r.size() == 0);
}
@Test
public void cacheRequest(){
System.out.println("\n----- testCacheRequest -----");
String token = UUID.randomUUID().toString();
shoppingService.viewItem(token, "username", "itemX");
String url = "http://test.com/?item=itemX";
System.out.println("We are going to cache a simple request against " + url);
String result = shoppingService.cacheRequest(url, () -> "content for " + url);
System.out.println("We got initial content:\n" + result);
System.out.println();
Assert.assertNotNull(result);
System.out.println("To test that we've cached the request, we'll pass a bad callback");
String result2 = shoppingService.cacheRequest(url, null);
System.out.println("We ended up getting the same response!\n" + result2);
Assert.assertTrue(result.equals(result2));
Assert.assertFalse(shoppingService.canCache("http://test.com/"));
Assert.assertFalse(shoppingService.canCache("http://test.com/?item=itemX&_=1234536"));
}
@Test
public void cacheRows() throws InterruptedException {
System.out.println("\n----- testCacheRows -----");
System.out.println("First, let's schedule caching of itemX every 5 seconds");
shoppingService.scheduleRowCache("itemX", 5);
System.out.println("Our schedule looks like:");
Set<ZSetOperations.TypedTuple> range = redisTemplate.boundZSetOps(ShoppingService.KEY_SCHEDULE).rangeWithScores(0, -1);
for (ZSetOperations.TypedTuple tuple : range){
System.out.println(" " + tuple.getValue() + ", " + tuple.getScore());
}
Assert.assertTrue(range.size() != 0);
System.out.println("We'll start a caching thread that will cache the data...");
shoppingService.startCacheRowTask();
Thread.sleep(1000);
System.out.println("Our cached data looks like:");
String r = (String) redisTemplate.opsForValue().get(shoppingService.formInventoryKey("itemX"));
System.out.println(r);
Assert.assertNotNull(r);
System.out.println();
System.out.println("We'll check again in 5 seconds...");
Thread.sleep(5000);
System.out.println("Notice that the data has changed...");
String r2 = (String) redisTemplate.opsForValue().get(shoppingService.formInventoryKey("itemX"));
System.out.println(r2);
System.out.println();
Assert.assertNotNull(r2);
Assert.assertFalse(r.equals(r2));
System.out.println("Let's force un-caching");
shoppingService.scheduleRowCache("itemX", -1);
Thread.sleep(1000);
r = (String) redisTemplate.opsForValue().get(shoppingService.formInventoryKey("itemX"));
System.out.println("The cache was cleared? " + (r == null));
Assert.assertNull(r);
}
}