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

Skip to content

L
liquidnet-bus-v1
Commit 98c887bf authored by 胡佳晨's avatar 胡佳晨
Browse files
Options

消费迁移 xls相关 转赠订单 定时过期 提交

parent d1e3d6bf
Hide whitespace changes
Inline Side-by-side
Showing with 992 additions and 671 deletions
liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-base/src/main/java/com/liquidnet/service/consumer/base/config/ConsumerGoblinXlsRedisStreamConfig.java 0 → 100644
View file @ 98c887bf
package com.liquidnet.service.consumer.base.config;
import com.liquidnet.common.cache.redis.config.RedisStreamConfig;
import com.liquidnet.service.base.constant.MQConst;
import com.liquidnet.service.consumer.base.receiver.ConsumerGoblinXlsRdsReceiver;
import lombok.var;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.stream.Consumer;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.connection.stream.StreamOffset;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;
import org.springframework.data.redis.stream.Subscription;
import java.util.ArrayList;
import java.util.List;
@Configuration
public class ConsumerGoblinXlsRedisStreamConfig extends RedisStreamConfig {
@Autowired
StringRedisTemplate stringRedisTemplate;
@Autowired
ConsumerGoblinXlsRdsReceiver consumerGoblinXlsRdsReceiver;
// xls黑白名单
@Bean
public List<Subscription> subscriptionGoblinXls(RedisConnectionFactory factory) {
List<Subscription> subscriptionList = new ArrayList<>();
MQConst.GoblinQueue stream = MQConst.GoblinQueue.GOBLIN_XLS_OPERA;
this.initStream(stringRedisTemplate, stream.getKey(), stream.getGroup());
for (int i = 0; i < 1; 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()), consumerGoblinXlsRdsReceiver
));
listenerContainer.start();
}
return subscriptionList;
}
/* -------------------------------------------------------- ｜ */
}
liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-base/src/main/java/com/liquidnet/service/consumer/base/config/ConsumerTransferRedisStreamConfig.java 0 → 100644
View file @ 98c887bf
package com.liquidnet.service.consumer.base.config;
import com.liquidnet.common.cache.redis.config.RedisStreamConfig;
import com.liquidnet.service.base.constant.MQConst;
import com.liquidnet.service.consumer.base.receiver.ConsumerKylinTransferOverTimeRdsReceiver;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.stream.Consumer;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.connection.stream.StreamOffset;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;
import org.springframework.data.redis.stream.Subscription;
import java.util.ArrayList;
import java.util.List;
@Configuration
public class ConsumerTransferRedisStreamConfig extends RedisStreamConfig {
@Autowired
ConsumerKylinTransferOverTimeRdsReceiver consumerKylinTransferOverTimeRdsReceiver;
@Autowired
StringRedisTemplate stringRedisTemplate;
/* —————————————————————————— ｜ —————————————————————————— ｜ —————————————————————————— */
/* -------------------------------------------------------- ｜ 验票更新 */
// 转赠过期定时任务
@Bean
public List<Subscription> subscriptionKylinTransferOverTime(RedisConnectionFactory factory) {
List<Subscription> subscriptionList = new ArrayList<>();
MQConst.KylinQueue stream = MQConst.KylinQueue.SQL_TRANSFER_OVERTIME;
this.initStream(stringRedisTemplate, stream.getKey(), stream.getGroup());
for (int i = 0; i < 1; 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()), consumerKylinTransferOverTimeRdsReceiver
));
listenerContainer.start();
}
return subscriptionList;
}
}
liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-base/src/main/java/com/liquidnet/service/consumer/base/receiver/AbstractOptOrderTransferOverTimeRedisReceiver.java 0 → 100644
View file @ 98c887bf
package com.liquidnet.service.consumer.base.receiver;
import com.fasterxml.jackson.core.type.TypeReference;
import com.liquidnet.commons.lang.util.CollectionUtil;
import com.liquidnet.commons.lang.util.HttpUtil;
import com.liquidnet.commons.lang.util.JsonUtils;
import com.liquidnet.service.base.OrderCloseMapping;
import com.liquidnet.service.base.ResponseDto;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
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 org.springframework.util.LinkedMultiValueMap;
import org.springframework.util.MultiValueMap;
import java.util.HashMap;
@Slf4j
public abstract class AbstractOptOrderTransferOverTimeRedisReceiver implements StreamListener<String, MapRecord<String, String, String>> {
@Autowired
StringRedisTemplate stringRedisTemplate;
@Value("${liquidnet.service.kylin.url}")
private String kylinUrl;
@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 {
OrderCloseMapping.orderCloseMessage mqMessage = JsonUtils.fromJson(msg, OrderCloseMapping.orderCloseMessage.class);
if (mqMessage == null) {
} else {
for (int x = 0; x < mqMessage.getOrderTicketIds().size(); x++) {
String orderTicketId = mqMessage.getOrderTicketIds().get(x);
MultiValueMap<String, String> params = new LinkedMultiValueMap();
params.add("orderId", orderTicketId);
MultiValueMap<String, String> headers = CollectionUtil.linkedMultiValueMapStringString();
headers.add("Accept", "application/json;charset=UTF-8");
String returnData = HttpUtil.post(kylinUrl + "/inner/consumer/orderTransferOverTime", params, headers);
ResponseDto<Boolean> rsp = JsonUtils.fromJson(returnData, new TypeReference<ResponseDto<Boolean>>() {});
}
}
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-base/src/main/java/com/liquidnet/service/consumer/base/receiver/AbstractXlsRedisReceiver.java 0 → 100644
View file @ 98c887bf
package com.liquidnet.service.consumer.base.receiver;
import com.alibaba.fastjson.JSON;
import com.fasterxml.jackson.core.type.TypeReference;
import com.liquidnet.common.cache.redis.util.RedisUtil;
import com.liquidnet.commons.lang.util.CollectionUtil;
import com.liquidnet.commons.lang.util.HttpUtil;
import com.liquidnet.commons.lang.util.JsonUtils;
import com.liquidnet.service.base.ResponseDto;
import com.liquidnet.service.consumer.base.service.IBaseDao;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.util.LinkedMultiValueMap;
import org.springframework.util.MultiValueMap;
import java.time.LocalDateTime;
import java.util.LinkedList;
import java.util.Map;
@Slf4j
public abstract class AbstractXlsRedisReceiver implements StreamListener<String, MapRecord<String, String, String>> {
@Autowired
private IBaseDao baseDao;
@Autowired
StringRedisTemplate stringRedisTemplate;
@Value("${liquidnet.service.goblin.url}")
private String goblinUrl;
private static final String SQL_INSERT_GOODS_BUY_ROSTER_LOG = "INSERT INTO goblin_goods_buy_roster_log (sku_id,buy_roster,buy_roster_type,parsing_result,created_at)VALUES(?,?,?,?,?)";
@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());
log.info("XLS MESSAGE 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(Map<String, String> message) {
LinkedList<Object[]> objs = CollectionUtil.linkedListObjectArr();
String xlsPath = null, skuId = null;
String oXlsPath = null;
Integer type = null;
boolean aBoolean = false;
try {
String[] path = message.get("message").split(",");
if (path.length == 0) {
xlsPath = "";
} else {
xlsPath = path[0];
}
if (path.length > 1) {
oXlsPath = path[1];
}
String finalSkuId = (skuId = message.get("skuId"));
String listId;
listId = message.getOrDefault("listId", "");
Integer finalType = (type = Integer.parseInt(message.get("type")));
if (finalType.equals(1) || finalType.equals(2)) {
MultiValueMap<String, String> params = new LinkedMultiValueMap();
params.add("finalSkuId", finalSkuId);
params.add("finalType", finalType.toString());
params.add("xlsPath", xlsPath);
MultiValueMap<String, String> headers = CollectionUtil.linkedMultiValueMapStringString();
headers.add("Accept", "application/json;charset=UTF-8");
String returnData = HttpUtil.post(goblinUrl + "/goblin/inner/consumerType12", params, headers);
ResponseDto<Boolean> rsp = JsonUtils.fromJson(returnData, new TypeReference<ResponseDto<Boolean>>() {});
objs.add(new Object[]{skuId, xlsPath, type, 1, LocalDateTime.now()});
aBoolean = baseDao.batchSql(SQL_INSERT_GOODS_BUY_ROSTER_LOG, objs);
} else if (finalType.equals(3) || finalType.equals(4)) {
MultiValueMap<String, String> params = new LinkedMultiValueMap();
params.add("finalSkuId", finalSkuId);
params.add("finalType", finalType.toString());
params.add("xlsPath", xlsPath);
params.add("oXlsPath", oXlsPath);
params.add("listId", listId);
MultiValueMap<String, String> headers = CollectionUtil.linkedMultiValueMapStringString();
headers.add("Accept", "application/json;charset=UTF-8");
String returnData = HttpUtil.post(goblinUrl + "/goblin/inner/consumerType34", params, headers);
ResponseDto<Boolean> rsp = JsonUtils.fromJson(returnData, new TypeReference<ResponseDto<Boolean>>() {});
aBoolean = true;
}
} catch (Exception e) {
log.error("CONSUMER MSG EX_HANDLE ==> [{}]:{}", this.getRedisStreamKey(), message, e);
try {
if (type.equals(1) || type.equals(2)) {
objs.add(new Object[]{skuId, xlsPath, type, 2, LocalDateTime.now()});
aBoolean = baseDao.batchSql(SQL_INSERT_GOODS_BUY_ROSTER_LOG, objs);
}
} catch (Exception ignored) {
}
}
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-base/src/main/java/com/liquidnet/service/consumer/base/receiver/ConsumerGoblinXlsRdsReceiver.java 0 → 100644
View file @ 98c887bf
package com.liquidnet.service.consumer.base.receiver;
import com.liquidnet.service.base.constant.MQConst;
import org.springframework.stereotype.Component;
@Component
public class ConsumerGoblinXlsRdsReceiver extends AbstractXlsRedisReceiver {
@Override
protected String getRedisStreamKey() {
return MQConst.GoblinQueue.GOBLIN_XLS_OPERA.getKey();
}
@Override
protected String getRedisStreamGroup() {
return MQConst.GoblinQueue.GOBLIN_XLS_OPERA.getGroup();
}
}
liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-base/src/main/java/com/liquidnet/service/consumer/base/receiver/ConsumerKylinTransferOverTimeRdsReceiver.java 0 → 100644
View file @ 98c887bf
package com.liquidnet.service.consumer.base.receiver;
import com.liquidnet.service.base.constant.MQConst;
import org.springframework.stereotype.Component;
@Component
public class ConsumerKylinTransferOverTimeRdsReceiver extends AbstractOptOrderTransferOverTimeRedisReceiver {
@Override
protected String getRedisStreamKey() {
return MQConst.KylinQueue.SQL_TRANSFER_OVERTIME.getKey();
}
@Override
protected String getRedisStreamGroup() {
return MQConst.KylinQueue.SQL_TRANSFER_OVERTIME.getGroup();
}
}
liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-kylin/src/main/java/com/liquidnet/service/consumer/kylin/receiver/AbstractOptOrderTransferOverTimeRedisReceiver.java
View file @ 98c887bf
package com.liquidnet.service.consumer.kylin.receiver; //package com.liquidnet.service.consumer.kylin.receiver;
//
import com.liquidnet.common.cache.redis.util.RedisUtil; //import com.liquidnet.common.cache.redis.util.RedisUtil;
import com.liquidnet.commons.lang.util.CollectionUtil; //import com.liquidnet.commons.lang.util.CollectionUtil;
import com.liquidnet.commons.lang.util.JsonUtils; //import com.liquidnet.commons.lang.util.JsonUtils;
import com.liquidnet.service.base.OrderCloseMapping; //import com.liquidnet.service.base.OrderCloseMapping;
import com.liquidnet.service.consumer.kylin.utils.KylinUtils; //import com.liquidnet.service.consumer.kylin.utils.KylinUtils;
import com.liquidnet.service.kylin.constant.KylinRedisConst; //import com.liquidnet.service.kylin.constant.KylinRedisConst;
import com.liquidnet.service.kylin.dto.vo.mongo.KylinOrderTicketVo; //import com.liquidnet.service.kylin.dto.vo.mongo.KylinOrderTicketVo;
import lombok.extern.slf4j.Slf4j; //import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired; //import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.stream.MapRecord; //import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.StreamRecords; //import org.springframework.data.redis.connection.stream.StreamRecords;
import org.springframework.data.redis.core.StringRedisTemplate; //import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.stream.StreamListener; //import org.springframework.data.redis.stream.StreamListener;
//
import java.util.HashMap; //import java.util.HashMap;
//
@Slf4j //@Slf4j
public abstract class AbstractOptOrderTransferOverTimeRedisReceiver implements StreamListener<String, MapRecord<String, String, String>> { //public abstract class AbstractOptOrderTransferOverTimeRedisReceiver implements StreamListener<String, MapRecord<String, String, String>> {
@Autowired // @Autowired
StringRedisTemplate stringRedisTemplate; // StringRedisTemplate stringRedisTemplate;
@Autowired // @Autowired
private RedisUtil redisUtil; // private RedisUtil redisUtil;
@Autowired // @Autowired
private KylinUtils kylinUtils; // private KylinUtils kylinUtils;
//
@Override // @Override
public void onMessage(MapRecord<String, String, String> message) { // public void onMessage(MapRecord<String, String, String> message) {
String redisStreamKey = this.getRedisStreamKey(); // String redisStreamKey = this.getRedisStreamKey();
log.debug("CONSUMER MSG[streamKey:{},messageId:{},stream:{},body:{}]", redisStreamKey, message.getId(), message.getStream(), message.getValue()); // log.debug("CONSUMER MSG[streamKey:{},messageId:{},stream:{},body:{}]", redisStreamKey, message.getId(), message.getStream(), message.getValue());
boolean result = this.consumerMessageHandler(message.getValue().get("message")); // boolean result = this.consumerMessageHandler(message.getValue().get("message"));
log.info("CONSUMER MSG RESULT:{} ==> [{}]MESSAGE_ID:{}", result, redisStreamKey, message.getId()); // log.info("CONSUMER MSG RESULT:{} ==> [{}]MESSAGE_ID:{}", result, redisStreamKey, message.getId());
//
try { // try {
stringRedisTemplate.opsForStream().acknowledge(getRedisStreamGroup(), message); // stringRedisTemplate.opsForStream().acknowledge(getRedisStreamGroup(), message);
} catch (Exception e) { // } catch (Exception e) {
log.error("#CONSUMER MSG EX_ACK ==> [{}]RESULT:{},MESSAGE:{}", redisStreamKey, result, message.getValue(), e); // log.error("#CONSUMER MSG EX_ACK ==> [{}]RESULT:{},MESSAGE:{}", redisStreamKey, result, message.getValue(), e);
} // }
try { // try {
stringRedisTemplate.opsForStream().delete(redisStreamKey, message.getId()); // stringRedisTemplate.opsForStream().delete(redisStreamKey, message.getId());
} catch (Exception e) { // } catch (Exception e) {
log.error("#CONSUMER MSG EX_DEL ==> [{}]RESULT:{},MESSAGE:{}", redisStreamKey, result, message.getValue(), e); // log.error("#CONSUMER MSG EX_DEL ==> [{}]RESULT:{},MESSAGE:{}", redisStreamKey, result, message.getValue(), e);
} // }
} // }
//
private boolean consumerMessageHandler(String msg) { // private boolean consumerMessageHandler(String msg) {
boolean aBoolean = false; // boolean aBoolean = false;
try { // try {
OrderCloseMapping.orderCloseMessage mqMessage = JsonUtils.fromJson(msg, OrderCloseMapping.orderCloseMessage.class); // OrderCloseMapping.orderCloseMessage mqMessage = JsonUtils.fromJson(msg, OrderCloseMapping.orderCloseMessage.class);
if (mqMessage == null) { // if (mqMessage == null) {
//
} else { // } else {
for (int x = 0; x < mqMessage.getOrderTicketIds().size(); x++) { // for (int x = 0; x < mqMessage.getOrderTicketIds().size(); x++) {
String orderTicketId = mqMessage.getOrderTicketIds().get(x); // String orderTicketId = mqMessage.getOrderTicketIds().get(x);
KylinOrderTicketVo vo = kylinUtils.getOrderTicketVo(orderTicketId); // KylinOrderTicketVo vo = kylinUtils.getOrderTicketVo(orderTicketId);
String uid = vo.getTransferUid(); // String uid = vo.getTransferUid();
vo.setTransferUid(""); // vo.setTransferUid("");
vo.setTransferStatus(0); // vo.setTransferStatus(0);
redisUtil.del(KylinRedisConst.ORDER_TRANSFER.concat(uid)); // redisUtil.del(KylinRedisConst.ORDER_TRANSFER.concat(uid));
redisUtil.set("kylin:order:id:" + orderTicketId, vo); // redisUtil.set("kylin:order:id:" + orderTicketId, vo);
kylinUtils.resetOrderListVo(vo.getUserId(), 2, orderTicketId, vo); // kylinUtils.resetOrderListVo(vo.getUserId(), 2, orderTicketId, vo);
} // }
} // }
aBoolean = true; // aBoolean = true;
} catch (Exception e) { // } catch (Exception e) {
log.error("CONSUMER MSG EX_HANDLE ==> [{}]:{}", this.getRedisStreamKey(), msg, e); // log.error("CONSUMER MSG EX_HANDLE ==> [{}]:{}", this.getRedisStreamKey(), msg, e);
} finally { // } finally {
if (!aBoolean) { // if (!aBoolean) {
HashMap<String, String> map = CollectionUtil.mapStringString(); // HashMap<String, String> map = CollectionUtil.mapStringString();
map.put("message", msg); // map.put("message", msg);
stringRedisTemplate.opsForStream().add(StreamRecords.mapBacked(map).withStreamKey(this.getRedisStreamKey())); // stringRedisTemplate.opsForStream().add(StreamRecords.mapBacked(map).withStreamKey(this.getRedisStreamKey()));
} // }
} // }
return aBoolean; // return aBoolean;
} // }
//
protected abstract String getRedisStreamKey(); // protected abstract String getRedisStreamKey();
//
protected abstract String getRedisStreamGroup(); // protected abstract String getRedisStreamGroup();
} //}
liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-kylin/src/main/java/com/liquidnet/service/consumer/kylin/receiver/AbstractXlsRedisReceiver.java
View file @ 98c887bf
package com.liquidnet.service.consumer.kylin.receiver; //package com.liquidnet.service.consumer.kylin.receiver;
//
import com.alibaba.excel.EasyExcel; //import com.alibaba.excel.EasyExcel;
import com.alibaba.excel.read.listener.PageReadListener; //import com.alibaba.excel.read.listener.PageReadListener;
import com.alibaba.fastjson.JSON; //import com.alibaba.fastjson.JSON;
import com.liquidnet.common.cache.redis.util.RedisUtil; //import com.liquidnet.common.cache.redis.util.RedisUtil;
import com.liquidnet.commons.lang.util.CollectionUtil; //import com.liquidnet.commons.lang.util.CollectionUtil;
import com.liquidnet.service.consumer.kylin.dto.PhoneDto; //import com.liquidnet.service.consumer.kylin.dto.PhoneDto;
import com.liquidnet.service.consumer.kylin.dto.PhoneUnIdDto; //import com.liquidnet.service.consumer.kylin.dto.PhoneUnIdDto;
import com.liquidnet.service.consumer.kylin.service.IBaseDao; //import com.liquidnet.service.consumer.kylin.service.IBaseDao;
import com.liquidnet.service.goblin.constant.GoblinRedisConst; //import com.liquidnet.service.goblin.constant.GoblinRedisConst;
import lombok.extern.slf4j.Slf4j; //import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired; //import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.stream.MapRecord; //import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.core.StringRedisTemplate; //import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.stream.StreamListener; //import org.springframework.data.redis.stream.StreamListener;
//
import java.net.URL; //import java.net.URL;
import java.time.LocalDateTime; //import java.time.LocalDateTime;
import java.util.HashMap; //import java.util.HashMap;
import java.util.LinkedList; //import java.util.LinkedList;
import java.util.Map; //import java.util.Map;
//
@Slf4j //@Slf4j
public abstract class AbstractXlsRedisReceiver implements StreamListener<String, MapRecord<String, String, String>> { //public abstract class AbstractXlsRedisReceiver implements StreamListener<String, MapRecord<String, String, String>> {
@Autowired // @Autowired
private IBaseDao baseDao; // private IBaseDao baseDao;
@Autowired // @Autowired
StringRedisTemplate stringRedisTemplate; // StringRedisTemplate stringRedisTemplate;
@Autowired // @Autowired
private RedisUtil redisUtil; // private RedisUtil redisUtil;
//
private static final String SQL_INSERT_GOODS_BUY_ROSTER_LOG = "INSERT INTO goblin_goods_buy_roster_log (sku_id,buy_roster,buy_roster_type,parsing_result,created_at)VALUES(?,?,?,?,?)"; // private static final String SQL_INSERT_GOODS_BUY_ROSTER_LOG = "INSERT INTO goblin_goods_buy_roster_log (sku_id,buy_roster,buy_roster_type,parsing_result,created_at)VALUES(?,?,?,?,?)";
//
@Override // @Override
public void onMessage(MapRecord<String, String, String> message) { // public void onMessage(MapRecord<String, String, String> message) {
String redisStreamKey = this.getRedisStreamKey(); // String redisStreamKey = this.getRedisStreamKey();
log.debug("CONSUMER MSG[streamKey:{},messageId:{},stream:{},body:{}]", redisStreamKey, message.getId(), message.getStream(), message.getValue()); // log.debug("CONSUMER MSG[streamKey:{},messageId:{},stream:{},body:{}]", redisStreamKey, message.getId(), message.getStream(), message.getValue());
boolean result = this.consumerMessageHandler(message.getValue()); // boolean result = this.consumerMessageHandler(message.getValue());
log.info("XLS MESSAGE CONSUMER MSG RESULT:{} ==> [{}]MESSAGE_ID:{}", result, redisStreamKey, message.getId()); // log.info("XLS MESSAGE CONSUMER MSG RESULT:{} ==> [{}]MESSAGE_ID:{}", result, redisStreamKey, message.getId());
//
try { // try {
stringRedisTemplate.opsForStream().acknowledge(getRedisStreamGroup(), message); // stringRedisTemplate.opsForStream().acknowledge(getRedisStreamGroup(), message);
} catch (Exception e) { // } catch (Exception e) {
log.error("#CONSUMER MSG EX_ACK ==> [{}]RESULT:{},MESSAGE:{}", redisStreamKey, result, message.getValue(), e); // log.error("#CONSUMER MSG EX_ACK ==> [{}]RESULT:{},MESSAGE:{}", redisStreamKey, result, message.getValue(), e);
} // }
try { // try {
stringRedisTemplate.opsForStream().delete(redisStreamKey, message.getId()); // stringRedisTemplate.opsForStream().delete(redisStreamKey, message.getId());
} catch (Exception e) { // } catch (Exception e) {
log.error("#CONSUMER MSG EX_DEL ==> [{}]RESULT:{},MESSAGE:{}", redisStreamKey, result, message.getValue(), e); // log.error("#CONSUMER MSG EX_DEL ==> [{}]RESULT:{},MESSAGE:{}", redisStreamKey, result, message.getValue(), e);
} // }
} // }
//
private boolean consumerMessageHandler(Map<String, String> message) { // private boolean consumerMessageHandler(Map<String, String> message) {
LinkedList<Object[]> objs = CollectionUtil.linkedListObjectArr(); // LinkedList<Object[]> objs = CollectionUtil.linkedListObjectArr();
String xlsPath = null, skuId = null; // String xlsPath = null, skuId = null;
String oXlsPath = null; // String oXlsPath = null;
Integer type = null; // Integer type = null;
boolean aBoolean = false; // boolean aBoolean = false;
try { // try {
String[] path = message.get("message").split(","); // String[] path = message.get("message").split(",");
if (path.length == 0) { // if (path.length == 0) {
xlsPath = ""; // xlsPath = "";
} else { // } else {
xlsPath = path[0]; // xlsPath = path[0];
} // }
if (path.length > 1) { // if (path.length > 1) {
oXlsPath = path[1]; // oXlsPath = path[1];
} // }
String finalSkuId = (skuId = message.get("skuId")); // String finalSkuId = (skuId = message.get("skuId"));
String listId; // String listId;
//// try {
// listId = message.getOrDefault("listId", "");
//// }catch (Exception e){
//// listId = "";
//// }
// Integer finalType = (type = Integer.parseInt(message.get("type")));
// if (finalType.equals(1) || finalType.equals(2)) {
// EasyExcel.read(new URL(xlsPath).openStream(), PhoneDto.class, new PageReadListener<PhoneDto>(dataList -> {
// for (PhoneDto data : dataList) {
// if (data.getMobile() == null) {
// continue;
// }
// String redisKey = GoblinRedisConst.REDIS_CAN_BUY.concat(finalSkuId + ":").concat(data.getMobile());
// if (finalType.equals(1)) {//添加
// if (log.isDebugEnabled()) {
// log.debug("添加 读取到一条数据{}", JSON.toJSONString(data));
// }
// redisUtil.set(redisKey, 0);
// } else {
// if (log.isDebugEnabled()) {
// log.debug("删除 读取到一条数据{}", JSON.toJSONString(data));
// }
// redisUtil.del(redisKey);
// }
// }
// })).sheet().doRead();
// objs.add(new Object[]{skuId, xlsPath, type, 1, LocalDateTime.now()});
// aBoolean = baseDao.batchSql(SQL_INSERT_GOODS_BUY_ROSTER_LOG, objs);
// } else if (finalType.equals(3) || finalType.equals(4)) {
// if (oXlsPath != null && !oXlsPath.equals("")) {//删除旧的黑白名单
// EasyExcel.read(new URL(oXlsPath).openStream(), new PageReadListener<HashMap<String, String>>(dataList -> {
// for (HashMap<String, String> data : dataList) {
// if (data.get(0) == null) {
// continue;
// }
// if (finalType.equals(3)) {
// log.debug("删除 白名单 读取到一条数据{}", JSON.toJSONString(data));
// redisUtil.del(GoblinRedisConst.REDIS_WHITE.concat(listId + ":").concat(finalSkuId + ":").concat(data.get(0)));
// } else {
// log.debug("删除 黑名单 读取到一条数据{}", JSON.toJSONString(data));
// redisUtil.del(GoblinRedisConst.REDIS_BLACK.concat(listId + ":").concat(finalSkuId + ":").concat(data.get(0)));
// }
// }
// })).sheet().doRead();
// }
// if (xlsPath != null && !xlsPath.equals("")) {//添加
// EasyExcel.read(new URL(xlsPath).openStream(), new PageReadListener<HashMap<String, String>>(dataList -> {
// for (HashMap<String, String> data : dataList) {
// if (data.get(0) == null) {
// continue;
// }
// if (finalType.equals(3)) {
// log.debug("添加 白名单 读取到一条数据{}", JSON.toJSONString(data));
// redisUtil.set(GoblinRedisConst.REDIS_WHITE.concat(listId + ":").concat(finalSkuId + ":").concat(data.get(0)), 1);
// } else {
// log.debug("添加 黑名单 读取到一条数据{}", JSON.toJSONString(data));
// redisUtil.set(GoblinRedisConst.REDIS_BLACK.concat(listId + ":").concat(finalSkuId + ":").concat(data.get(0)), 1);
// }
// }
// })).sheet().doRead();
// }
// aBoolean = true;
// }
// } catch (Exception e) {
// log.error("CONSUMER MSG EX_HANDLE ==> [{}]:{}", this.getRedisStreamKey(), message, e);
// try { // try {
listId = message.getOrDefault("listId", ""); // if (type.equals(1) || type.equals(2)) {
// }catch (Exception e){ // objs.add(new Object[]{skuId, xlsPath, type, 2, LocalDateTime.now()});
// listId = ""; // aBoolean = baseDao.batchSql(SQL_INSERT_GOODS_BUY_ROSTER_LOG, objs);
// }
// } catch (Exception ignored) {
// } // }
Integer finalType = (type = Integer.parseInt(message.get("type"))); // }
if (finalType.equals(1) || finalType.equals(2)) { // return aBoolean;
EasyExcel.read(new URL(xlsPath).openStream(), PhoneDto.class, new PageReadListener<PhoneDto>(dataList -> { // }
for (PhoneDto data : dataList) { //
if (data.getMobile() == null) { // protected abstract String getRedisStreamKey();
continue; //
} // protected abstract String getRedisStreamGroup();
String redisKey = GoblinRedisConst.REDIS_CAN_BUY.concat(finalSkuId + ":").concat(data.getMobile()); //}
if (finalType.equals(1)) {//添加 \ No newline at end of file
if (log.isDebugEnabled()) {
log.debug("添加 读取到一条数据{}", JSON.toJSONString(data));
}
redisUtil.set(redisKey, 0);
} else {
if (log.isDebugEnabled()) {
log.debug("删除 读取到一条数据{}", JSON.toJSONString(data));
}
redisUtil.del(redisKey);
}
}
})).sheet().doRead();
objs.add(new Object[]{skuId, xlsPath, type, 1, LocalDateTime.now()});
aBoolean = baseDao.batchSql(SQL_INSERT_GOODS_BUY_ROSTER_LOG, objs);
} else if (finalType.equals(3) || finalType.equals(4)) {
if (oXlsPath != null && !oXlsPath.equals("")) {//删除旧的黑白名单
EasyExcel.read(new URL(oXlsPath).openStream(), new PageReadListener<HashMap<String, String>>(dataList -> {
for (HashMap<String, String> data : dataList) {
if (data.get(0) == null) {
continue;
}
if (finalType.equals(3)) {
log.debug("删除 白名单 读取到一条数据{}", JSON.toJSONString(data));
redisUtil.del(GoblinRedisConst.REDIS_WHITE.concat(listId + ":").concat(finalSkuId + ":").concat(data.get(0)));
} else {
log.debug("删除 黑名单 读取到一条数据{}", JSON.toJSONString(data));
redisUtil.del(GoblinRedisConst.REDIS_BLACK.concat(listId + ":").concat(finalSkuId + ":").concat(data.get(0)));
}
}
})).sheet().doRead();
}
if (xlsPath != null && !xlsPath.equals("")) {//添加
EasyExcel.read(new URL(xlsPath).openStream(), new PageReadListener<HashMap<String, String>>(dataList -> {
for (HashMap<String, String> data : dataList) {
if (data.get(0) == null) {
continue;
}
if (finalType.equals(3)) {
log.debug("添加 白名单 读取到一条数据{}", JSON.toJSONString(data));
redisUtil.set(GoblinRedisConst.REDIS_WHITE.concat(listId + ":").concat(finalSkuId + ":").concat(data.get(0)), 1);
} else {
log.debug("添加 黑名单 读取到一条数据{}", JSON.toJSONString(data));
redisUtil.set(GoblinRedisConst.REDIS_BLACK.concat(listId + ":").concat(finalSkuId + ":").concat(data.get(0)), 1);
}
}
})).sheet().doRead();
}
aBoolean = true;
}
} catch (Exception e) {
log.error("CONSUMER MSG EX_HANDLE ==> [{}]:{}", this.getRedisStreamKey(), message, e);
try {
if (type.equals(1) || type.equals(2)) {
objs.add(new Object[]{skuId, xlsPath, type, 2, LocalDateTime.now()});
aBoolean = baseDao.batchSql(SQL_INSERT_GOODS_BUY_ROSTER_LOG, objs);
}
} catch (Exception ignored) {
}
}
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-kylin/src/main/java/com/liquidnet/service/consumer/kylin/receiver/ConsumerGoblinXlsRdsReceiver.java
View file @ 98c887bf
package com.liquidnet.service.consumer.kylin.receiver; //package com.liquidnet.service.consumer.kylin.receiver;
//
import com.liquidnet.service.base.constant.MQConst; //import com.liquidnet.service.base.constant.MQConst;
import org.springframework.stereotype.Component; //import org.springframework.stereotype.Component;
//
@Component //@Component
public class ConsumerGoblinXlsRdsReceiver extends AbstractXlsRedisReceiver { //public class ConsumerGoblinXlsRdsReceiver extends AbstractXlsRedisReceiver {
@Override // @Override
protected String getRedisStreamKey() { // protected String getRedisStreamKey() {
return MQConst.GoblinQueue.GOBLIN_XLS_OPERA.getKey(); // return MQConst.GoblinQueue.GOBLIN_XLS_OPERA.getKey();
} // }
//
@Override // @Override
protected String getRedisStreamGroup() { // protected String getRedisStreamGroup() {
return MQConst.GoblinQueue.GOBLIN_XLS_OPERA.getGroup(); // return MQConst.GoblinQueue.GOBLIN_XLS_OPERA.getGroup();
} // }
} //}
liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-kylin/src/main/java/com/liquidnet/service/consumer/kylin/receiver/ConsumerKylinTransferOverTimeRdsReceiver.java
View file @ 98c887bf
package com.liquidnet.service.consumer.kylin.receiver; //package com.liquidnet.service.consumer.kylin.receiver;
//
import com.liquidnet.service.base.constant.MQConst; //import com.liquidnet.service.base.constant.MQConst;
import org.springframework.stereotype.Component; //import org.springframework.stereotype.Component;
//
@Component //@Component
public class ConsumerKylinTransferOverTimeRdsReceiver extends AbstractOptOrderTransferOverTimeRedisReceiver { //public class ConsumerKylinTransferOverTimeRdsReceiver extends AbstractOptOrderTransferOverTimeRedisReceiver {
@Override // @Override
protected String getRedisStreamKey() { // protected String getRedisStreamKey() {
return MQConst.KylinQueue.SQL_TRANSFER_OVERTIME.getKey(); // return MQConst.KylinQueue.SQL_TRANSFER_OVERTIME.getKey();
} // }
//
@Override // @Override
protected String getRedisStreamGroup() { // protected String getRedisStreamGroup() {
return MQConst.KylinQueue.SQL_TRANSFER_OVERTIME.getGroup(); // return MQConst.KylinQueue.SQL_TRANSFER_OVERTIME.getGroup();
} // }
} //}
liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-slowly/src/main/java/com/liquidnet/service/consumer/slowly/config/ConsumerGoblinXlsRedisStreamConfig.java
View file @ 98c887bf
package com.liquidnet.service.consumer.slowly.config; //package com.liquidnet.service.consumer.slowly.config;
//
import com.liquidnet.common.cache.redis.config.RedisStreamConfig; //import com.liquidnet.common.cache.redis.config.RedisStreamConfig;
import com.liquidnet.service.base.constant.MQConst; //import com.liquidnet.service.base.constant.MQConst;
import com.liquidnet.service.consumer.slowly.receiver.ConsumerGoblinXlsRdsReceiver; //import com.liquidnet.service.consumer.slowly.receiver.ConsumerGoblinXlsRdsReceiver;
import lombok.var; //import lombok.var;
import org.springframework.beans.factory.annotation.Autowired; //import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean; //import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration; //import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory; //import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.stream.Consumer; //import org.springframework.data.redis.connection.stream.Consumer;
import org.springframework.data.redis.connection.stream.MapRecord; //import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.ReadOffset; //import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.connection.stream.StreamOffset; //import org.springframework.data.redis.connection.stream.StreamOffset;
import org.springframework.data.redis.stream.StreamMessageListenerContainer; //import org.springframework.data.redis.stream.StreamMessageListenerContainer;
import org.springframework.data.redis.stream.Subscription; //import org.springframework.data.redis.stream.Subscription;
//
@Configuration //@Configuration
public class ConsumerGoblinXlsRedisStreamConfig extends RedisStreamConfig { //public class ConsumerGoblinXlsRedisStreamConfig extends RedisStreamConfig {
@Autowired // @Autowired
ConsumerGoblinXlsRdsReceiver consumerGoblinXlsRdsReceiver; // ConsumerGoblinXlsRdsReceiver consumerGoblinXlsRdsReceiver;
//
/** // /**
* 缺票登记 // * 缺票登记
* // *
* @param listenerContainer // * @param listenerContainer
* @param t // * @param t
* @return // * @return
*/ // */
private Subscription receiveGoblinXls(StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer, int t) { // private Subscription receiveGoblinXls(StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer, int t) {
return listenerContainer.receiveAutoAck(Consumer.from(MQConst.GoblinQueue.GOBLIN_XLS_OPERA.getGroup(), getConsumerName(MQConst.GoblinQueue.GOBLIN_XLS_OPERA.name() + t)), // return listenerContainer.receiveAutoAck(Consumer.from(MQConst.GoblinQueue.GOBLIN_XLS_OPERA.getGroup(), getConsumerName(MQConst.GoblinQueue.GOBLIN_XLS_OPERA.name() + t)),
StreamOffset.create(MQConst.GoblinQueue.GOBLIN_XLS_OPERA.getKey(), ReadOffset.lastConsumed()), consumerGoblinXlsRdsReceiver); // StreamOffset.create(MQConst.GoblinQueue.GOBLIN_XLS_OPERA.getKey(), ReadOffset.lastConsumed()), consumerGoblinXlsRdsReceiver);
} // }
//
/* —————————————————————————— ｜ —————————————————————————— ｜ —————————————————————————— */ // /* —————————————————————————— ｜ —————————————————————————— ｜ —————————————————————————— */
//
/* -------------------------------------------------------- ｜ 缺票登记 */ // /* -------------------------------------------------------- ｜ 缺票登记 */
@Bean
public Subscription subscriptionGoblinXls0(RedisConnectionFactory factory) {
var listenerContainer = this.buildStreamMessageListenerContainer(factory);
var subscription = receiveGoblinXls(listenerContainer, 0);
listenerContainer.start();
return subscription;
}
@Bean
public Subscription subscriptionGoblinXls1(RedisConnectionFactory factory) {
var listenerContainer = this.buildStreamMessageListenerContainer(factory);
var subscription = receiveGoblinXls(listenerContainer, 1);
listenerContainer.start();
return subscription;
}
// //
// @Bean // @Bean
// public Subscription subscriptionGoblinXls2(RedisConnectionFactory factory) { // public Subscription subscriptionGoblinXls0(RedisConnectionFactory factory) {
// var listenerContainer = this.buildStreamMessageListenerContainer(factory); // var listenerContainer = this.buildStreamMessageListenerContainer(factory);
// var subscription = receiveGoblinXls(listenerContainer, 2); // var subscription = receiveGoblinXls(listenerContainer, 0);
// listenerContainer.start(); // listenerContainer.start();
// return subscription; // return subscription;
// } // }
//
/* -------------------------------------------------------- ｜ */ // @Bean
} // public Subscription subscriptionGoblinXls1(RedisConnectionFactory factory) {
// var listenerContainer = this.buildStreamMessageListenerContainer(factory);
// var subscription = receiveGoblinXls(listenerContainer, 1);
// listenerContainer.start();
// return subscription;
// }
////
//// @Bean
//// public Subscription subscriptionGoblinXls2(RedisConnectionFactory factory) {
//// var listenerContainer = this.buildStreamMessageListenerContainer(factory);
//// var subscription = receiveGoblinXls(listenerContainer, 2);
//// listenerContainer.start();
//// return subscription;
//// }
//
// /* -------------------------------------------------------- ｜ */
//}
liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-slowly/src/main/java/com/liquidnet/service/consumer/slowly/config/ConsumerKylinOptTransferOverTimeRedisStreamConfig.java
View file @ 98c887bf
package com.liquidnet.service.consumer.slowly.config; //package com.liquidnet.service.consumer.slowly.config;
//
import com.liquidnet.common.cache.redis.config.RedisStreamConfig; //import com.liquidnet.common.cache.redis.config.RedisStreamConfig;
import com.liquidnet.service.consumer.slowly.receiver.ConsumerKylinTransferOverTimeRdsReceiver; //import com.liquidnet.service.consumer.slowly.receiver.ConsumerKylinTransferOverTimeRdsReceiver;
import lombok.var; //import lombok.var;
import org.springframework.beans.factory.annotation.Autowired; //import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean; //import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration; //import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory; //import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.stream.Consumer; //import org.springframework.data.redis.connection.stream.Consumer;
import org.springframework.data.redis.connection.stream.MapRecord; //import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.ReadOffset; //import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.connection.stream.StreamOffset; //import org.springframework.data.redis.connection.stream.StreamOffset;
import org.springframework.data.redis.stream.StreamMessageListenerContainer; //import org.springframework.data.redis.stream.StreamMessageListenerContainer;
import org.springframework.data.redis.stream.Subscription; //import org.springframework.data.redis.stream.Subscription;
//
import static com.liquidnet.service.base.constant.MQConst.KylinQueue.SQL_TRANSFER_OVERTIME; //import static com.liquidnet.service.base.constant.MQConst.KylinQueue.SQL_TRANSFER_OVERTIME;
//
@Configuration //@Configuration
public class ConsumerKylinOptTransferOverTimeRedisStreamConfig extends RedisStreamConfig { //public class ConsumerKylinOptTransferOverTimeRedisStreamConfig extends RedisStreamConfig {
@Autowired // @Autowired
ConsumerKylinTransferOverTimeRdsReceiver consumerKylinTransferOverTimeRdsReceiver; // ConsumerKylinTransferOverTimeRdsReceiver consumerKylinTransferOverTimeRdsReceiver;
//
/** // /**
* 验票更新 // * 验票更新
* // *
* @param listenerContainer // * @param listenerContainer
* @param t // * @param t
* @return // * @return
*/ // */
private Subscription receiveTransferOverTimeResult(StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer, int t) { // private Subscription receiveTransferOverTimeResult(StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer, int t) {
return listenerContainer.receiveAutoAck(Consumer.from(SQL_TRANSFER_OVERTIME.getGroup(), getConsumerName(SQL_TRANSFER_OVERTIME.name() + t)), // return listenerContainer.receiveAutoAck(Consumer.from(SQL_TRANSFER_OVERTIME.getGroup(), getConsumerName(SQL_TRANSFER_OVERTIME.name() + t)),
StreamOffset.create(SQL_TRANSFER_OVERTIME.getKey(), ReadOffset.lastConsumed()), consumerKylinTransferOverTimeRdsReceiver); // StreamOffset.create(SQL_TRANSFER_OVERTIME.getKey(), ReadOffset.lastConsumed()), consumerKylinTransferOverTimeRdsReceiver);
} // }
//
/* —————————————————————————— ｜ —————————————————————————— ｜ —————————————————————————— */ // /* —————————————————————————— ｜ —————————————————————————— ｜ —————————————————————————— */
//
/* -------------------------------------------------------- ｜ 验票更新 */ // /* -------------------------------------------------------- ｜ 验票更新 */
//
@Bean // @Bean
public Subscription subscriptionReceiveTransferOverTime0(RedisConnectionFactory factory) { // public Subscription subscriptionReceiveTransferOverTime0(RedisConnectionFactory factory) {
var listenerContainer = this.buildStreamMessageListenerContainer(factory); // var listenerContainer = this.buildStreamMessageListenerContainer(factory);
var subscription = receiveTransferOverTimeResult(listenerContainer, 0); // var subscription = receiveTransferOverTimeResult(listenerContainer, 0);
listenerContainer.start(); // listenerContainer.start();
return subscription; // return subscription;
} // }
//
@Bean // @Bean
public Subscription subscriptionReceiveTransferOverTime1(RedisConnectionFactory factory) { // public Subscription subscriptionReceiveTransferOverTime1(RedisConnectionFactory factory) {
var listenerContainer = this.buildStreamMessageListenerContainer(factory); // var listenerContainer = this.buildStreamMessageListenerContainer(factory);
var subscription = receiveTransferOverTimeResult(listenerContainer, 1); // var subscription = receiveTransferOverTimeResult(listenerContainer, 1);
listenerContainer.start(); // listenerContainer.start();
return subscription; // return subscription;
} // }
//
@Bean // @Bean
public Subscription subscriptionReceiveTransferOverTime2(RedisConnectionFactory factory) { // public Subscription subscriptionReceiveTransferOverTime2(RedisConnectionFactory factory) {
var listenerContainer = this.buildStreamMessageListenerContainer(factory); // var listenerContainer = this.buildStreamMessageListenerContainer(factory);
var subscription = receiveTransferOverTimeResult(listenerContainer, 2); // var subscription = receiveTransferOverTimeResult(listenerContainer, 2);
listenerContainer.start(); // listenerContainer.start();
return subscription; // return subscription;
} // }
//
@Bean // @Bean
public Subscription subscriptionReceiveTransferOverTime3(RedisConnectionFactory factory) { // public Subscription subscriptionReceiveTransferOverTime3(RedisConnectionFactory factory) {
var listenerContainer = this.buildStreamMessageListenerContainer(factory); // var listenerContainer = this.buildStreamMessageListenerContainer(factory);
var subscription = receiveTransferOverTimeResult(listenerContainer, 3); // var subscription = receiveTransferOverTimeResult(listenerContainer, 3);
listenerContainer.start(); // listenerContainer.start();
return subscription; // return subscription;
} // }
//
@Bean // @Bean
public Subscription subscriptionReceiveTransferOverTime4(RedisConnectionFactory factory) { // public Subscription subscriptionReceiveTransferOverTime4(RedisConnectionFactory factory) {
var listenerContainer = this.buildStreamMessageListenerContainer(factory); // var listenerContainer = this.buildStreamMessageListenerContainer(factory);
var subscription = receiveTransferOverTimeResult(listenerContainer, 4); // var subscription = receiveTransferOverTimeResult(listenerContainer, 4);
listenerContainer.start(); // listenerContainer.start();
return subscription; // return subscription;
} // }
//
/* -------------------------------------------------------- ｜ */ // /* -------------------------------------------------------- ｜ */
} //}
liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-slowly/src/main/java/com/liquidnet/service/consumer/slowly/receiver/AbstractOptOrderTransferOverTimeRedisReceiver.java
View file @ 98c887bf
package com.liquidnet.service.consumer.slowly.receiver; //package com.liquidnet.service.consumer.slowly.receiver;
//
import com.liquidnet.common.cache.redis.util.RedisUtil; //import com.liquidnet.common.cache.redis.util.RedisUtil;
import com.liquidnet.commons.lang.util.CollectionUtil; //import com.liquidnet.commons.lang.util.CollectionUtil;
import com.liquidnet.commons.lang.util.JsonUtils; //import com.liquidnet.commons.lang.util.JsonUtils;
import com.liquidnet.service.base.OrderCloseMapping; //import com.liquidnet.service.base.OrderCloseMapping;
import com.liquidnet.service.consumer.slowly.utils.KylinUtils; //import com.liquidnet.service.consumer.slowly.utils.KylinUtils;
import com.liquidnet.service.kylin.constant.KylinRedisConst; //import com.liquidnet.service.kylin.constant.KylinRedisConst;
import com.liquidnet.service.kylin.dto.vo.mongo.KylinOrderTicketVo; //import com.liquidnet.service.kylin.dto.vo.mongo.KylinOrderTicketVo;
import lombok.extern.slf4j.Slf4j; //import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired; //import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.stream.MapRecord; //import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.StreamRecords; //import org.springframework.data.redis.connection.stream.StreamRecords;
import org.springframework.data.redis.core.StringRedisTemplate; //import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.stream.StreamListener; //import org.springframework.data.redis.stream.StreamListener;
//
import java.util.HashMap; //import java.util.HashMap;
//
@Slf4j //@Slf4j
public abstract class AbstractOptOrderTransferOverTimeRedisReceiver implements StreamListener<String, MapRecord<String, String, String>> { //public abstract class AbstractOptOrderTransferOverTimeRedisReceiver implements StreamListener<String, MapRecord<String, String, String>> {
@Autowired // @Autowired
StringRedisTemplate stringRedisTemplate; // StringRedisTemplate stringRedisTemplate;
@Autowired // @Autowired
private RedisUtil redisUtil; // private RedisUtil redisUtil;
@Autowired // @Autowired
private KylinUtils kylinUtils; // private KylinUtils kylinUtils;
//
@Override // @Override
public void onMessage(MapRecord<String, String, String> message) { // public void onMessage(MapRecord<String, String, String> message) {
String redisStreamKey = this.getRedisStreamKey(); // String redisStreamKey = this.getRedisStreamKey();
log.debug("CONSUMER MSG[streamKey:{},messageId:{},stream:{},body:{}]", redisStreamKey, message.getId(), message.getStream(), message.getValue()); // log.debug("CONSUMER MSG[streamKey:{},messageId:{},stream:{},body:{}]", redisStreamKey, message.getId(), message.getStream(), message.getValue());
boolean result = this.consumerMessageHandler(message.getValue().get("message")); // boolean result = this.consumerMessageHandler(message.getValue().get("message"));
log.info("CONSUMER MSG RESULT:{} ==> [{}]MESSAGE_ID:{}", result, redisStreamKey, message.getId()); // log.info("CONSUMER MSG RESULT:{} ==> [{}]MESSAGE_ID:{}", result, redisStreamKey, message.getId());
//
try { // try {
stringRedisTemplate.opsForStream().acknowledge(getRedisStreamGroup(), message); // stringRedisTemplate.opsForStream().acknowledge(getRedisStreamGroup(), message);
} catch (Exception e) { // } catch (Exception e) {
log.error("#CONSUMER MSG EX_ACK ==> [{}]RESULT:{},MESSAGE:{}", redisStreamKey, result, message.getValue(), e); // log.error("#CONSUMER MSG EX_ACK ==> [{}]RESULT:{},MESSAGE:{}", redisStreamKey, result, message.getValue(), e);
} // }
try { // try {
stringRedisTemplate.opsForStream().delete(redisStreamKey, message.getId()); // stringRedisTemplate.opsForStream().delete(redisStreamKey, message.getId());
} catch (Exception e) { // } catch (Exception e) {
log.error("#CONSUMER MSG EX_DEL ==> [{}]RESULT:{},MESSAGE:{}", redisStreamKey, result, message.getValue(), e); // log.error("#CONSUMER MSG EX_DEL ==> [{}]RESULT:{},MESSAGE:{}", redisStreamKey, result, message.getValue(), e);
} // }
} // }
//
private boolean consumerMessageHandler(String msg) { // private boolean consumerMessageHandler(String msg) {
boolean aBoolean = false; // boolean aBoolean = false;
try { // try {
OrderCloseMapping.orderCloseMessage mqMessage = JsonUtils.fromJson(msg, OrderCloseMapping.orderCloseMessage.class); // OrderCloseMapping.orderCloseMessage mqMessage = JsonUtils.fromJson(msg, OrderCloseMapping.orderCloseMessage.class);
if (mqMessage == null) { // if (mqMessage == null) {
//
} else { // } else {
for (int x = 0; x < mqMessage.getOrderTicketIds().size(); x++) { // for (int x = 0; x < mqMessage.getOrderTicketIds().size(); x++) {
String orderTicketId = mqMessage.getOrderTicketIds().get(x); // String orderTicketId = mqMessage.getOrderTicketIds().get(x);
KylinOrderTicketVo vo = kylinUtils.getOrderTicketVo(orderTicketId); // KylinOrderTicketVo vo = kylinUtils.getOrderTicketVo(orderTicketId);
String uid = vo.getTransferUid(); // String uid = vo.getTransferUid();
vo.setTransferUid(""); // vo.setTransferUid("");
vo.setTransferStatus(0); // vo.setTransferStatus(0);
redisUtil.del(KylinRedisConst.ORDER_TRANSFER.concat(uid)); // redisUtil.del(KylinRedisConst.ORDER_TRANSFER.concat(uid));
redisUtil.set("kylin:order:id:" + orderTicketId, vo); // redisUtil.set("kylin:order:id:" + orderTicketId, vo);
kylinUtils.resetOrderListVo(vo.getUserId(), 2, orderTicketId, vo); // kylinUtils.resetOrderListVo(vo.getUserId(), 2, orderTicketId, vo);
} // }
} // }
aBoolean = true; // aBoolean = true;
} catch (Exception e) { // } catch (Exception e) {
log.error("CONSUMER MSG EX_HANDLE ==> [{}]:{}", this.getRedisStreamKey(), msg, e); // log.error("CONSUMER MSG EX_HANDLE ==> [{}]:{}", this.getRedisStreamKey(), msg, e);
} finally { // } finally {
if (!aBoolean) { // if (!aBoolean) {
HashMap<String, String> map = CollectionUtil.mapStringString(); // HashMap<String, String> map = CollectionUtil.mapStringString();
map.put("message", msg); // map.put("message", msg);
stringRedisTemplate.opsForStream().add(StreamRecords.mapBacked(map).withStreamKey(this.getRedisStreamKey())); // stringRedisTemplate.opsForStream().add(StreamRecords.mapBacked(map).withStreamKey(this.getRedisStreamKey()));
} // }
} // }
return aBoolean; // return aBoolean;
} // }
//
protected abstract String getRedisStreamKey(); // protected abstract String getRedisStreamKey();
//
protected abstract String getRedisStreamGroup(); // protected abstract String getRedisStreamGroup();
} //}
liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-slowly/src/main/java/com/liquidnet/service/consumer/slowly/receiver/AbstractXlsRedisReceiver.java
View file @ 98c887bf
package com.liquidnet.service.consumer.slowly.receiver; //package com.liquidnet.service.consumer.slowly.receiver;
//
import com.alibaba.excel.EasyExcel; //import com.alibaba.excel.EasyExcel;
import com.alibaba.excel.read.listener.PageReadListener; //import com.alibaba.excel.read.listener.PageReadListener;
import com.alibaba.fastjson.JSON; //import com.alibaba.fastjson.JSON;
import com.liquidnet.common.cache.redis.util.RedisUtil; //import com.liquidnet.common.cache.redis.util.RedisUtil;
import com.liquidnet.commons.lang.util.CollectionUtil; //import com.liquidnet.commons.lang.util.CollectionUtil;
import com.liquidnet.service.consumer.slowly.dto.PhoneDto; //import com.liquidnet.service.consumer.slowly.dto.PhoneDto;
import com.liquidnet.service.consumer.slowly.service.IBaseDao; //import com.liquidnet.service.consumer.slowly.service.IBaseDao;
import com.liquidnet.service.goblin.constant.GoblinRedisConst; //import com.liquidnet.service.goblin.constant.GoblinRedisConst;
import lombok.extern.slf4j.Slf4j; //import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired; //import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.stream.MapRecord; //import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.core.StringRedisTemplate; //import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.stream.StreamListener; //import org.springframework.data.redis.stream.StreamListener;
//
import java.net.URL; //import java.net.URL;
import java.time.LocalDateTime; //import java.time.LocalDateTime;
import java.util.HashMap; //import java.util.HashMap;
import java.util.LinkedList; //import java.util.LinkedList;
import java.util.Map; //import java.util.Map;
//
@Slf4j //@Slf4j
public abstract class AbstractXlsRedisReceiver implements StreamListener<String, MapRecord<String, String, String>> { //public abstract class AbstractXlsRedisReceiver implements StreamListener<String, MapRecord<String, String, String>> {
@Autowired // @Autowired
private IBaseDao baseDao; // private IBaseDao baseDao;
@Autowired // @Autowired
StringRedisTemplate stringRedisTemplate; // StringRedisTemplate stringRedisTemplate;
@Autowired // @Autowired
private RedisUtil redisUtil; // private RedisUtil redisUtil;
//
private static final String SQL_INSERT_GOODS_BUY_ROSTER_LOG = "INSERT INTO goblin_goods_buy_roster_log (sku_id,buy_roster,buy_roster_type,parsing_result,created_at)VALUES(?,?,?,?,?)"; // private static final String SQL_INSERT_GOODS_BUY_ROSTER_LOG = "INSERT INTO goblin_goods_buy_roster_log (sku_id,buy_roster,buy_roster_type,parsing_result,created_at)VALUES(?,?,?,?,?)";
//
@Override // @Override
public void onMessage(MapRecord<String, String, String> message) { // public void onMessage(MapRecord<String, String, String> message) {
String redisStreamKey = this.getRedisStreamKey(); // String redisStreamKey = this.getRedisStreamKey();
log.debug("CONSUMER MSG[streamKey:{},messageId:{},stream:{},body:{}]", redisStreamKey, message.getId(), message.getStream(), message.getValue()); // log.debug("CONSUMER MSG[streamKey:{},messageId:{},stream:{},body:{}]", redisStreamKey, message.getId(), message.getStream(), message.getValue());
boolean result = this.consumerMessageHandler(message.getValue()); // boolean result = this.consumerMessageHandler(message.getValue());
log.info("XLS MESSAGE CONSUMER MSG RESULT:{} ==> [{}]MESSAGE_ID:{}", result, redisStreamKey, message.getId()); // log.info("XLS MESSAGE CONSUMER MSG RESULT:{} ==> [{}]MESSAGE_ID:{}", result, redisStreamKey, message.getId());
//
try { // try {
stringRedisTemplate.opsForStream().acknowledge(getRedisStreamGroup(), message); // stringRedisTemplate.opsForStream().acknowledge(getRedisStreamGroup(), message);
} catch (Exception e) { // } catch (Exception e) {
log.error("#CONSUMER MSG EX_ACK ==> [{}]RESULT:{},MESSAGE:{}", redisStreamKey, result, message.getValue(), e); // log.error("#CONSUMER MSG EX_ACK ==> [{}]RESULT:{},MESSAGE:{}", redisStreamKey, result, message.getValue(), e);
} // }
try { // try {
stringRedisTemplate.opsForStream().delete(redisStreamKey, message.getId()); // stringRedisTemplate.opsForStream().delete(redisStreamKey, message.getId());
} catch (Exception e) { // } catch (Exception e) {
log.error("#CONSUMER MSG EX_DEL ==> [{}]RESULT:{},MESSAGE:{}", redisStreamKey, result, message.getValue(), e); // log.error("#CONSUMER MSG EX_DEL ==> [{}]RESULT:{},MESSAGE:{}", redisStreamKey, result, message.getValue(), e);
} // }
} // }
//
private boolean consumerMessageHandler(Map<String, String> message) { // private boolean consumerMessageHandler(Map<String, String> message) {
LinkedList<Object[]> objs = CollectionUtil.linkedListObjectArr(); // LinkedList<Object[]> objs = CollectionUtil.linkedListObjectArr();
String xlsPath = null, skuId = null; // String xlsPath = null, skuId = null;
String oXlsPath = null; // String oXlsPath = null;
Integer type = null; // Integer type = null;
boolean aBoolean = false; // boolean aBoolean = false;
try { // try {
String[] path = message.get("message").split(","); // String[] path = message.get("message").split(",");
if (path.length == 0) { // if (path.length == 0) {
xlsPath = ""; // xlsPath = "";
} else { // } else {
xlsPath = path[0]; // xlsPath = path[0];
} // }
if (path.length > 1) { // if (path.length > 1) {
oXlsPath = path[1]; // oXlsPath = path[1];
} // }
String finalSkuId = (skuId = message.get("skuId")); // String finalSkuId = (skuId = message.get("skuId"));
String listId; // String listId;
//// try {
// listId = message.getOrDefault("listId", "");
//// }catch (Exception e){
//// listId = "";
//// }
// Integer finalType = (type = Integer.parseInt(message.get("type")));
// if (finalType.equals(1) || finalType.equals(2)) {
// EasyExcel.read(new URL(xlsPath).openStream(), PhoneDto.class, new PageReadListener<PhoneDto>(dataList -> {
// for (PhoneDto data : dataList) {
// if (data.getMobile() == null) {
// continue;
// }
// String redisKey = GoblinRedisConst.REDIS_CAN_BUY.concat(finalSkuId + ":").concat(data.getMobile());
// if (finalType.equals(1)) {//添加
// if (log.isDebugEnabled()) {
// log.debug("添加 读取到一条数据{}", JSON.toJSONString(data));
// }
// redisUtil.set(redisKey, 0);
// } else {
// if (log.isDebugEnabled()) {
// log.debug("删除 读取到一条数据{}", JSON.toJSONString(data));
// }
// redisUtil.del(redisKey);
// }
// }
// })).sheet().doRead();
// objs.add(new Object[]{skuId, xlsPath, type, 1, LocalDateTime.now()});
// aBoolean = baseDao.batchSql(SQL_INSERT_GOODS_BUY_ROSTER_LOG, objs);
// } else if (finalType.equals(3) || finalType.equals(4)) {
// if (oXlsPath != null && !oXlsPath.equals("")) {//删除旧的黑白名单
// EasyExcel.read(new URL(oXlsPath).openStream(), new PageReadListener<HashMap<String, String>>(dataList -> {
// for (HashMap<String, String> data : dataList) {
// if (data.get(0) == null) {
// continue;
// }
// if (finalType.equals(3)) {
// log.debug("删除 白名单 读取到一条数据{}", JSON.toJSONString(data));
// redisUtil.del(GoblinRedisConst.REDIS_WHITE.concat(listId + ":").concat(finalSkuId + ":").concat(data.get(0)));
// } else {
// log.debug("删除 黑名单 读取到一条数据{}", JSON.toJSONString(data));
// redisUtil.del(GoblinRedisConst.REDIS_BLACK.concat(listId + ":").concat(finalSkuId + ":").concat(data.get(0)));
// }
// }
// })).sheet().doRead();
// }
// if (xlsPath != null && !xlsPath.equals("")) {//添加
// EasyExcel.read(new URL(xlsPath).openStream(), new PageReadListener<HashMap<String, String>>(dataList -> {
// for (HashMap<String, String> data : dataList) {
// if (data.get(0) == null) {
// continue;
// }
// if (finalType.equals(3)) {
// log.debug("添加 白名单 读取到一条数据{}", JSON.toJSONString(data));
// redisUtil.set(GoblinRedisConst.REDIS_WHITE.concat(listId + ":").concat(finalSkuId + ":").concat(data.get(0)), 1);
// } else {
// log.debug("添加 黑名单 读取到一条数据{}", JSON.toJSONString(data));
// redisUtil.set(GoblinRedisConst.REDIS_BLACK.concat(listId + ":").concat(finalSkuId + ":").concat(data.get(0)), 1);
// }
// }
// })).sheet().doRead();
// }
// aBoolean = true;
// }
// } catch (Exception e) {
// log.error("CONSUMER MSG EX_HANDLE ==> [{}]:{}", this.getRedisStreamKey(), message, e);
// try { // try {
listId = message.getOrDefault("listId", ""); // if (type.equals(1) || type.equals(2)) {
// }catch (Exception e){ // objs.add(new Object[]{skuId, xlsPath, type, 2, LocalDateTime.now()});
// listId = ""; // aBoolean = baseDao.batchSql(SQL_INSERT_GOODS_BUY_ROSTER_LOG, objs);
// }
// } catch (Exception ignored) {
// } // }
Integer finalType = (type = Integer.parseInt(message.get("type"))); // }
if (finalType.equals(1) || finalType.equals(2)) { // return aBoolean;
EasyExcel.read(new URL(xlsPath).openStream(), PhoneDto.class, new PageReadListener<PhoneDto>(dataList -> { // }
for (PhoneDto data : dataList) { //
if (data.getMobile() == null) { // protected abstract String getRedisStreamKey();
continue; //
} // protected abstract String getRedisStreamGroup();
String redisKey = GoblinRedisConst.REDIS_CAN_BUY.concat(finalSkuId + ":").concat(data.getMobile()); //}
if (finalType.equals(1)) {//添加 \ No newline at end of file
if (log.isDebugEnabled()) {
log.debug("添加 读取到一条数据{}", JSON.toJSONString(data));
}
redisUtil.set(redisKey, 0);
} else {
if (log.isDebugEnabled()) {
log.debug("删除 读取到一条数据{}", JSON.toJSONString(data));
}
redisUtil.del(redisKey);
}
}
})).sheet().doRead();
objs.add(new Object[]{skuId, xlsPath, type, 1, LocalDateTime.now()});
aBoolean = baseDao.batchSql(SQL_INSERT_GOODS_BUY_ROSTER_LOG, objs);
} else if (finalType.equals(3) || finalType.equals(4)) {
if (oXlsPath != null && !oXlsPath.equals("")) {//删除旧的黑白名单
EasyExcel.read(new URL(oXlsPath).openStream(), new PageReadListener<HashMap<String, String>>(dataList -> {
for (HashMap<String, String> data : dataList) {
if (data.get(0) == null) {
continue;
}
if (finalType.equals(3)) {
log.debug("删除 白名单 读取到一条数据{}", JSON.toJSONString(data));
redisUtil.del(GoblinRedisConst.REDIS_WHITE.concat(listId + ":").concat(finalSkuId + ":").concat(data.get(0)));
} else {
log.debug("删除 黑名单 读取到一条数据{}", JSON.toJSONString(data));
redisUtil.del(GoblinRedisConst.REDIS_BLACK.concat(listId + ":").concat(finalSkuId + ":").concat(data.get(0)));
}
}
})).sheet().doRead();
}
if (xlsPath != null && !xlsPath.equals("")) {//添加
EasyExcel.read(new URL(xlsPath).openStream(), new PageReadListener<HashMap<String, String>>(dataList -> {
for (HashMap<String, String> data : dataList) {
if (data.get(0) == null) {
continue;
}
if (finalType.equals(3)) {
log.debug("添加 白名单 读取到一条数据{}", JSON.toJSONString(data));
redisUtil.set(GoblinRedisConst.REDIS_WHITE.concat(listId + ":").concat(finalSkuId + ":").concat(data.get(0)), 1);
} else {
log.debug("添加 黑名单 读取到一条数据{}", JSON.toJSONString(data));
redisUtil.set(GoblinRedisConst.REDIS_BLACK.concat(listId + ":").concat(finalSkuId + ":").concat(data.get(0)), 1);
}
}
})).sheet().doRead();
}
aBoolean = true;
}
} catch (Exception e) {
log.error("CONSUMER MSG EX_HANDLE ==> [{}]:{}", this.getRedisStreamKey(), message, e);
try {
if (type.equals(1) || type.equals(2)) {
objs.add(new Object[]{skuId, xlsPath, type, 2, LocalDateTime.now()});
aBoolean = baseDao.batchSql(SQL_INSERT_GOODS_BUY_ROSTER_LOG, objs);
}
} catch (Exception ignored) {
}
}
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-slowly/src/main/java/com/liquidnet/service/consumer/slowly/receiver/ConsumerGoblinXlsRdsReceiver.java
View file @ 98c887bf
package com.liquidnet.service.consumer.slowly.receiver; //package com.liquidnet.service.consumer.slowly.receiver;
//
import com.liquidnet.service.base.constant.MQConst; //import com.liquidnet.service.base.constant.MQConst;
import org.springframework.stereotype.Component; //import org.springframework.stereotype.Component;
//
@Component //@Component
public class ConsumerGoblinXlsRdsReceiver extends AbstractXlsRedisReceiver { //public class ConsumerGoblinXlsRdsReceiver extends AbstractXlsRedisReceiver {
@Override // @Override
protected String getRedisStreamKey() { // protected String getRedisStreamKey() {
return MQConst.GoblinQueue.GOBLIN_XLS_OPERA.getKey(); // return MQConst.GoblinQueue.GOBLIN_XLS_OPERA.getKey();
} // }
//
@Override // @Override
protected String getRedisStreamGroup() { // protected String getRedisStreamGroup() {
return MQConst.GoblinQueue.GOBLIN_XLS_OPERA.getGroup(); // return MQConst.GoblinQueue.GOBLIN_XLS_OPERA.getGroup();
} // }
} //}
liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-slowly/src/main/java/com/liquidnet/service/consumer/slowly/receiver/ConsumerKylinTransferOverTimeRdsReceiver.java
View file @ 98c887bf
package com.liquidnet.service.consumer.slowly.receiver; //package com.liquidnet.service.consumer.slowly.receiver;
//
import com.liquidnet.service.base.constant.MQConst; //import com.liquidnet.service.base.constant.MQConst;
import org.springframework.stereotype.Component; //import org.springframework.stereotype.Component;
//
@Component //@Component
public class ConsumerKylinTransferOverTimeRdsReceiver extends AbstractOptOrderTransferOverTimeRedisReceiver { //public class ConsumerKylinTransferOverTimeRdsReceiver extends AbstractOptOrderTransferOverTimeRedisReceiver {
@Override // @Override
protected String getRedisStreamKey() { // protected String getRedisStreamKey() {
return MQConst.KylinQueue.SQL_TRANSFER_OVERTIME.getKey(); // return MQConst.KylinQueue.SQL_TRANSFER_OVERTIME.getKey();
} // }
//
@Override // @Override
protected String getRedisStreamGroup() { // protected String getRedisStreamGroup() {
return MQConst.KylinQueue.SQL_TRANSFER_OVERTIME.getGroup(); // return MQConst.KylinQueue.SQL_TRANSFER_OVERTIME.getGroup();
} // }
} //}
liquidnet-bus-service/liquidnet-service-kylin/liquidnet-service-kylin-impl/src/main/java/com/liquidnet/service/kylin/controller/InnerController.java
View file @ 98c887bf
...@@ -14,10 +14,7 @@ import lombok.extern.slf4j.Slf4j; ...@@ -14,10 +14,7 @@ import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value; import org.springframework.beans.factory.annotation.Value;
import org.springframework.util.MultiValueMap; import org.springframework.util.MultiValueMap;
import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.*;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import javax.validation.constraints.NotBlank; import javax.validation.constraints.NotBlank;
import java.time.LocalDateTime; import java.time.LocalDateTime;
...@@ -66,7 +63,7 @@ public class InnerController { ...@@ -66,7 +63,7 @@ public class InnerController {
} }
} }
@GetMapping("consumer/orderTransferOverTime") @PostMapping("consumer/orderTransferOverTime")
@ApiOperation("转赠过期消费") @ApiOperation("转赠过期消费")
public ResponseDto<Boolean> orderTransferOverTime(String orderId) { public ResponseDto<Boolean> orderTransferOverTime(String orderId) {
return kylinConsmerService.orderTransferOverTime(orderId); return kylinConsmerService.orderTransferOverTime(orderId);
......
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