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

Skip to content

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

消费迁移 xls相关 转赠订单定时过期 批量空投二维码 提交

parent 98c887bf
Hide whitespace changes
Inline Side-by-side
Showing with 472 additions and 303 deletions
liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-base/src/main/java/com/liquidnet/service/consumer/base/config/ConsumerGoblinPhoneCodeStreamConfig.java 0 → 100644
View file @ be62c0a2
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.ConsumerGoblinPhoneCodeReceiver;
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 ConsumerGoblinPhoneCodeStreamConfig extends RedisStreamConfig {
@Autowired
StringRedisTemplate stringRedisTemplate;
@Autowired
ConsumerGoblinPhoneCodeReceiver consumerGoblinPhoneCodeReceiver;
//批量空投
@Bean
public List<Subscription> subscriptionPhoneCodeExamine(RedisConnectionFactory factory) {
List<Subscription> subscriptionList = new ArrayList<>();
MQConst.GoblinQueue stream = MQConst.GoblinQueue.GOBLIN_PHONE_CODE_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()), consumerGoblinPhoneCodeReceiver
));
listenerContainer.start();
}
return subscriptionList;
}
}
liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-base/src/main/java/com/liquidnet/service/consumer/base/receiver/AbstractHttpRedisReceiver.java 0 → 100644
View file @ be62c0a2
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.ResponseDto;
import com.liquidnet.service.base.SqlMapping;
import com.liquidnet.service.base.constant.MQConst;
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.*;
@Slf4j
public abstract class AbstractHttpRedisReceiver implements StreamListener<String, MapRecord<String, String, String>> {
@Autowired
StringRedisTemplate stringRedisTemplate;
@Value("${liquidnet.service.order.url}")
private String orderUrl;
@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("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) {
try {
String codestr = message.get("codes");
String userIds = message.get("userIds");
String phonestr = message.get("phones");
String adminUid = message.get("adminUid");
List<String> codeList = Arrays.asList(codestr.split(","));
List<String> userIdList = Arrays.asList(userIds.split(","));
List<String> phoneList = Arrays.asList(phonestr.split(","));
LinkedList<Object[]> addLink = CollectionUtil.linkedListObjectArr();
for (int i = 0; i < userIdList.size(); i++) {
MultiValueMap<String, String> params = new LinkedMultiValueMap();
params.add("code", codeList.get(i));
params.add("userId", userIdList.get(i));
MultiValueMap<String, String> headers = CollectionUtil.linkedMultiValueMapStringString();
headers.add("Accept", "application/json;charset=UTF-8");
String post = HttpUtil.post(orderUrl + "/order/goblin/nft/airdrop", params, headers);
ResponseDto<Boolean> rsp = JsonUtils.fromJson(post, new TypeReference<ResponseDto<Boolean>>() {
});
if (rsp.isSuccess()) {
addLink.add(new Object[]{adminUid, codeList.get(i)});
} else {
log.error("airdrop send fail:[{}]" + phoneList.get(i));
}
}
if (addLink != null && !addLink.isEmpty()) {
LinkedList<String> sqls = CollectionUtil.linkedListString();
sqls.add(SqlMapping.get("goblin_activity.code.admin"));
String sqlData = SqlMapping.gets(sqls, addLink);
sendMsgByRedis(MQConst.GoblinQueue.GOBLIN_CODE_OPERA.getKey(), sqlData);
}
} catch (Exception e) {
log.error("airdrop send fail:[{}]");
}
return true;
}
public void sendMsgByRedis(String streamKey, String jsonMsg) {
HashMap<String, String> map = CollectionUtil.mapStringString();
map.put(MQConst.QUEUE_MESSAGE_KEY, jsonMsg);
stringRedisTemplate.opsForStream().add(StreamRecords.mapBacked(map).withStreamKey(streamKey));
}
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/ConsumerGoblinPhoneCodeReceiver.java 0 → 100644
View file @ be62c0a2
package com.liquidnet.service.consumer.base.receiver;
import com.liquidnet.service.base.constant.MQConst;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
@Slf4j
@Component
public class ConsumerGoblinPhoneCodeReceiver extends AbstractHttpRedisReceiver {
@Override
protected String getRedisStreamKey() {
return MQConst.GoblinQueue.GOBLIN_PHONE_CODE_OPERA.getKey();
}
@Override
protected String getRedisStreamGroup() {
return MQConst.GoblinQueue.GOBLIN_PHONE_CODE_OPERA.getGroup();
}
}
liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-kylin/src/main/java/com/liquidnet/service/consumer/kylin/receiver/AbstractHttpRedisReceiver.java
View file @ be62c0a2
package com.liquidnet.service.consumer.kylin.receiver; //package com.liquidnet.service.consumer.kylin.receiver;
//
import com.fasterxml.jackson.core.type.TypeReference; //import com.fasterxml.jackson.core.type.TypeReference;
import com.liquidnet.commons.lang.util.CollectionUtil; //import com.liquidnet.commons.lang.util.CollectionUtil;
import com.liquidnet.commons.lang.util.HttpUtil; //import com.liquidnet.commons.lang.util.HttpUtil;
import com.liquidnet.commons.lang.util.JsonUtils; //import com.liquidnet.commons.lang.util.JsonUtils;
import com.liquidnet.service.base.ResponseDto; //import com.liquidnet.service.base.ResponseDto;
import com.liquidnet.service.base.SqlMapping; //import com.liquidnet.service.base.SqlMapping;
import com.liquidnet.service.base.constant.MQConst; //import com.liquidnet.service.base.constant.MQConst;
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.beans.factory.annotation.Value; //import org.springframework.beans.factory.annotation.Value;
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 org.springframework.util.LinkedMultiValueMap; //import org.springframework.util.LinkedMultiValueMap;
import org.springframework.util.MultiValueMap; //import org.springframework.util.MultiValueMap;
//
import java.util.*; //import java.util.*;
//
@Slf4j //@Slf4j
public abstract class AbstractHttpRedisReceiver implements StreamListener<String, MapRecord<String, String, String>> { //public abstract class AbstractHttpRedisReceiver implements StreamListener<String, MapRecord<String, String, String>> {
//
@Autowired // @Autowired
StringRedisTemplate stringRedisTemplate; // StringRedisTemplate stringRedisTemplate;
//
@Value("${liquidnet.service.order.url}") // @Value("${liquidnet.service.order.url}")
private String orderUrl; // private String orderUrl;
//
@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("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(Map<String, String> message) { // private boolean consumerMessageHandler(Map<String, String> message) {
try { // try {
String codestr = message.get("codes"); // String codestr = message.get("codes");
String userIds = message.get("userIds"); // String userIds = message.get("userIds");
String phonestr = message.get("phones"); // String phonestr = message.get("phones");
String adminUid = message.get("adminUid"); // String adminUid = message.get("adminUid");
List<String> codeList = Arrays.asList(codestr.split(",")); // List<String> codeList = Arrays.asList(codestr.split(","));
List<String> userIdList = Arrays.asList(userIds.split(",")); // List<String> userIdList = Arrays.asList(userIds.split(","));
List<String> phoneList = Arrays.asList(phonestr.split(",")); // List<String> phoneList = Arrays.asList(phonestr.split(","));
LinkedList<Object[]> addLink = CollectionUtil.linkedListObjectArr(); // LinkedList<Object[]> addLink = CollectionUtil.linkedListObjectArr();
for (int i = 0; i < userIdList.size(); i++) { // for (int i = 0; i < userIdList.size(); i++) {
MultiValueMap<String, String> params = new LinkedMultiValueMap(); // MultiValueMap<String, String> params = new LinkedMultiValueMap();
params.add("code", codeList.get(i)); // params.add("code", codeList.get(i));
params.add("userId", userIdList.get(i)); // params.add("userId", userIdList.get(i));
MultiValueMap<String, String> headers = CollectionUtil.linkedMultiValueMapStringString(); // MultiValueMap<String, String> headers = CollectionUtil.linkedMultiValueMapStringString();
headers.add("Accept", "application/json;charset=UTF-8"); // headers.add("Accept", "application/json;charset=UTF-8");
String post = HttpUtil.post(orderUrl + "/order/goblin/nft/airdrop", params, headers); // String post = HttpUtil.post(orderUrl + "/order/goblin/nft/airdrop", params, headers);
ResponseDto<Boolean> rsp = JsonUtils.fromJson(post, new TypeReference<ResponseDto<Boolean>>() { // ResponseDto<Boolean> rsp = JsonUtils.fromJson(post, new TypeReference<ResponseDto<Boolean>>() {
}); // });
if (rsp.isSuccess()) { // if (rsp.isSuccess()) {
addLink.add(new Object[]{adminUid, codeList.get(i)}); // addLink.add(new Object[]{adminUid, codeList.get(i)});
} else { // } else {
log.error("airdrop send fail:[{}]" + phoneList.get(i)); // log.error("airdrop send fail:[{}]" + phoneList.get(i));
} // }
} // }
//
if (addLink != null && !addLink.isEmpty()) { // if (addLink != null && !addLink.isEmpty()) {
LinkedList<String> sqls = CollectionUtil.linkedListString(); // LinkedList<String> sqls = CollectionUtil.linkedListString();
sqls.add(SqlMapping.get("goblin_activity.code.admin")); // sqls.add(SqlMapping.get("goblin_activity.code.admin"));
String sqlData = SqlMapping.gets(sqls, addLink); // String sqlData = SqlMapping.gets(sqls, addLink);
sendMsgByRedis(MQConst.GoblinQueue.GOBLIN_CODE_OPERA.getKey(), sqlData); // sendMsgByRedis(MQConst.GoblinQueue.GOBLIN_CODE_OPERA.getKey(), sqlData);
} // }
//
} catch (Exception e) { // } catch (Exception e) {
log.error("airdrop send fail:[{}]"); // log.error("airdrop send fail:[{}]");
} // }
//
return true; // return true;
} // }
//
//
public void sendMsgByRedis(String streamKey, String jsonMsg) { // public void sendMsgByRedis(String streamKey, String jsonMsg) {
HashMap<String, String> map = CollectionUtil.mapStringString(); // HashMap<String, String> map = CollectionUtil.mapStringString();
map.put(MQConst.QUEUE_MESSAGE_KEY, jsonMsg); // map.put(MQConst.QUEUE_MESSAGE_KEY, jsonMsg);
stringRedisTemplate.opsForStream().add(StreamRecords.mapBacked(map).withStreamKey(streamKey)); // stringRedisTemplate.opsForStream().add(StreamRecords.mapBacked(map).withStreamKey(streamKey));
} // }
//
//
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/ConsumerGoblinPhoneCodeReceiver.java
View file @ be62c0a2
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 lombok.extern.slf4j.Slf4j; //import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component; //import org.springframework.stereotype.Component;
//
@Slf4j //@Slf4j
@Component //@Component
public class ConsumerGoblinPhoneCodeReceiver extends AbstractHttpRedisReceiver { //public class ConsumerGoblinPhoneCodeReceiver extends AbstractHttpRedisReceiver {
//
@Override // @Override
protected String getRedisStreamKey() { // protected String getRedisStreamKey() {
return MQConst.GoblinQueue.GOBLIN_PHONE_CODE_OPERA.getKey(); // return MQConst.GoblinQueue.GOBLIN_PHONE_CODE_OPERA.getKey();
} // }
//
@Override // @Override
protected String getRedisStreamGroup() { // protected String getRedisStreamGroup() {
return MQConst.GoblinQueue.GOBLIN_PHONE_CODE_OPERA.getGroup(); // return MQConst.GoblinQueue.GOBLIN_PHONE_CODE_OPERA.getGroup();
} // }
//
} //}
liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-slowly/src/main/java/com/liquidnet/service/consumer/slowly/config/ConsumerGoblinPhoneCodeStreamConfig.java
View file @ be62c0a2
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.ConsumerGoblinPhoneCodeReceiver; //import com.liquidnet.service.consumer.slowly.receiver.ConsumerGoblinPhoneCodeReceiver;
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.core.StringRedisTemplate; //import org.springframework.data.redis.core.StringRedisTemplate;
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 java.util.ArrayList; //import java.util.ArrayList;
import java.util.List; //import java.util.List;
@Configuration
public class ConsumerGoblinPhoneCodeStreamConfig extends RedisStreamConfig {
@Autowired
StringRedisTemplate stringRedisTemplate;
@Autowired
ConsumerGoblinPhoneCodeReceiver consumerGoblinPhoneCodeReceiver;
// public Subscription receivePhoneCodeExamine(StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer, int t) {
// return listenerContainer.receiveAutoAck(Consumer.from(MQConst.GoblinQueue.GOBLIN_PHONE_CODE_OPERA.getGroup(), getConsumerName(MQConst.GoblinQueue.GOBLIN_PHONE_CODE_OPERA.name() + t)),
// StreamOffset.create(MQConst.GoblinQueue.GOBLIN_PHONE_CODE_OPERA.getKey(), ReadOffset.lastConsumed()), consumerGoblinPhoneCodeReceiver);
// }
// //
// @Bean //@Configuration
// public Subscription phoneCodeExamine0(RedisConnectionFactory factory) { //public class ConsumerGoblinPhoneCodeStreamConfig extends RedisStreamConfig {
// var listenerContainer = this.buildStreamMessageListenerContainer(factory); // @Autowired
// var subscription = receivePhoneCodeExamine(listenerContainer, 0); // StringRedisTemplate stringRedisTemplate;
// listenerContainer.start(); // @Autowired
// return subscription; // ConsumerGoblinPhoneCodeReceiver consumerGoblinPhoneCodeReceiver;
//
//// public Subscription receivePhoneCodeExamine(StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer, int t) {
//// return listenerContainer.receiveAutoAck(Consumer.from(MQConst.GoblinQueue.GOBLIN_PHONE_CODE_OPERA.getGroup(), getConsumerName(MQConst.GoblinQueue.GOBLIN_PHONE_CODE_OPERA.name() + t)),
//// StreamOffset.create(MQConst.GoblinQueue.GOBLIN_PHONE_CODE_OPERA.getKey(), ReadOffset.lastConsumed()), consumerGoblinPhoneCodeReceiver);
//// }
////
//// @Bean
//// public Subscription phoneCodeExamine0(RedisConnectionFactory factory) {
//// var listenerContainer = this.buildStreamMessageListenerContainer(factory);
//// var subscription = receivePhoneCodeExamine(listenerContainer, 0);
//// listenerContainer.start();
//// return subscription;
//// }
//
// @Bean//
// public List<Subscription> subscriptionPhoneCodeExamine(RedisConnectionFactory factory) {
// List<Subscription> subscriptionList = new ArrayList<>();
// MQConst.GoblinQueue stream = MQConst.GoblinQueue.GOBLIN_PHONE_CODE_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()), consumerGoblinPhoneCodeReceiver
// ));
// listenerContainer.start();
// }
// return subscriptionList;
// } // }
//
@Bean// //}
public List<Subscription> subscriptionPhoneCodeExamine(RedisConnectionFactory factory) {
List<Subscription> subscriptionList = new ArrayList<>();
MQConst.GoblinQueue stream = MQConst.GoblinQueue.GOBLIN_PHONE_CODE_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()), consumerGoblinPhoneCodeReceiver
));
listenerContainer.start();
}
return subscriptionList;
}
}
liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-slowly/src/main/java/com/liquidnet/service/consumer/slowly/receiver/AbstractHttpRedisReceiver.java
View file @ be62c0a2
package com.liquidnet.service.consumer.slowly.receiver; //package com.liquidnet.service.consumer.slowly.receiver;
//
import com.fasterxml.jackson.core.type.TypeReference; //import com.fasterxml.jackson.core.type.TypeReference;
import com.liquidnet.commons.lang.util.CollectionUtil; //import com.liquidnet.commons.lang.util.CollectionUtil;
import com.liquidnet.commons.lang.util.HttpUtil; //import com.liquidnet.commons.lang.util.HttpUtil;
import com.liquidnet.commons.lang.util.JsonUtils; //import com.liquidnet.commons.lang.util.JsonUtils;
import com.liquidnet.service.base.ResponseDto; //import com.liquidnet.service.base.ResponseDto;
import com.liquidnet.service.base.SqlMapping; //import com.liquidnet.service.base.SqlMapping;
import com.liquidnet.service.base.constant.MQConst; //import com.liquidnet.service.base.constant.MQConst;
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.beans.factory.annotation.Value; //import org.springframework.beans.factory.annotation.Value;
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 org.springframework.util.LinkedMultiValueMap; //import org.springframework.util.LinkedMultiValueMap;
import org.springframework.util.MultiValueMap; //import org.springframework.util.MultiValueMap;
//
import java.util.*; //import java.util.*;
//
@Slf4j //@Slf4j
public abstract class AbstractHttpRedisReceiver implements StreamListener<String, MapRecord<String, String, String>> { //public abstract class AbstractHttpRedisReceiver implements StreamListener<String, MapRecord<String, String, String>> {
//
@Autowired // @Autowired
StringRedisTemplate stringRedisTemplate; // StringRedisTemplate stringRedisTemplate;
//
@Value("${liquidnet.service.order.url}") // @Value("${liquidnet.service.order.url}")
private String orderUrl; // private String orderUrl;
//
@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("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(Map<String, String> message) { // private boolean consumerMessageHandler(Map<String, String> message) {
try { // try {
String codestr = message.get("codes"); // String codestr = message.get("codes");
String userIds = message.get("userIds"); // String userIds = message.get("userIds");
String phonestr = message.get("phones"); // String phonestr = message.get("phones");
String adminUid = message.get("adminUid"); // String adminUid = message.get("adminUid");
List<String> codeList = Arrays.asList(codestr.split(",")); // List<String> codeList = Arrays.asList(codestr.split(","));
List<String> userIdList = Arrays.asList(userIds.split(",")); // List<String> userIdList = Arrays.asList(userIds.split(","));
List<String> phoneList = Arrays.asList(phonestr.split(",")); // List<String> phoneList = Arrays.asList(phonestr.split(","));
LinkedList<Object[]> addLink = CollectionUtil.linkedListObjectArr(); // LinkedList<Object[]> addLink = CollectionUtil.linkedListObjectArr();
for (int i = 0; i < userIdList.size(); i++) { // for (int i = 0; i < userIdList.size(); i++) {
MultiValueMap<String, String> params = new LinkedMultiValueMap(); // MultiValueMap<String, String> params = new LinkedMultiValueMap();
params.add("code", codeList.get(i)); // params.add("code", codeList.get(i));
params.add("userId", userIdList.get(i)); // params.add("userId", userIdList.get(i));
MultiValueMap<String, String> headers = CollectionUtil.linkedMultiValueMapStringString(); // MultiValueMap<String, String> headers = CollectionUtil.linkedMultiValueMapStringString();
headers.add("Accept", "application/json;charset=UTF-8"); // headers.add("Accept", "application/json;charset=UTF-8");
String post = HttpUtil.post(orderUrl + "/order/goblin/nft/airdrop", params, headers); // String post = HttpUtil.post(orderUrl + "/order/goblin/nft/airdrop", params, headers);
ResponseDto<Boolean> rsp = JsonUtils.fromJson(post, new TypeReference<ResponseDto<Boolean>>() { // ResponseDto<Boolean> rsp = JsonUtils.fromJson(post, new TypeReference<ResponseDto<Boolean>>() {
}); // });
if (rsp.isSuccess()) { // if (rsp.isSuccess()) {
addLink.add(new Object[]{adminUid, codeList.get(i)}); // addLink.add(new Object[]{adminUid, codeList.get(i)});
} else { // } else {
log.error("airdrop send fail:[{}]" + phoneList.get(i)); // log.error("airdrop send fail:[{}]" + phoneList.get(i));
} // }
} // }
//
if (addLink != null && !addLink.isEmpty()) { // if (addLink != null && !addLink.isEmpty()) {
LinkedList<String> sqls = CollectionUtil.linkedListString(); // LinkedList<String> sqls = CollectionUtil.linkedListString();
sqls.add(SqlMapping.get("goblin_activity.code.admin")); // sqls.add(SqlMapping.get("goblin_activity.code.admin"));
String sqlData = SqlMapping.gets(sqls, addLink); // String sqlData = SqlMapping.gets(sqls, addLink);
sendMsgByRedis(MQConst.GoblinQueue.GOBLIN_CODE_OPERA.getKey(), sqlData); // sendMsgByRedis(MQConst.GoblinQueue.GOBLIN_CODE_OPERA.getKey(), sqlData);
} // }
//
} catch (Exception e) { // } catch (Exception e) {
log.error("airdrop send fail:[{}]"); // log.error("airdrop send fail:[{}]");
} // }
//
return true; // return true;
} // }
//
//
public void sendMsgByRedis(String streamKey, String jsonMsg) { // public void sendMsgByRedis(String streamKey, String jsonMsg) {
HashMap<String, String> map = CollectionUtil.mapStringString(); // HashMap<String, String> map = CollectionUtil.mapStringString();
map.put(MQConst.QUEUE_MESSAGE_KEY, jsonMsg); // map.put(MQConst.QUEUE_MESSAGE_KEY, jsonMsg);
stringRedisTemplate.opsForStream().add(StreamRecords.mapBacked(map).withStreamKey(streamKey)); // stringRedisTemplate.opsForStream().add(StreamRecords.mapBacked(map).withStreamKey(streamKey));
} // }
//
//
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/ConsumerGoblinPhoneCodeReceiver.java
View file @ be62c0a2
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 lombok.extern.slf4j.Slf4j; //import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component; //import org.springframework.stereotype.Component;
//
@Slf4j //@Slf4j
@Component //@Component
public class ConsumerGoblinPhoneCodeReceiver extends AbstractHttpRedisReceiver { //public class ConsumerGoblinPhoneCodeReceiver extends AbstractHttpRedisReceiver {
//
@Override // @Override
protected String getRedisStreamKey() { // protected String getRedisStreamKey() {
return MQConst.GoblinQueue.GOBLIN_PHONE_CODE_OPERA.getKey(); // return MQConst.GoblinQueue.GOBLIN_PHONE_CODE_OPERA.getKey();
} // }
//
@Override // @Override
protected String getRedisStreamGroup() { // protected String getRedisStreamGroup() {
return MQConst.GoblinQueue.GOBLIN_PHONE_CODE_OPERA.getGroup(); // return MQConst.GoblinQueue.GOBLIN_PHONE_CODE_OPERA.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