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

Skip to content

L
liquidnet-bus-v1
Commit c153d97f authored by 张国柄's avatar 张国柄
Browse files
Options

~opt;

parent 3b6c7030
Expand all Show whitespace changes
Inline Side-by-side
Showing with 353 additions and 353 deletions
liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-nft/src/main/java/com/liquidnet/service/consumer/nft/config/ConsumerCommonBizRedisStreamConfig.java
View file @ c153d97f
package com.liquidnet.service.consumer.nft.config;
import com.liquidnet.common.cache.redis.config.RedisStreamConfig;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.core.StringRedisTemplate;
@Configuration
public class ConsumerCommonBizRedisStreamConfig extends RedisStreamConfig {
@Autowired
StringRedisTemplate stringRedisTemplate;
}
//package com.liquidnet.service.consumer.nft.config;
//
//import com.liquidnet.common.cache.redis.config.RedisStreamConfig;
//import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.context.annotation.Configuration;
//import org.springframework.data.redis.core.StringRedisTemplate;
//
//@Configuration
//public class ConsumerCommonBizRedisStreamConfig extends RedisStreamConfig {
// @Autowired
// StringRedisTemplate stringRedisTemplate;
//}
liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-nft/src/main/java/com/liquidnet/service/consumer/nft/receiver/AbstractBizRedisReceiver.java
View file @ c153d97f
package com.liquidnet.service.consumer.nft.receiver;
import com.liquidnet.service.base.constant.MQConst;
import com.liquidnet.service.consumer.nft.service.IBaseDao;
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.core.StringRedisTemplate;
import org.springframework.data.redis.stream.StreamListener;
/**
* 公共的业务队列消息监听器，具体业务消费逻辑通过`consumerMessageHandler`实现
*
* @author zhanggb
* Created by IntelliJ IDEA at 2022/3/31
*/
@Slf4j
public abstract class AbstractBizRedisReceiver implements StreamListener<String, MapRecord<String, String, String>> {
@Autowired
public IBaseDao baseDao;
@Autowired
public StringRedisTemplate stringRedisTemplate;
@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(MQConst.QUEUE_MESSAGE_KEY));
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);
}
}
protected abstract boolean consumerMessageHandler(String msg);
protected abstract String getRedisStreamKey();
protected abstract String getRedisStreamGroup();
}
\ No newline at end of file
//package com.liquidnet.service.consumer.nft.receiver;
//
//import com.liquidnet.service.base.constant.MQConst;
//import com.liquidnet.service.consumer.nft.service.IBaseDao;
//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.core.StringRedisTemplate;
//import org.springframework.data.redis.stream.StreamListener;
//
///**
// * 公共的业务队列消息监听器，具体业务消费逻辑通过`consumerMessageHandler`实现
// *
// * @author zhanggb
// * Created by IntelliJ IDEA at 2022/3/31
// */
//@Slf4j
//public abstract class AbstractBizRedisReceiver implements StreamListener<String, MapRecord<String, String, String>> {
// @Autowired
// public IBaseDao baseDao;
// @Autowired
// public StringRedisTemplate stringRedisTemplate;
//
// @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(MQConst.QUEUE_MESSAGE_KEY));
// 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);
// }
// }
//
// protected abstract boolean consumerMessageHandler(String msg);
//
// protected abstract String getRedisStreamKey();
//
// protected abstract String getRedisStreamGroup();
//}
\ No newline at end of file
liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-nft/src/main/java/com/liquidnet/service/consumer/nft/receiver/AbstractSqlRedisReceiver.java
View file @ c153d97f
package com.liquidnet.service.consumer.nft.receiver;
import com.liquidnet.commons.lang.util.CollectionUtil;
import com.liquidnet.commons.lang.util.JsonUtils;
import com.liquidnet.service.base.SqlMapping;
import com.liquidnet.service.consumer.nft.service.IBaseDao;
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;
/**
* 公共的SQL队列消息监听器，具体SQL消费逻辑统一使用`consumerMessageHandler`
*
* @author zhanggb
* Created by IntelliJ IDEA at 2022/3/31
*/
@Slf4j
public abstract class AbstractSqlRedisReceiver implements StreamListener<String, MapRecord<String, String, String>> {
@Autowired
private IBaseDao baseDao;
@Autowired
StringRedisTemplate stringRedisTemplate;
@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 {
SqlMapping.SqlMessage sqlMessage = JsonUtils.fromJson(msg, SqlMapping.SqlMessage.class);
if (sqlMessage == null) {
aBoolean = true;
} else {
aBoolean = baseDao.batchSqls(sqlMessage.getSqls(), sqlMessage.getArgs());
}
} 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();
}
\ No newline at end of file
//package com.liquidnet.service.consumer.nft.receiver;
//
//import com.liquidnet.commons.lang.util.CollectionUtil;
//import com.liquidnet.commons.lang.util.JsonUtils;
//import com.liquidnet.service.base.SqlMapping;
//import com.liquidnet.service.consumer.nft.service.IBaseDao;
//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;
//
///**
// * 公共的SQL队列消息监听器，具体SQL消费逻辑统一使用`consumerMessageHandler`
// *
// * @author zhanggb
// * Created by IntelliJ IDEA at 2022/3/31
// */
//@Slf4j
//public abstract class AbstractSqlRedisReceiver implements StreamListener<String, MapRecord<String, String, String>> {
// @Autowired
// private IBaseDao baseDao;
// @Autowired
// StringRedisTemplate stringRedisTemplate;
//
// @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 {
// SqlMapping.SqlMessage sqlMessage = JsonUtils.fromJson(msg, SqlMapping.SqlMessage.class);
// if (sqlMessage == null) {
// aBoolean = true;
// } else {
// aBoolean = baseDao.batchSqls(sqlMessage.getSqls(), sqlMessage.getArgs());
// }
// } 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();
//}
\ No newline at end of file
liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-nft/src/main/java/com/liquidnet/service/consumer/nft/receiver/ConsumerGalaxyJsonNftPublishAndBuyReceiver.java
View file @ c153d97f
package com.liquidnet.service.consumer.nft.receiver;
import com.liquidnet.commons.lang.util.CollectionUtil;
import com.liquidnet.commons.lang.util.JsonUtils;
import com.liquidnet.service.base.ResponseDto;
import com.liquidnet.service.base.constant.MQConst;
import com.liquidnet.service.consumer.nft.service.processor.ConsumerGalaxyJsonNftPublishAndBuyProcessor;
import com.liquidnet.service.galaxy.dto.param.GalaxyNftPublishAndBuyReqDto;
import com.liquidnet.service.galaxy.dto.param.GalaxyNftPublishAndBuyRespDto;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.stream.StreamRecords;
import org.springframework.stereotype.Component;
import java.util.HashMap;
@Slf4j
@Component
public class ConsumerGalaxyJsonNftPublishAndBuyReceiver extends AbstractBizRedisReceiver {
@Autowired
private ConsumerGalaxyJsonNftPublishAndBuyProcessor jsonNftPublishAndBuyProcessor;
@Override
protected boolean consumerMessageHandler(String msg) {
boolean aBoolean = false;
try {
GalaxyNftPublishAndBuyReqDto textMessage = JsonUtils.fromJson(msg, GalaxyNftPublishAndBuyReqDto.class);
if (textMessage == null) {
aBoolean = true;
} else {
//执行计数
ResponseDto<GalaxyNftPublishAndBuyRespDto> responseDto = jsonNftPublishAndBuyProcessor.nftPublishAndBuy(textMessage);
if(responseDto.isSuccess()){
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.GalaxyQueue.JSON_NFT_PUBLISH_AND_BUY.getKey();
}
@Override
protected String getRedisStreamGroup() {
return MQConst.GalaxyQueue.JSON_NFT_PUBLISH_AND_BUY.getGroup();
}
}
//package com.liquidnet.service.consumer.nft.receiver;
//
//import com.liquidnet.commons.lang.util.CollectionUtil;
//import com.liquidnet.commons.lang.util.JsonUtils;
//import com.liquidnet.service.base.ResponseDto;
//import com.liquidnet.service.base.constant.MQConst;
//import com.liquidnet.service.consumer.nft.service.processor.ConsumerGalaxyJsonNftPublishAndBuyProcessor;
//import com.liquidnet.service.galaxy.dto.param.GalaxyNftPublishAndBuyReqDto;
//import com.liquidnet.service.galaxy.dto.param.GalaxyNftPublishAndBuyRespDto;
//import lombok.extern.slf4j.Slf4j;
//import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.data.redis.connection.stream.StreamRecords;
//import org.springframework.stereotype.Component;
//
//import java.util.HashMap;
//
//@Slf4j
//@Component
//public class ConsumerGalaxyJsonNftPublishAndBuyReceiver extends AbstractBizRedisReceiver {
// @Autowired
// private ConsumerGalaxyJsonNftPublishAndBuyProcessor jsonNftPublishAndBuyProcessor;
//
// @Override
// protected boolean consumerMessageHandler(String msg) {
// boolean aBoolean = false;
// try {
// GalaxyNftPublishAndBuyReqDto textMessage = JsonUtils.fromJson(msg, GalaxyNftPublishAndBuyReqDto.class);
// if (textMessage == null) {
// aBoolean = true;
// } else {
// //执行计数
// ResponseDto<GalaxyNftPublishAndBuyRespDto> responseDto = jsonNftPublishAndBuyProcessor.nftPublishAndBuy(textMessage);
// if(responseDto.isSuccess()){
// 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.GalaxyQueue.JSON_NFT_PUBLISH_AND_BUY.getKey();
// }
//
// @Override
// protected String getRedisStreamGroup() {
// return MQConst.GalaxyQueue.JSON_NFT_PUBLISH_AND_BUY.getGroup();
// }
//}
liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-nft/src/main/java/com/liquidnet/service/consumer/nft/receiver/ConsumerGalaxyJsonNftUserRegisterReceiver.java
View file @ c153d97f
package com.liquidnet.service.consumer.nft.receiver;
import com.liquidnet.commons.lang.util.CollectionUtil;
import com.liquidnet.commons.lang.util.JsonUtils;
import com.liquidnet.service.base.ResponseDto;
import com.liquidnet.service.base.constant.MQConst;
import com.liquidnet.service.consumer.nft.service.processor.ConsumerGalaxyJsonNftUserRegisterProcessor;
import com.liquidnet.service.galaxy.dto.param.GalaxyUserRegisterReqDto;
import com.liquidnet.service.galaxy.dto.param.GalaxyUserRegisterRespDto;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.stream.StreamRecords;
import org.springframework.stereotype.Component;
import java.util.HashMap;
@Slf4j
@Component
public class ConsumerGalaxyJsonNftUserRegisterReceiver extends AbstractBizRedisReceiver {
@Autowired
private ConsumerGalaxyJsonNftUserRegisterProcessor jsonNftUserRegisterProcessor;
@Override
protected boolean consumerMessageHandler(String msg) {
boolean aBoolean = false;
try {
GalaxyUserRegisterReqDto textMessage = JsonUtils.fromJson(msg, GalaxyUserRegisterReqDto.class);
if (textMessage == null) {
aBoolean = true;
} else {
//执行计数
ResponseDto<GalaxyUserRegisterRespDto> responseDto = jsonNftUserRegisterProcessor.userRegister(textMessage);
// if(responseDto.isSuccess()){
aBoolean = true;
//package com.liquidnet.service.consumer.nft.receiver;
//
//import com.liquidnet.commons.lang.util.CollectionUtil;
//import com.liquidnet.commons.lang.util.JsonUtils;
//import com.liquidnet.service.base.ResponseDto;
//import com.liquidnet.service.base.constant.MQConst;
//import com.liquidnet.service.consumer.nft.service.processor.ConsumerGalaxyJsonNftUserRegisterProcessor;
//import com.liquidnet.service.galaxy.dto.param.GalaxyUserRegisterReqDto;
//import com.liquidnet.service.galaxy.dto.param.GalaxyUserRegisterRespDto;
//import lombok.extern.slf4j.Slf4j;
//import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.data.redis.connection.stream.StreamRecords;
//import org.springframework.stereotype.Component;
//
//import java.util.HashMap;
//
//@Slf4j
//@Component
//public class ConsumerGalaxyJsonNftUserRegisterReceiver extends AbstractBizRedisReceiver {
// @Autowired
// private ConsumerGalaxyJsonNftUserRegisterProcessor jsonNftUserRegisterProcessor;
//
// @Override
// protected boolean consumerMessageHandler(String msg) {
// boolean aBoolean = false;
// try {
// GalaxyUserRegisterReqDto textMessage = JsonUtils.fromJson(msg, GalaxyUserRegisterReqDto.class);
// if (textMessage == null) {
// aBoolean = true;
// } else {
// //执行计数
// ResponseDto<GalaxyUserRegisterRespDto> responseDto = jsonNftUserRegisterProcessor.userRegister(textMessage);
//// if(responseDto.isSuccess()){
// 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.GalaxyQueue.JSON_NFT_USER_REGISTER.getKey();
}
@Override
protected String getRedisStreamGroup() {
return MQConst.GalaxyQueue.JSON_NFT_USER_REGISTER.getGroup();
}
}
// } 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.GalaxyQueue.JSON_NFT_USER_REGISTER.getKey();
// }
//
// @Override
// protected String getRedisStreamGroup() {
// return MQConst.GalaxyQueue.JSON_NFT_USER_REGISTER.getGroup();
// }
//}
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