记得上下班打卡 | git大法好，push需谨慎

Skip to content

L
liquidnet-bus-v1
Commit 2d29b94d authored by zhanggb's avatar zhanggb
Browse files
Options

~queue:ChimeQueue.USER_OPERATION_LIKE; 

~queue:ChimeQueue.USER_OPERATION_DISLIKE;
parent 8bd1a3cf
Expand all Hide whitespace changes
Inline Side-by-side
Showing with 718 additions and 409 deletions
liquidnet-bus-common/liquidnet-common-service-base/src/main/java/com/liquidnet/service/base/constant/MQConst.java
View file @ 2d29b94d
......@@ -389,7 +389,5 @@ public class MQConst {
}
public static void main(String[] args) {
System.out.println(ChimeQueue.USER_OPERATION_LIKE.name());
}
}
liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-adam/src/main/java/com/liquidnet/service/consumer/adam/receiver/AbstractChimeRedisReceiver.java
View file @ 2d29b94d
package com.liquidnet.service.consumer.adam.receiver;
import com.liquidnet.commons.lang.util.CollectionUtil;
import com.liquidnet.commons.lang.util.JsonUtils;
import com.liquidnet.service.chime.vo.mongo.ChimeUserOperLogVo;
import com.liquidnet.service.consumer.adam.util.ChimeDataUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.StreamRecords;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.stream.StreamListener;
import java.util.HashMap;
@Slf4j
public abstract class AbstractChimeRedisReceiver implements StreamListener<String, MapRecord<String, String, String>> {
@Autowired
StringRedisTemplate stringRedisTemplate;
@Autowired
private ChimeDataUtils chimeDataUtils;
@Override
public void onMessage(MapRecord<String, String, String> message) {
String redisStreamKey = this.getRedisStreamKey();
log.debug("CONSUMER MSG[streamKey:{},messageId:{},stream:{},body:{}]", redisStreamKey, message.getId(), message.getStream(), message.getValue());
boolean result = this.consumerMessageHandler(message.getValue().get("message"));
log.info("CONSUMER MSG RESULT:{} ==> [{}]MESSAGE_ID:{}", result, redisStreamKey, message.getId());
try {
stringRedisTemplate.opsForStream().acknowledge(getRedisStreamGroup(), message);
} catch (Exception e) {
log.error("#CONSUMER MSG EX_ACK ==> [{}]RESULT:{},MESSAGE:{}", redisStreamKey, result, message.getValue(), e);
}
try {
stringRedisTemplate.opsForStream().delete(redisStreamKey, message.getId());
} catch (Exception e) {
log.error("#CONSUMER MSG EX_DEL ==> [{}]RESULT:{},MESSAGE:{}", redisStreamKey, result, message.getValue(), e);
}
}
private boolean consumerMessageHandler(String msg) {
boolean aBoolean = false;
try {
ChimeUserOperLogVo textMessage = JsonUtils.fromJson(msg, ChimeUserOperLogVo.class);
if (textMessage == null) {
aBoolean = true;
} else {
//执行计数
chimeDataUtils.updateChimeUser(textMessage);
//创建操作日志
chimeDataUtils.createUserOperLog(textMessage);
aBoolean = true;
}
} catch (Exception e) {
log.error("CONSUMER MSG EX_HANDLE ==> [{}]:{}", this.getRedisStreamKey(), msg, e);
} finally {
if (!aBoolean) {
HashMap<String, String> map = CollectionUtil.mapStringString();
map.put("message", msg);
stringRedisTemplate.opsForStream().add(StreamRecords.mapBacked(map).withStreamKey(this.getRedisStreamKey()));
}
}
return aBoolean;
}
protected abstract String getRedisStreamKey();
protected abstract String getRedisStreamGroup();
}
//package com.liquidnet.service.consumer.adam.receiver;
//
//import com.liquidnet.commons.lang.util.CollectionUtil;
//import com.liquidnet.commons.lang.util.JsonUtils;
//import com.liquidnet.service.chime.vo.mongo.ChimeUserOperLogVo;
//import com.liquidnet.service.consumer.adam.util.ChimeDataUtils;
//import lombok.extern.slf4j.Slf4j;
//import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.data.redis.connection.stream.MapRecord;
//import org.springframework.data.redis.connection.stream.StreamRecords;
//import org.springframework.data.redis.core.StringRedisTemplate;
//import org.springframework.data.redis.stream.StreamListener;
//
//import java.util.HashMap;
//
//@Slf4j
//public abstract class AbstractChimeRedisReceiver implements StreamListener<String, MapRecord<String, String, String>> {
// @Autowired
// StringRedisTemplate stringRedisTemplate;
//
// @Autowired
// private ChimeDataUtils chimeDataUtils;
//
// @Override
// public void onMessage(MapRecord<String, String, String> message) {
// String redisStreamKey = this.getRedisStreamKey();
// log.debug("CONSUMER MSG[streamKey:{},messageId:{},stream:{},body:{}]", redisStreamKey, message.getId(), message.getStream(), message.getValue());
// boolean result = this.consumerMessageHandler(message.getValue().get("message"));
// log.info("CONSUMER MSG RESULT:{} ==> [{}]MESSAGE_ID:{}", result, redisStreamKey, message.getId());
//
// try {
// stringRedisTemplate.opsForStream().acknowledge(getRedisStreamGroup(), message);
// } catch (Exception e) {
// log.error("#CONSUMER MSG EX_ACK ==> [{}]RESULT:{},MESSAGE:{}", redisStreamKey, result, message.getValue(), e);
// }
// try {
// stringRedisTemplate.opsForStream().delete(redisStreamKey, message.getId());
// } catch (Exception e) {
// log.error("#CONSUMER MSG EX_DEL ==> [{}]RESULT:{},MESSAGE:{}", redisStreamKey, result, message.getValue(), e);
// }
// }
//
// private boolean consumerMessageHandler(String msg) {
// boolean aBoolean = false;
// try {
// ChimeUserOperLogVo textMessage = JsonUtils.fromJson(msg, ChimeUserOperLogVo.class);
// if (textMessage == null) {
// aBoolean = true;
// } else {
// //执行计数
// chimeDataUtils.updateChimeUser(textMessage);
// //创建操作日志
// chimeDataUtils.createUserOperLog(textMessage);
// aBoolean = true;
// }
// } catch (Exception e) {
// log.error("CONSUMER MSG EX_HANDLE ==> [{}]:{}", this.getRedisStreamKey(), msg, e);
// } finally {
// if (!aBoolean) {
// HashMap<String, String> map = CollectionUtil.mapStringString();
// map.put("message", msg);
// stringRedisTemplate.opsForStream().add(StreamRecords.mapBacked(map).withStreamKey(this.getRedisStreamKey()));
// }
// }
// return aBoolean;
// }
//
// protected abstract String getRedisStreamKey();
//
// protected abstract String getRedisStreamGroup();
//}
liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-adam/src/main/java/com/liquidnet/service/consumer/adam/receiver/ConsumerChimeUserOperationDisLikeRdsReceiver.java
View file @ 2d29b94d
package com.liquidnet.service.consumer.adam.receiver;
import com.liquidnet.service.base.constant.MQConst;
import org.springframework.stereotype.Component;
@Component
public class ConsumerChimeUserOperationDisLikeRdsReceiver extends AbstractChimeRedisReceiver {
@Override
protected String getRedisStreamKey() {
return MQConst.ChimeQueue.USER_OPERATION_DISLIKE.getKey();
}
@Override
protected String getRedisStreamGroup() {
return MQConst.ChimeQueue.USER_OPERATION_DISLIKE.getGroup();
}
}
//package com.liquidnet.service.consumer.adam.receiver;
//
//import com.liquidnet.service.base.constant.MQConst;
//import org.springframework.stereotype.Component;
//
//@Component
//public class ConsumerChimeUserOperationDisLikeRdsReceiver extends AbstractChimeRedisReceiver {
// @Override
// protected String getRedisStreamKey() {
// return MQConst.ChimeQueue.USER_OPERATION_DISLIKE.getKey();
// }
//
// @Override
// protected String getRedisStreamGroup() {
// return MQConst.ChimeQueue.USER_OPERATION_DISLIKE.getGroup();
// }
//}
liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-adam/src/main/java/com/liquidnet/service/consumer/adam/receiver/ConsumerChimeUserOperationLikeRdsReceiver.java
View file @ 2d29b94d
package com.liquidnet.service.consumer.adam.receiver;
import com.liquidnet.service.base.constant.MQConst;
import org.springframework.stereotype.Component;
@Component
public class ConsumerChimeUserOperationLikeRdsReceiver extends AbstractChimeRedisReceiver {
@Override
protected String getRedisStreamKey() {
return MQConst.ChimeQueue.USER_OPERATION_LIKE.getKey();
}
@Override
protected String getRedisStreamGroup() {
return MQConst.ChimeQueue.USER_OPERATION_LIKE.getGroup();
}
}
//package com.liquidnet.service.consumer.adam.receiver;
//
//import com.liquidnet.service.base.constant.MQConst;
//import org.springframework.stereotype.Component;
//
//@Component
//public class ConsumerChimeUserOperationLikeRdsReceiver extends AbstractChimeRedisReceiver {
// @Override
// protected String getRedisStreamKey() {
// return MQConst.ChimeQueue.USER_OPERATION_LIKE.getKey();
// }
//
// @Override
// protected String getRedisStreamGroup() {
// return MQConst.ChimeQueue.USER_OPERATION_LIKE.getGroup();
// }
//}
liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-adam/src/main/java/com/liquidnet/service/consumer/adam/util/ChimeDataUtils.java
View file @ 2d29b94d
package com.liquidnet.service.consumer.adam.util;
import com.liquidnet.service.chime.constant.ChimeConstant;
import com.liquidnet.service.chime.vo.mongo.ChimeUserInfoVo;
import com.liquidnet.service.chime.vo.mongo.ChimeUserOperLogVo;
import com.mongodb.client.result.UpdateResult;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.data.mongodb.core.query.Update;
import org.springframework.stereotype.Component;
@Slf4j
@Component
public class ChimeDataUtils {
@Autowired
private MongoTemplate mongoTemplate;
/**
* 添加操作日志
* @param chimeUserOperLogVo
*/
public void createUserOperLog(ChimeUserOperLogVo chimeUserOperLogVo){
mongoTemplate.save(chimeUserOperLogVo,ChimeUserOperLogVo.class.getSimpleName());
}
/**
* 修改社交用户操作计数
* @param chimeUserOperLogVo
*/
public void updateChimeUser(ChimeUserOperLogVo chimeUserOperLogVo) {
ChimeUserInfoVo chimeUserInfoVo = null;
if(chimeUserOperLogVo.getOpType().equalsIgnoreCase(ChimeConstant.LOG_USER_OPERATION_LIKE)||chimeUserOperLogVo.getOpType().equalsIgnoreCase(ChimeConstant.LOG_USER_OPERATION_DISLIKE)){
chimeUserInfoVo = this.getUserByUserId(chimeUserOperLogVo.getTargetUserId());
}else if(chimeUserOperLogVo.getOpType().equalsIgnoreCase(ChimeConstant.LOG_USER_OPERATION_LIKE_MYSELF_CLICK)||chimeUserOperLogVo.getOpType().equalsIgnoreCase(ChimeConstant.LOG_USER_OPERATION_DISLIKE_MYSELF_CLICK)){
chimeUserInfoVo = this.getUserByUserId(chimeUserOperLogVo.getCurrentUserId());
}
if(chimeUserInfoVo == null){
log.error("chimeUserInfoVo is null userId is not exist:");
return;
}
Query query = Query.query(Criteria.where("userId").is(chimeUserInfoVo.getUserId()));
Update update = new Update();
if(chimeUserOperLogVo.getOpType().equalsIgnoreCase(ChimeConstant.LOG_USER_OPERATION_LIKE)){
update.set("likeCount", chimeUserInfoVo.getLikeCount() + 1);
}else if(chimeUserOperLogVo.getOpType().equalsIgnoreCase(ChimeConstant.LOG_USER_OPERATION_DISLIKE)){
update.set("disLikeCount", chimeUserInfoVo.getDisLikeCount() + 1);
}else if(chimeUserOperLogVo.getOpType().equalsIgnoreCase(ChimeConstant.LOG_USER_OPERATION_LIKE_MYSELF_CLICK)){
update.set("clickLikeCount", chimeUserInfoVo.getClickLikeCount() + 1);
}else if(chimeUserOperLogVo.getOpType().equalsIgnoreCase(ChimeConstant.LOG_USER_OPERATION_DISLIKE_MYSELF_CLICK)){
update.set("clickDisLikeCount", chimeUserInfoVo.getClickDisLikeCount() + 1);
}
UpdateResult result = mongoTemplate.updateFirst(query,update,ChimeUserInfoVo.class,ChimeUserInfoVo.class.getSimpleName());
log.info("updateChimeUser result:{}",result.toString());
//package com.liquidnet.service.consumer.adam.util;
//
//import com.liquidnet.service.chime.constant.ChimeConstant;
//import com.liquidnet.service.chime.vo.mongo.ChimeUserInfoVo;
//import com.liquidnet.service.chime.vo.mongo.ChimeUserOperLogVo;
//import com.mongodb.client.result.UpdateResult;
//import lombok.extern.slf4j.Slf4j;
//import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.data.mongodb.core.MongoTemplate;
//import org.springframework.data.mongodb.core.query.Criteria;
//import org.springframework.data.mongodb.core.query.Query;
//import org.springframework.data.mongodb.core.query.Update;
//import org.springframework.stereotype.Component;
//
//@Slf4j
//@Component
//public class ChimeDataUtils {
//
// @Autowired
// private MongoTemplate mongoTemplate;
//
// /**
// * 添加操作日志
// * @param chimeUserOperLogVo
// */
// public void createUserOperLog(ChimeUserOperLogVo chimeUserOperLogVo){
// mongoTemplate.save(chimeUserOperLogVo,ChimeUserOperLogVo.class.getSimpleName());
// }
//
// /**
// * 修改社交用户操作计数
// * @param chimeUserOperLogVo
// */
// public void updateChimeUser(ChimeUserOperLogVo chimeUserOperLogVo) {
// ChimeUserInfoVo chimeUserInfoVo = null;
// if(chimeUserOperLogVo.getOpType().equalsIgnoreCase(ChimeConstant.LOG_USER_OPERATION_LIKE)||chimeUserOperLogVo.getOpType().equalsIgnoreCase(ChimeConstant.LOG_USER_OPERATION_DISLIKE)){
// chimeUserInfoVo = this.getUserByUserId(chimeUserOperLogVo.getTargetUserId());
// }else if(chimeUserOperLogVo.getOpType().equalsIgnoreCase(ChimeConstant.LOG_USER_OPERATION_LIKE_MYSELF_CLICK)||chimeUserOperLogVo.getOpType().equalsIgnoreCase(ChimeConstant.LOG_USER_OPERATION_DISLIKE_MYSELF_CLICK)){
// chimeUserInfoVo = this.getUserByUserId(chimeUserOperLogVo.getCurrentUserId());
// }
// if(chimeUserInfoVo == null){
// log.error("chimeUserInfoVo is null userId is not exist:");
// return;
// }
// Query query = Query.query(Criteria.where("userId").is(chimeUserInfoVo.getUserId()));
//// Update update = Update.fromDocument(Document.parse(JsonUtils.toJson(chimeUserInfoVo)));
// Update update = Update.fromDocument(Document.parse(JsonUtils.toJson(chimeUserInfoVo)));
// Update update = new Update();
// if(chimeUserOperLogVo.getOpType().equalsIgnoreCase(ChimeConstant.LOG_USER_OPERATION_LIKE)){
// update.set("likeCount", chimeUserInfoVo.getLikeCount() + 1);
// }else if(chimeUserOperLogVo.getOpType().equalsIgnoreCase(ChimeConstant.LOG_USER_OPERATION_DISLIKE)){
// update.set("disLikeCount", chimeUserInfoVo.getDisLikeCount() + 1);
// }else if(chimeUserOperLogVo.getOpType().equalsIgnoreCase(ChimeConstant.LOG_USER_OPERATION_LIKE_MYSELF_CLICK)){
// update.set("clickLikeCount", chimeUserInfoVo.getClickLikeCount() + 1);
// }else if(chimeUserOperLogVo.getOpType().equalsIgnoreCase(ChimeConstant.LOG_USER_OPERATION_DISLIKE_MYSELF_CLICK)){
// update.set("clickDisLikeCount", chimeUserInfoVo.getClickDisLikeCount() + 1);
// }
// update.set("likeCount", 8);
// update.set("disLikeCount", 9);
//
// UpdateResult result = mongoTemplate.updateFirst(query,update,ChimeUserInfoVo.class,ChimeUserInfoVo.class.getSimpleName());
// log.info("updateChimeUser result:{}",result.toString());
}
/**
* 获取单个用户信息
* @param userId
* @return
*/
public ChimeUserInfoVo getUserByUserId(String userId){
Query query = Query.query(Criteria.where("userId").is(userId));
ChimeUserInfoVo chimeUserInfoVo = mongoTemplate.findOne(query,ChimeUserInfoVo.class,ChimeUserInfoVo.class.getSimpleName());
return chimeUserInfoVo;
}
}
//
//// Query query = Query.query(Criteria.where("userId").is(chimeUserInfoVo.getUserId()));
////// Update update = Update.fromDocument(Document.parse(JsonUtils.toJson(chimeUserInfoVo)));
//// Update update = Update.fromDocument(Document.parse(JsonUtils.toJson(chimeUserInfoVo)));
//// if(chimeUserOperLogVo.getOpType().equalsIgnoreCase(ChimeConstant.LOG_USER_OPERATION_LIKE)){
//// update.set("likeCount", chimeUserInfoVo.getLikeCount() + 1);
//// }else if(chimeUserOperLogVo.getOpType().equalsIgnoreCase(ChimeConstant.LOG_USER_OPERATION_DISLIKE)){
//// update.set("disLikeCount", chimeUserInfoVo.getDisLikeCount() + 1);
//// }
//// update.set("likeCount", 8);
//// update.set("disLikeCount", 9);
//// UpdateResult result = mongoTemplate.updateFirst(query,update,ChimeUserInfoVo.class,ChimeUserInfoVo.class.getSimpleName());
//// log.info("updateChimeUser result:{}",result.toString());
// }
//
// /**
// * 获取单个用户信息
// * @param userId
// * @return
// */
// public ChimeUserInfoVo getUserByUserId(String userId){
// Query query = Query.query(Criteria.where("userId").is(userId));
// ChimeUserInfoVo chimeUserInfoVo = mongoTemplate.findOne(query,ChimeUserInfoVo.class,ChimeUserInfoVo.class.getSimpleName());
// return chimeUserInfoVo;
// }
//}
liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-base/pom.xml
View file @ 2d29b94d
......@@ -49,17 +49,17 @@
</dependency>
<dependency>
<groupId>com.liquidnet</groupId>
<artifactId>liquidnet-service-sweet-do</artifactId>
<artifactId>liquidnet-api-feign-stone</artifactId>
<version>1.0-SNAPSHOT</version>
</dependency>
<dependency>
<groupId>com.liquidnet</groupId>
<artifactId>liquidnet-api-feign-stone</artifactId>
<artifactId>liquidnet-service-sweet-api</artifactId>
<version>1.0-SNAPSHOT</version>
</dependency>
<dependency>
<groupId>com.liquidnet</groupId>
<artifactId>liquidnet-service-sweet-api</artifactId>
<artifactId>liquidnet-service-chime-api</artifactId>
<version>1.0-SNAPSHOT</version>
</dependency>
......
liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-base/src/main/java/com/liquidnet/service/consumer/base/config/ConsumerCommonBizRedisStreamConfig.java
View file @ 2d29b94d
......@@ -47,6 +47,10 @@ public class ConsumerCommonBizRedisStreamConfig extends RedisStreamConfig {
ConsumerSweetTemplateMsgReceiver consumerSweetTemplateMsgReceiver;
@Autowired
ConsumerSweetStoneIntegralReceiver consumerSweetStoneIntegralReceiver;
@Autowired
ConsumerChimeUserOperationLikeReceiver consumerChimeUserOperationLikeReceiver;
@Autowired
ConsumerChimeUserOperationDislikeReceiver consumerChimeUserOperationDislikeReceiver;
/*------sweet------*/
@Bean // 发送模版消息
......@@ -64,7 +68,7 @@ public class ConsumerCommonBizRedisStreamConfig extends RedisStreamConfig {
}
return subscriptionList;
}
@Bean // stone积分处理
@Bean
public List<Subscription> subscriptionSweetSqlApiStoneIntgral(RedisConnectionFactory factory) {
List<Subscription> subscriptionList = new ArrayList<>();
MQConst.SweetQueue stream = MQConst.SweetQueue.SWEET_STONE_INTEGRAL;
......@@ -80,7 +84,7 @@ public class ConsumerCommonBizRedisStreamConfig extends RedisStreamConfig {
return subscriptionList;
}
@Bean// 短信通知
@Bean
public List<Subscription> subscriptionCommonSmsNotice(RedisConnectionFactory factory) {
List<Subscription> subscriptionList = new ArrayList<>();
MQConst.KylinQueue stream = MQConst.KylinQueue.SMS_NOTICE;
......@@ -96,7 +100,7 @@ public class ConsumerCommonBizRedisStreamConfig extends RedisStreamConfig {
return subscriptionList;
}
@Bean// 藏品上传声明
@Bean
public List<Subscription> subscriptionGoblinBizArtworkUpl(RedisConnectionFactory factory) {
List<Subscription> subscriptionList = new ArrayList<>();
MQConst.GoblinQueue stream = MQConst.GoblinQueue.BIZ_ARTWORK_UPL;
......@@ -112,7 +116,7 @@ public class ConsumerCommonBizRedisStreamConfig extends RedisStreamConfig {
return subscriptionList;
}
@Bean// 藏品声明查询
@Bean
public List<Subscription> subscriptionGoblinBizArtworkClq(RedisConnectionFactory factory) {
List<Subscription> subscriptionList = new ArrayList<>();
MQConst.GoblinQueue stream = MQConst.GoblinQueue.BIZ_ARTWORK_CLQ;
......@@ -172,7 +176,7 @@ public class ConsumerCommonBizRedisStreamConfig extends RedisStreamConfig {
return subscriptionList;
}
@Bean// 增减积分
@Bean
public List<Subscription> subscriptionGoblinBizIntegral(RedisConnectionFactory factory) {
List<Subscription> subscriptionList = new ArrayList<>();
MQConst.GoblinQueue stream = MQConst.GoblinQueue.BIZ_INTEGRAL;
......@@ -187,4 +191,36 @@ public class ConsumerCommonBizRedisStreamConfig extends RedisStreamConfig {
}
return subscriptionList;
}
@Bean
public List<Subscription> subscriptionChimeUserOperationLike(RedisConnectionFactory factory) {
List<Subscription> subscriptionList = new ArrayList<>();
MQConst.ChimeQueue stream = MQConst.ChimeQueue.USER_OPERATION_LIKE;
this.initStream(stringRedisTemplate, stream.getKey(), stream.getGroup());
for (int i = 0; i < 10; i++) {
StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer = this.buildStreamMessageListenerContainer(factory);
subscriptionList.add(listenerContainer.receiveAutoAck(
Consumer.from(stream.getGroup(), getConsumerName(stream.name() + i)),
StreamOffset.create(stream.getKey(), ReadOffset.lastConsumed()), consumerChimeUserOperationLikeReceiver
));
listenerContainer.start();
}
return subscriptionList;
}
@Bean
public List<Subscription> subscriptionChimeUserOperationDislike(RedisConnectionFactory factory) {
List<Subscription> subscriptionList = new ArrayList<>();
MQConst.ChimeQueue stream = MQConst.ChimeQueue.USER_OPERATION_DISLIKE;
this.initStream(stringRedisTemplate, stream.getKey(), stream.getGroup());
for (int i = 0; i < 10; i++) {
StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer = this.buildStreamMessageListenerContainer(factory);
subscriptionList.add(listenerContainer.receiveAutoAck(
Consumer.from(stream.getGroup(), getConsumerName(stream.name() + i)),
StreamOffset.create(stream.getKey(), ReadOffset.lastConsumed()), consumerChimeUserOperationDislikeReceiver
));
listenerContainer.start();
}
return subscriptionList;
}
}
liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-base/src/main/java/com/liquidnet/service/consumer/base/receiver/ConsumerChimeUserOperationDislikeReceiver.java 0 → 100644
View file @ 2d29b94d
package com.liquidnet.service.consumer.base.receiver;
import com.liquidnet.commons.lang.util.CollectionUtil;
import com.liquidnet.commons.lang.util.JsonUtils;
import com.liquidnet.service.base.constant.MQConst;
import com.liquidnet.service.chime.constant.ChimeConstant;
import com.liquidnet.service.chime.vo.mongo.ChimeUserInfoVo;
import com.liquidnet.service.chime.vo.mongo.ChimeUserOperLogVo;
import com.liquidnet.service.consumer.base.util.ChimeDataUtils;
import com.mongodb.client.result.UpdateResult;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.data.mongodb.core.query.Update;
import org.springframework.data.redis.connection.stream.StreamRecords;
import org.springframework.stereotype.Component;
import java.util.HashMap;
@Slf4j
@Component
public class ConsumerChimeUserOperationDislikeReceiver extends AbstractBizRedisReceiver {
@Autowired
private ChimeDataUtils chimeDataUtils;
@Override
protected boolean consumerMessageHandler(String msg) {
boolean aBoolean = false;
try {
ChimeUserOperLogVo textMessage = JsonUtils.fromJson(msg, ChimeUserOperLogVo.class);
if (textMessage == null) {
aBoolean = true;
} else {
//执行计数
chimeDataUtils.updateChimeUser(textMessage);
//创建操作日志
chimeDataUtils.createUserOperLog(textMessage);
aBoolean = true;
}
} catch (Exception e) {
log.error("CONSUMER MSG EX_HANDLE ==> [{}]:{}", this.getRedisStreamKey(), msg, e);
} finally {
if (!aBoolean) {
HashMap<String, String> map = CollectionUtil.mapStringString();
map.put("message", msg);
stringRedisTemplate.opsForStream().add(StreamRecords.mapBacked(map).withStreamKey(this.getRedisStreamKey()));
}
}
return aBoolean;
}
@Override
protected String getRedisStreamKey() {
return MQConst.ChimeQueue.USER_OPERATION_DISLIKE.getKey();
}
@Override
protected String getRedisStreamGroup() {
return MQConst.ChimeQueue.USER_OPERATION_DISLIKE.getGroup();
}
/* -------------------------------------------------------------------------- */
/**
* 添加操作日志
* @param chimeUserOperLogVo
*/
private void createUserOperLog(ChimeUserOperLogVo chimeUserOperLogVo){
mongoTemplate.save(chimeUserOperLogVo,ChimeUserOperLogVo.class.getSimpleName());
}
/**
* 修改社交用户操作计数
* @param chimeUserOperLogVo
*/
private void updateChimeUser(ChimeUserOperLogVo chimeUserOperLogVo) {
ChimeUserInfoVo chimeUserInfoVo = null;
if(chimeUserOperLogVo.getOpType().equalsIgnoreCase(ChimeConstant.LOG_USER_OPERATION_LIKE)||chimeUserOperLogVo.getOpType().equalsIgnoreCase(ChimeConstant.LOG_USER_OPERATION_DISLIKE)){
chimeUserInfoVo = this.getUserByUserId(chimeUserOperLogVo.getTargetUserId());
}else if(chimeUserOperLogVo.getOpType().equalsIgnoreCase(ChimeConstant.LOG_USER_OPERATION_LIKE_MYSELF_CLICK)||chimeUserOperLogVo.getOpType().equalsIgnoreCase(ChimeConstant.LOG_USER_OPERATION_DISLIKE_MYSELF_CLICK)){
chimeUserInfoVo = this.getUserByUserId(chimeUserOperLogVo.getCurrentUserId());
}
if(chimeUserInfoVo == null){
log.error("chimeUserInfoVo is null userId is not exist:");
return;
}
Query query = Query.query(Criteria.where("userId").is(chimeUserInfoVo.getUserId()));
Update update = new Update();
if(chimeUserOperLogVo.getOpType().equalsIgnoreCase(ChimeConstant.LOG_USER_OPERATION_LIKE)){
update.set("likeCount", chimeUserInfoVo.getLikeCount() + 1);
}else if(chimeUserOperLogVo.getOpType().equalsIgnoreCase(ChimeConstant.LOG_USER_OPERATION_DISLIKE)){
update.set("disLikeCount", chimeUserInfoVo.getDisLikeCount() + 1);
}else if(chimeUserOperLogVo.getOpType().equalsIgnoreCase(ChimeConstant.LOG_USER_OPERATION_LIKE_MYSELF_CLICK)){
update.set("clickLikeCount", chimeUserInfoVo.getClickLikeCount() + 1);
}else if(chimeUserOperLogVo.getOpType().equalsIgnoreCase(ChimeConstant.LOG_USER_OPERATION_DISLIKE_MYSELF_CLICK)){
update.set("clickDisLikeCount", chimeUserInfoVo.getClickDisLikeCount() + 1);
}
UpdateResult result = mongoTemplate.updateFirst(query,update,ChimeUserInfoVo.class,ChimeUserInfoVo.class.getSimpleName());
log.info("updateChimeUser result:{}",result.toString());
// Query query = Query.query(Criteria.where("userId").is(chimeUserInfoVo.getUserId()));
//// Update update = Update.fromDocument(Document.parse(JsonUtils.toJson(chimeUserInfoVo)));
// Update update = Update.fromDocument(Document.parse(JsonUtils.toJson(chimeUserInfoVo)));
// if(chimeUserOperLogVo.getOpType().equalsIgnoreCase(ChimeConstant.LOG_USER_OPERATION_LIKE)){
// update.set("likeCount", chimeUserInfoVo.getLikeCount() + 1);
// }else if(chimeUserOperLogVo.getOpType().equalsIgnoreCase(ChimeConstant.LOG_USER_OPERATION_DISLIKE)){
// update.set("disLikeCount", chimeUserInfoVo.getDisLikeCount() + 1);
// }
// update.set("likeCount", 8);
// update.set("disLikeCount", 9);
// UpdateResult result = mongoTemplate.updateFirst(query,update,ChimeUserInfoVo.class,ChimeUserInfoVo.class.getSimpleName());
// log.info("updateChimeUser result:{}",result.toString());
}
/**
* 获取单个用户信息
* @param userId
* @return
*/
private ChimeUserInfoVo getUserByUserId(String userId){
Query query = Query.query(Criteria.where("userId").is(userId));
ChimeUserInfoVo chimeUserInfoVo = mongoTemplate.findOne(query,ChimeUserInfoVo.class,ChimeUserInfoVo.class.getSimpleName());
return chimeUserInfoVo;
}
}
liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-base/src/main/java/com/liquidnet/service/consumer/base/receiver/ConsumerChimeUserOperationLikeReceiver.java 0 → 100644
View file @ 2d29b94d
package com.liquidnet.service.consumer.base.receiver;
import com.liquidnet.commons.lang.util.CollectionUtil;
import com.liquidnet.commons.lang.util.JsonUtils;
import com.liquidnet.service.base.constant.MQConst;
import com.liquidnet.service.chime.constant.ChimeConstant;
import com.liquidnet.service.chime.vo.mongo.ChimeUserInfoVo;
import com.liquidnet.service.chime.vo.mongo.ChimeUserOperLogVo;
import com.liquidnet.service.consumer.base.util.ChimeDataUtils;
import com.mongodb.client.result.UpdateResult;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.data.mongodb.core.query.Update;
import org.springframework.data.redis.connection.stream.StreamRecords;
import org.springframework.stereotype.Component;
import java.util.HashMap;
@Slf4j
@Component
public class ConsumerChimeUserOperationLikeReceiver extends AbstractBizRedisReceiver {
@Autowired
private ChimeDataUtils chimeDataUtils;
@Override
protected boolean consumerMessageHandler(String msg) {
boolean aBoolean = false;
try {
ChimeUserOperLogVo textMessage = JsonUtils.fromJson(msg, ChimeUserOperLogVo.class);
if (textMessage == null) {
aBoolean = true;
} else {
//执行计数
chimeDataUtils.updateChimeUser(textMessage);
//创建操作日志
chimeDataUtils.createUserOperLog(textMessage);
aBoolean = true;
}
} catch (Exception e) {
log.error("CONSUMER MSG EX_HANDLE ==> [{}]:{}", this.getRedisStreamKey(), msg, e);
} finally {
if (!aBoolean) {
HashMap<String, String> map = CollectionUtil.mapStringString();
map.put("message", msg);
stringRedisTemplate.opsForStream().add(StreamRecords.mapBacked(map).withStreamKey(this.getRedisStreamKey()));
}
}
return aBoolean;
}
@Override
protected String getRedisStreamKey() {
return MQConst.ChimeQueue.USER_OPERATION_LIKE.getKey();
}
@Override
protected String getRedisStreamGroup() {
return MQConst.ChimeQueue.USER_OPERATION_LIKE.getGroup();
}
}
liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-base/src/main/java/com/liquidnet/service/consumer/base/util/ChimeDataUtils.java 0 → 100644
View file @ 2d29b94d
package com.liquidnet.service.consumer.base.util;
import com.liquidnet.service.chime.constant.ChimeConstant;
import com.liquidnet.service.chime.vo.mongo.ChimeUserInfoVo;
import com.liquidnet.service.chime.vo.mongo.ChimeUserOperLogVo;
import com.mongodb.client.result.UpdateResult;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.data.mongodb.core.query.Update;
import org.springframework.stereotype.Component;
@Slf4j
@Component
public class ChimeDataUtils {
@Autowired
private MongoTemplate mongoTemplate;
/**
* 添加操作日志
* @param chimeUserOperLogVo
*/
public void createUserOperLog(ChimeUserOperLogVo chimeUserOperLogVo){
mongoTemplate.save(chimeUserOperLogVo,ChimeUserOperLogVo.class.getSimpleName());
}
/**
* 修改社交用户操作计数
* @param chimeUserOperLogVo
*/
public void updateChimeUser(ChimeUserOperLogVo chimeUserOperLogVo) {
ChimeUserInfoVo chimeUserInfoVo = null;
if(chimeUserOperLogVo.getOpType().equalsIgnoreCase(ChimeConstant.LOG_USER_OPERATION_LIKE)||chimeUserOperLogVo.getOpType().equalsIgnoreCase(ChimeConstant.LOG_USER_OPERATION_DISLIKE)){
chimeUserInfoVo = this.getUserByUserId(chimeUserOperLogVo.getTargetUserId());
}else if(chimeUserOperLogVo.getOpType().equalsIgnoreCase(ChimeConstant.LOG_USER_OPERATION_LIKE_MYSELF_CLICK)||chimeUserOperLogVo.getOpType().equalsIgnoreCase(ChimeConstant.LOG_USER_OPERATION_DISLIKE_MYSELF_CLICK)){
chimeUserInfoVo = this.getUserByUserId(chimeUserOperLogVo.getCurrentUserId());
}
if(chimeUserInfoVo == null){
log.error("chimeUserInfoVo is null userId is not exist:");
return;
}
Query query = Query.query(Criteria.where("userId").is(chimeUserInfoVo.getUserId()));
Update update = new Update();
if(chimeUserOperLogVo.getOpType().equalsIgnoreCase(ChimeConstant.LOG_USER_OPERATION_LIKE)){
update.set("likeCount", chimeUserInfoVo.getLikeCount() + 1);
}else if(chimeUserOperLogVo.getOpType().equalsIgnoreCase(ChimeConstant.LOG_USER_OPERATION_DISLIKE)){
update.set("disLikeCount", chimeUserInfoVo.getDisLikeCount() + 1);
}else if(chimeUserOperLogVo.getOpType().equalsIgnoreCase(ChimeConstant.LOG_USER_OPERATION_LIKE_MYSELF_CLICK)){
update.set("clickLikeCount", chimeUserInfoVo.getClickLikeCount() + 1);
}else if(chimeUserOperLogVo.getOpType().equalsIgnoreCase(ChimeConstant.LOG_USER_OPERATION_DISLIKE_MYSELF_CLICK)){
update.set("clickDisLikeCount", chimeUserInfoVo.getClickDisLikeCount() + 1);
}
UpdateResult result = mongoTemplate.updateFirst(query,update,ChimeUserInfoVo.class,ChimeUserInfoVo.class.getSimpleName());
log.info("updateChimeUser result:{}",result.toString());
// Query query = Query.query(Criteria.where("userId").is(chimeUserInfoVo.getUserId()));
//// Update update = Update.fromDocument(Document.parse(JsonUtils.toJson(chimeUserInfoVo)));
// Update update = Update.fromDocument(Document.parse(JsonUtils.toJson(chimeUserInfoVo)));
// if(chimeUserOperLogVo.getOpType().equalsIgnoreCase(ChimeConstant.LOG_USER_OPERATION_LIKE)){
// update.set("likeCount", chimeUserInfoVo.getLikeCount() + 1);
// }else if(chimeUserOperLogVo.getOpType().equalsIgnoreCase(ChimeConstant.LOG_USER_OPERATION_DISLIKE)){
// update.set("disLikeCount", chimeUserInfoVo.getDisLikeCount() + 1);
// }
// update.set("likeCount", 8);
// update.set("disLikeCount", 9);
// UpdateResult result = mongoTemplate.updateFirst(query,update,ChimeUserInfoVo.class,ChimeUserInfoVo.class.getSimpleName());
// log.info("updateChimeUser result:{}",result.toString());
}
/**
* 获取单个用户信息
* @param userId
* @return
*/
public ChimeUserInfoVo getUserByUserId(String userId){
Query query = Query.query(Criteria.where("userId").is(userId));
ChimeUserInfoVo chimeUserInfoVo = mongoTemplate.findOne(query,ChimeUserInfoVo.class,ChimeUserInfoVo.class.getSimpleName());
return chimeUserInfoVo;
}
}
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment