Skip to content

N
nyx
Commit e155a72b by jscat
Options

nyx javaapp: 

添加redis支持
parent ba584bdc
正在显示 包含 291 行增加 和 60 行删除
app/nyx_app_key/api/src/main/java/cn/com/fun/nyxkey/api/config/WebSocketServer.java
......@@ -8,6 +8,7 @@ import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CopyOnWriteArraySet;
import javax.annotation.PostConstruct;
import javax.websocket.OnClose;
import javax.websocket.OnError;
import javax.websocket.OnMessage;
......@@ -16,8 +17,9 @@ import javax.websocket.Session;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import com.alibaba.fastjson.JSON;
import cn.com.fun.nyxkey.api.service.Rockwell_keyService;
import com.alibaba.fastjson.JSONObject;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
......@@ -39,8 +41,10 @@ import org.mortbay.util.UrlEncoded;
@Component
public class WebSocketServer {
private static final Logger log = LoggerFactory.getLogger(WebSocketServer.class);
@Autowired
private Rockwell_keyService keyService;
private static final Logger log = LoggerFactory.getLogger(WebSocketServer.class);
//静态变量，用来记录当前在线连接数。应该把它设计成线程安全的。
private static int onlineCount = 0;
......@@ -58,6 +62,20 @@ public class WebSocketServer {
//接收uid
private String uid="";
//所在城市
private String city="";
/*
用于在Utils工具类或者非controller普通类中使用@Autowired注解注入Service或者Mapper接口
因为Utils使用了静态的方法，我们是无法直接使用非静态接口的
refer: https://blog.csdn.net/wqc19920906/article/details/80009929
*/
public static WebSocketServer webSocketServer;
@PostConstruct
public void init() {
webSocketServer = this;
}
/**
* 连接建立成功调用的方法*/
@OnOpen
......@@ -86,6 +104,7 @@ public class WebSocketServer {
this.session = session;
this.sid = multiMap.getString("sid");
this.uid = multiMap.getString("uid");
this.city = multiMap.getString("city");
String msg = multiMap.getString("msg");
webSocketSet.add(this); //加入set中
addOnlineCount(this.sid); //在线数加1
......@@ -97,8 +116,8 @@ public class WebSocketServer {
msgMap.put("count", onlineCountMap.get(this.sid));
try {
//sendMessage(msgMap.toString());
sendInfo(msgMap.toString(), this.sid);
webSocketServer.keyService.Rockwell_keyServiceUpdateWatcherToRedis(sid, city, Double.valueOf(onlineCountMap.get(this.sid)));
} catch (IOException e) {
log.error("websocket IO异常");
}
......@@ -119,11 +138,12 @@ public class WebSocketServer {
msgMap.put("count", onlineCountMap.get(this.sid));
try {
//sendMessage(msgMap.toString());
sendInfo(msgMap.toString(), this.sid);
webSocketServer.keyService.Rockwell_keyServiceUpdateWatcherToRedis(sid, city, Double.valueOf(onlineCountMap.get(this.sid)));
} catch (IOException e) {
log.error("websocket IO异常");
}
}
/**
......
app/nyx_app_key/api/src/main/java/cn/com/fun/nyxkey/api/repository/Rockwell_keyMapper.java
......@@ -3,10 +3,10 @@ package cn.com.fun.nyxkey.api.repository;
import cn.com.fun.nyxkey.api.common.Pagination;
import cn.com.fun.nyxkey.api.domain.*;
import cn.com.fun.nyxkey.api.domain.query.*;
import org.apache.ibatis.annotations.Param;
import org.springframework.stereotype.Repository;
import java.util.List;
import java.util.Map;
/**
* Created by jscat on 2019-11-11.
......@@ -78,6 +78,8 @@ public interface Rockwell_keyMapper {
/* live api */
List<Rockwell_keyV_live_info> getV_live_info(Rockwell_keyT_liveQuery rockwell_keyT_liveQuery);
int addT_live(Rockwell_keyV_live_info rockwell_keyV_live_info);
int updateT_liveNum_watcherByRedis(List<Map<String, String>> list);
int updateT_live(Rockwell_keyV_live_info rockwell_keyV_live_info);
/* testing */
void test(Rockwell_keyT_matchQuery rockwell_keyT_matchQuery);
......
app/nyx_app_key/api/src/main/java/cn/com/fun/nyxkey/api/service/Rockwell_keyService.java
......@@ -6,10 +6,12 @@ import cn.com.fun.nyxkey.api.domain.*;
import cn.com.fun.nyxkey.api.domain.query.*;
import com.alibaba.fastjson.JSON;
import org.joda.time.DateTime;
import redis.clients.jedis.Tuple;
import java.io.UnsupportedEncodingException;
import java.util.Date;
import java.util.List;
import java.util.Set;
/**
* Created by jscat on 2019-11-11.
......@@ -97,7 +99,11 @@ public interface Rockwell_keyService {
jscat 20201127
*/
JSONResult Rockwell_keyServiceGetLives(String orderType, int pageNum, int pageCount);
void Rockwell_keyServiceUpdateWatcherToRedis(String sid, String city, Double count);
int Rockwell_keyServiceUpdateWatcherByRedisToDB(Set<Tuple> set);
JSONResult Rockwell_keyServiceGetLivesByRedis(String city, String title, String orderType, int pageNum, int pageCount);
JSONResult Rockwell_keyServiceAddLive(String title, String city, String address, String noteImage, String liveId);
int Rockwell_keyServiceUpdateLive(String title, String liveStatus, String userId);
// testing
void Rockwell_keyServiceTest(List<Rockwell_keyT_match> listRockwell_keyT_match);
......
app/nyx_app_key/api/src/main/java/cn/com/fun/nyxkey/api/service/impl/Rockwell_keyServiceImpl.java
......@@ -7,10 +7,9 @@ import cn.com.fun.nyxkey.api.repository.*;
import cn.com.fun.nyxkey.api.service.*;
import cn.com.fun.nyxkey.api.domain.*;
import cn.com.fun.nyxkey.api.domain.query.*;
import cn.com.fun.nyxkey.api.utils.JedisUtil;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang3.StringUtils;
import org.checkerframework.checker.units.qual.K;
import org.joda.time.DateTime;
import org.mortbay.util.UrlEncoded;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
......@@ -23,11 +22,14 @@ import org.springframework.transaction.annotation.Transactional;
import javax.servlet.http.HttpServletResponse;
import javax.sql.DataSource;
import java.io.UnsupportedEncodingException;
import java.lang.reflect.Array;
import java.net.URLDecoder;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.*;
import redis.clients.jedis.*;
/**
* Created by jscat on 2019-11-11.
*/
......@@ -39,6 +41,9 @@ public class Rockwell_keyServiceImpl implements Rockwell_keyService {
@Autowired
private Rockwell_keyMapper rockwell_keyMapper;
@Autowired
JedisUtil jedisUtil;
@Value("${aliyun.oss.endpoint}")
private String aliyunOssEndpoint;
......@@ -1017,6 +1022,107 @@ public class Rockwell_keyServiceImpl implements Rockwell_keyService {
}
/*
zset/hash 不设置过期时间 jscat 20201202
假设redis里 在线人数 数据永远是最新
获取4条人数最多现场 之redis方案
1. 从redis里读取4条数据,
2.1 if title == "0", 则直接从redis读取
2.2 if title != "", 查看redis,
if len<4 && pageNum == 1, 则redis write-through(), 再从数据库读取数据, 写入redis
3. 返回数据
*/
public JSONResult Rockwell_keyServiceGetLivesByRedis(String city, String title, String orderType, int pageNum, int pageCount)
{
LOGGER.debug("find Rockwell_keyServiceGetLivesByRedis");
List<Rockwell_keyV_live_info> listRockwell_keyV_live_info = new ArrayList<>();
long start = (pageNum - 1) * pageCount;
long end = (pageNum ) * pageCount - 1;
String keyname = "nyx:live:num_watcher";
String strKeyZset = keyname + ":zset:" + city;
String strKeyHash = keyname + ":hash:" + city;
String strKeyZset_title = strKeyZset;
String strKeyHash_title = strKeyHash;
if( !title.equals("0") )
{
strKeyZset_title = strKeyZset + "_" + title;
strKeyHash_title = strKeyHash + "_" + title;
}
Set<Tuple> set = jedisUtil.zrevrangeWithScores(strKeyZset_title, start, end);
//从数据库读取
if( !title.equals("0") && (set == null || set.size() < pageCount) && pageNum == 1)
{
//step2: 先更新num_watcher数据,否则会造成数据紊乱 write-through
Set<Tuple> setAll = jedisUtil.zrevrangeWithScores(strKeyZset, 0, -1);
int ret = Rockwell_keyServiceUpdateWatcherByRedisToDB(setAll);
Rockwell_keyT_liveQuery rockwell_keyT_liveQuery = new Rockwell_keyT_liveQuery();
//step3: 查询数据
rockwell_keyT_liveQuery.setTitle(title);
rockwell_keyT_liveQuery.setCity(city);
rockwell_keyT_liveQuery.setOrderType(orderType);
rockwell_keyT_liveQuery.setPagination(new Pagination(pageNum, pageCount));
listRockwell_keyV_live_info = rockwell_keyMapper.getV_live_info(rockwell_keyT_liveQuery);
JSONObject jsonObject = new JSONObject();
for(Rockwell_keyV_live_info rockwell_keyV_live_info : listRockwell_keyV_live_info)
{
jsonObject.put("title", rockwell_keyV_live_info.getTitle());
jsonObject.put("city", rockwell_keyV_live_info.getCity());
jsonObject.put("address", rockwell_keyV_live_info.getAddress());
jsonObject.put("nyxName", rockwell_keyV_live_info.getNyxName());
jsonObject.put("avatarUrl", rockwell_keyV_live_info.getAvatarUrl());
jsonObject.put("noteImage", rockwell_keyV_live_info.getNoteImage());
Double score = Double.valueOf(rockwell_keyV_live_info.getNumWatcher());
jedisUtil.zadd(strKeyZset_title, score, rockwell_keyV_live_info.getLiveId());
jedisUtil.hset(strKeyHash_title, rockwell_keyV_live_info.getLiveId(), jsonObject.toJSONString());
}
jedisUtil.expire(strKeyZset_title, 15*60, 0); //tips: strKeyZset不设置expire
jedisUtil.expire(strKeyHash_title, 15*60, 0);
set = jedisUtil.zrevrangeWithScores(strKeyZset_title, start, end);
}
//step4: 得到liveId keys
Set<String> keys = new HashSet<>();
for (Tuple key : set) {
keys.add(key.getElement());
}
//step5: set 序列化
List<String> valueList = jedisUtil.hmget(strKeyHash_title, 0, setToStrings(keys));
JSONObject jSONObject = new JSONObject();
List<JSONObject> obj = new ArrayList<JSONObject>();
int i = 0;
for (Tuple key : set) {
jSONObject = JSONObject.parseObject(valueList.get(i));
jSONObject.put("numWatcher", key.getScore());
jSONObject.put("liveId", key.getElement());
i++;
obj.add(jSONObject);
}
int totalCount = obj.size();
JSONResult jsonResult = new JSONResult(totalCount, obj);
return jsonResult;
}
public void Rockwell_keyServiceUpdateWatcherToRedis(String sid, String city, Double count)
{
String keyname = "nyx:live:num_watcher:zset:"+city;
if( sid.startsWith("uid") )
{
jedisUtil.zadd( keyname, count, sid);
}
}
public JSONResult Rockwell_keyServiceGetLives(String orderType, int pageNum, int pageCount)
{
LOGGER.debug("find Rockwell_keyServiceGetLives");
......@@ -1065,6 +1171,49 @@ public class Rockwell_keyServiceImpl implements Rockwell_keyService {
}
}
public int Rockwell_keyServiceUpdateWatcherByRedisToDB(Set<Tuple> set)
{
int ret = 0;
if(set != null && set.size() != 0)
{
List<Map<String, String>> list = new ArrayList<>();
//step1: get liveId set
for (Tuple key : set) {
Map<String, String> map = new HashMap<String, String>();
map.put("sid", key.getElement());
map.put("score", String.valueOf(key.getScore()));
list.add(map);
}
//更新数据
ret = rockwell_keyMapper.updateT_liveNum_watcherByRedis(list);
}
return ret;
}
public int Rockwell_keyServiceUpdateLive(String title, String liveStatus, String userId)
{
Rockwell_keyV_live_info rockwell_keyV_live_info = new Rockwell_keyV_live_info();
rockwell_keyV_live_info.setLiveId(userId);
if(!liveStatus.equals("0"))
{
rockwell_keyV_live_info.setLiveStatus(liveStatus);
}
if(!title.equals("0"))
{
rockwell_keyV_live_info.setTitle(title);
}
//更新数据
return rockwell_keyMapper.updateT_live(rockwell_keyV_live_info);
}
// testing
public void Rockwell_keyServiceTest(List<Rockwell_keyT_match> listRockwell_keyT_match)
......@@ -1074,4 +1223,9 @@ public class Rockwell_keyServiceImpl implements Rockwell_keyService {
rockwell_keyMapper.test(rockwell_keyT_matchQuery);
}
public String[] setToStrings(Set<String> keys){
List<String> list = new ArrayList<>(keys);
return list.toArray(new String[list.size()]);
}
}
app/nyx_app_key/api/src/main/java/cn/com/fun/nyxkey/api/utils/JedisUtil.java
......@@ -12,10 +12,7 @@ import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.ListPosition;
import redis.clients.jedis.SortingParams;
import redis.clients.jedis.*;
@Component
@Slf4j
......@@ -1970,6 +1967,21 @@ public class JedisUtil{
return res;
}
public Set<Tuple> zrevrangeWithScores(String key, long start, long end) {
Jedis jedis = null;
Set<Tuple> res = null;
try {
jedis = jedisPool.getResource();
res = jedis.zrevrangeWithScores(key, start, end);
} catch (Exception e) {
log.error(e.getMessage());
} finally {
returnResource(jedisPool, jedis);
}
return res;
}
/**
* <p>
* 通过key返回指定score内zset中的value
......
app/nyx_app_key/api/src/main/java/cn/com/fun/nyxkey/api/web/controller/NyxApiController.java
......@@ -675,16 +675,21 @@ public class NyxApiController {
- 修改记录: client -> java -> redis(当前人数, 峰值人数)
- 关闭记录: client -> java -> redis(峰值人数) -> java(live_status="00")
- 增加记录: client -> java -> redis (当前人数, 峰值人数)
8.1 live_info 的相关查询
1. 按照 user_id
2. 按照 member_id
status == '01' && member_status == '01'
order by default_member desc, create_datetime asc
返回 Rockwell_keyV_member_user
8.1 查询city最多4场现场， 按orderType排序
live_info 的相关查询
1. 按照 city
2. 按照 title
3. 按照 orderType, default is num_watcher desc
order by num_watcher desc
返回 Rockwell_keyV_live_info
*/
@ApiOperation(value="按page获取live信息", notes="按page获取live信息")
@ApiImplicitParams({
@ApiImplicitParam(name = "city", value = "按城市筛选", required = false, dataType = "String"),
@ApiImplicitParam(name = "title", value = "按关键词搜索", required = false, dataType = "String"),
@ApiImplicitParam(name = "orderType", value = "排序类型, 默认desc", required = false, dataType = "String"),
@ApiImplicitParam(name = "pageNum", value = "页码，从1开始", required = false, dataType = "String", defaultValue = "1"),
@ApiImplicitParam(name = "pageCount", value = "每页的数据个数，默认是4个", required = false, dataType = "String", defaultValue = "4")
......@@ -692,12 +697,14 @@ public class NyxApiController {
})
@RequestMapping(value = "/nyx/live/query", method = RequestMethod.GET)
public JSONResult RockwellGetLives(
@RequestParam(value = "city", required = false, defaultValue = "上海") String city,
@RequestParam(value = "title", required = false, defaultValue = "0") String title,
@RequestParam(value = "orderType", required = false, defaultValue = "desc") String orderType,
@RequestParam(value = "pageNum", required = false, defaultValue = "1") int pageNum,
@RequestParam(value = "pageCount", required = false, defaultValue = "4") int pageCount
) {
return keyService.Rockwell_keyServiceGetLives(orderType, pageNum, pageCount);
return keyService.Rockwell_keyServiceGetLivesByRedis(city, title, orderType, pageNum, pageCount);
}
/*
......@@ -724,6 +731,27 @@ public class NyxApiController {
return keyService.Rockwell_keyServiceAddLive(title, city, address, noteImage, liveId);
}
/*
8.2 更新live信息
*/
@ApiOperation(value="更新live信息", notes="更新live信息")
@ApiImplicitParams({
@ApiImplicitParam(name = "userId", value = "用户id", required = true, dataType = "String"),
@ApiImplicitParam(name = "title", value = "直播间名称", required = false, dataType = "String"),
@ApiImplicitParam(name = "liveStatus", value = "直播状态", required = false, dataType = "String")
})
@RequestMapping(value = "/nyx/live/update", method = RequestMethod.GET)
public int RockwellUpdateLive(
@RequestParam(value = "userId", required = true, defaultValue = "0") String userId,
@RequestParam(value = "title", required = false, defaultValue = "0") String title,
@RequestParam(value = "liveStatus", required = false, defaultValue = "0") String liveStatus
) {
return keyService.Rockwell_keyServiceUpdateLive(title, liveStatus, userId);
}
// 2.0 test
@ApiOperation(value="test", notes="test")
@ApiImplicitParams({
......
app/nyx_app_key/api/src/main/java/cn/com/fun/nyxkey/api/web/controller/RedisController.java
......@@ -3,6 +3,7 @@ package cn.com.fun.nyxkey.api.web.controller;
import cn.com.fun.nyxkey.api.common.ExceptionMsg;
import cn.com.fun.nyxkey.api.common.JSONResult;
import cn.com.fun.nyxkey.api.utils.JedisUtil;
import io.swagger.annotations.ApiImplicitParam;
import io.swagger.annotations.ApiImplicitParams;
import io.swagger.annotations.ApiOperation;
import org.slf4j.Logger;
......@@ -116,7 +117,7 @@ public class RedisController {
address: "",
nyxName: "",
avatarUrl: "",
note_image: "",
noteImage: "",
}
......@@ -130,51 +131,22 @@ public class RedisController {
*/
@ApiOperation(value="更新在线人数", notes="更新在线人数")
@ApiImplicitParams({
@ApiImplicitParam(name = "sid", value = "sid", required = false, dataType = "String"),
@ApiImplicitParam(name = "city", value = "city", required = false, dataType = "String"),
@ApiImplicitParam(name = "score", value = "score", required = false, dataType = "String"),
})
@RequestMapping(value = "/nyx/redis/update/watcher", method = RequestMethod.GET)
public JSONResult updateWatcher() throws Exception {
// 返回json显示的话 把MAP转json toJsonFromMap
String pattern = "[@#]*";
Set<String> keys = jedisUtil.keys(pattern);
Map<String, String> map = new HashMap<>();
List<String> list = new ArrayList<>(keys);
List<String> valueList = jedisUtil.mget(list.toArray(new String[list.size()]));
int i = 0;
for (String key : keys) {
map.put(key, valueList.get(i));
i++;
}
return new JSONResult(i, map);
}
/*
refer:
获取在线人数
*/
@ApiOperation(value="获取在线人数", notes="获取在线人数")
@ApiImplicitParams({
})
@RequestMapping(value = "/nyx/redis/get/watcher", method = RequestMethod.GET)
public JSONResult getWatcher() throws Exception {
public JSONResult updateWatcher(
@RequestParam(value = "sid", required = true, defaultValue = "0") String sid,
@RequestParam(value = "city", required = true, defaultValue = "上海") String city,
@RequestParam(value = "score", required = true, defaultValue = "0") String score
) throws Exception {
// 返回json显示的话 把MAP转json toJsonFromMap
String pattern = "[@#]*";
Set<String> keys = jedisUtil.keys(pattern);
String keyname = "nyx:live:num_watcher";
Map<String, String> map = new HashMap<>();
List<String> list = new ArrayList<>(keys);
List<String> valueList = jedisUtil.mget(list.toArray(new String[list.size()]));
int i = 0;
for (String key : keys) {
map.put(key, valueList.get(i));
i++;
}
jedisUtil.zadd(keyname+":zset:"+city, Double.valueOf(score), sid);
return new JSONResult(i, map);
return new JSONResult();
}
......
app/nyx_app_key/api/src/main/resources/cn/com/fun/nyxkey/api/repository/Rockwell_keyMapper.xml
......@@ -707,6 +707,12 @@ limitations under the License.
<if test="liveStatus != null and liveStatus != '' ">
AND live_status=#{liveStatus}
</if>
<if test="city != null and city != '' ">
AND city=#{city}
</if>
<if test="title != null and title != '' ">
AND ( title LIKE CONCAT('%', #{title} ,'%') OR city LIKE CONCAT('%', #{title} ,'%') OR address LIKE CONCAT('%', #{title} ,'%') )
</if>
<if test="orderType != null and orderType != '' ">
order by num_watcher desc
</if>
......@@ -733,5 +739,35 @@ limitations under the License.
)
</insert>
<update id="updateT_liveNum_watcherByRedis" parameterType="java.util.List">
update tbl_live
<trim prefix="set" suffixOverrides=",">
<trim prefix="num_watcher =case" suffix="end,">
<foreach collection="list" item="item" index="index">
when live_id=#{item.sid} then #{item.score}
</foreach>
</trim>
</trim>
where live_id in
<foreach collection="list" index="index" item="item" separator="," open="(" close=")">
#{item.sid}
</foreach>
</update>
<!-- 更新t_live -->
<update id="updateT_live" parameterType="cn.com.fun.nyxkey.api.domain.Rockwell_keyV_live_info">
update tbl_live
<set>
<if test="title != null and title != '' ">
title=#{title},
</if>
<if test="liveStatus != null and liveStatus != '' ">
live_status=#{liveStatus},
</if>
</set>
where 1=1
AND live_id=#{liveId}
</update>
<!-- ending API interface -->
</mapper>
\ No newline at end of file
Markdown 格式
0% 或
您添加了 0 人 到此讨论。请谨慎行事。
请先完成此评论的编辑！
请 注册 或者 后发表评论