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

Skip to content

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

迁移 consumer-goblin

parent c074aba1
Hide whitespace changes
Inline Side-by-side
Showing with 586 additions and 543 deletions
liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-base/src/main/java/com/liquidnet/service/consumer/base/config/candy/ConsumerCandyCouponBackRedisStreamConfig.java
View file @ fd709cc4
package com.liquidnet.service.consumer.base.config.candy;
import com.liquidnet.common.cache.redis.config.RedisStreamConfig;
import com.liquidnet.service.base.constant.MQConst;
import com.liquidnet.service.consumer.base.receiver.candy.ConsumerCandyCouponBackRdsReceiver;
import lombok.var;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
......@@ -11,55 +11,36 @@ 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 static com.liquidnet.service.base.constant.MQConst.CandyQueue.COUPON_BACK;
import java.util.ArrayList;
import java.util.List;
@Configuration
public class ConsumerCandyCouponBackRedisStreamConfig extends RedisStreamConfig {
@Autowired
ConsumerCandyCouponBackRdsReceiver consumerCandyCouponBackRdsReceiver;
@Autowired
StringRedisTemplate stringRedisTemplate;
/**
* 缺票登记
*
* @param listenerContainer
* @param t
* @return
*/
private Subscription receiveSqlCandyCouponBack(StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer, int t) {
return listenerContainer.receiveAutoAck(Consumer.from(COUPON_BACK.getGroup(), getConsumerName(COUPON_BACK.name() + t)),
StreamOffset.create(COUPON_BACK.getKey(), ReadOffset.lastConsumed()), consumerCandyCouponBackRdsReceiver);
}
/* —————————————————————————— ｜ —————————————————————————— ｜ —————————————————————————— */
/* -------------------------------------------------------- ｜ 缺票登记 */
@Bean
public Subscription subscriptionSqlCandyCouponBack0(RedisConnectionFactory factory) {
var listenerContainer = this.buildStreamMessageListenerContainer(factory);
var subscription = receiveSqlCandyCouponBack(listenerContainer, 0);
listenerContainer.start();
return subscription;
}
@Bean
public Subscription subscriptionSqlCandyCouponBack1(RedisConnectionFactory factory) {
var listenerContainer = this.buildStreamMessageListenerContainer(factory);
var subscription = receiveSqlCandyCouponBack(listenerContainer, 1);
listenerContainer.start();
return subscription;
}
@Bean
public Subscription subscriptionSqlCandyCouponBack2(RedisConnectionFactory factory) {
var listenerContainer = this.buildStreamMessageListenerContainer(factory);
var subscription = receiveSqlCandyCouponBack(listenerContainer, 2);
listenerContainer.start();
return subscription;
public List<Subscription> couponBack(RedisConnectionFactory factory) {
List<Subscription> subscriptionList = new ArrayList<>();
MQConst.CandyQueue stream = MQConst.CandyQueue.COUPON_BACK;
this.initStream(stringRedisTemplate, stream.getKey(), stream.getGroup());
for (int i = 0; i < 5; 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()), consumerCandyCouponBackRdsReceiver
));
listenerContainer.start();
}
return subscriptionList;
}
/* -------------------------------------------------------- ｜ */
}
liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-base/src/main/java/com/liquidnet/service/consumer/base/config/candy/ConsumerCandyCouponOrderBackRedisStreamConfig.java
View file @ fd709cc4
package com.liquidnet.service.consumer.base.config.candy;
import com.liquidnet.common.cache.redis.config.RedisStreamConfig;
import com.liquidnet.service.base.constant.MQConst;
import com.liquidnet.service.consumer.base.receiver.candy.ConsumerCandyCouponOrderBackRdsReceiver;
import lombok.var;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
......@@ -11,55 +11,35 @@ 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 static com.liquidnet.service.base.constant.MQConst.CandyQueue.COUPON_ORDER_BACK;
import java.util.ArrayList;
import java.util.List;
@Configuration
public class ConsumerCandyCouponOrderBackRedisStreamConfig extends RedisStreamConfig {
@Autowired
ConsumerCandyCouponOrderBackRdsReceiver consumerCandyCouponOrderBackRdsReceiver;
@Autowired
StringRedisTemplate stringRedisTemplate;
/**
* 缺票登记
*
* @param listenerContainer
* @param t
* @return
*/
private Subscription receiveSqlCandyCouponBack(StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer, int t) {
return listenerContainer.receiveAutoAck(Consumer.from(COUPON_ORDER_BACK.getGroup(), getConsumerName(COUPON_ORDER_BACK.name() + t)),
StreamOffset.create(COUPON_ORDER_BACK.getKey(), ReadOffset.lastConsumed()), consumerCandyCouponOrderBackRdsReceiver);
}
/* —————————————————————————— ｜ —————————————————————————— ｜ —————————————————————————— */
/* -------------------------------------------------------- ｜ 缺票登记 */
@Bean
public Subscription subscriptionSqlCandyCouponOrderBack0(RedisConnectionFactory factory) {
var listenerContainer = this.buildStreamMessageListenerContainer(factory);
var subscription = receiveSqlCandyCouponBack(listenerContainer, 0);
listenerContainer.start();
return subscription;
}
@Bean
public Subscription subscriptionSqlCandyCouponOrderBack1(RedisConnectionFactory factory) {
var listenerContainer = this.buildStreamMessageListenerContainer(factory);
var subscription = receiveSqlCandyCouponBack(listenerContainer, 1);
listenerContainer.start();
return subscription;
}
@Bean
public Subscription subscriptionSqlCandyCouponOrderBack2(RedisConnectionFactory factory) {
var listenerContainer = this.buildStreamMessageListenerContainer(factory);
var subscription = receiveSqlCandyCouponBack(listenerContainer, 2);
listenerContainer.start();
return subscription;
public List<Subscription> couponOrderBack(RedisConnectionFactory factory) {
List<Subscription> subscriptionList = new ArrayList<>();
MQConst.CandyQueue stream = MQConst.CandyQueue.COUPON_ORDER_BACK;
this.initStream(stringRedisTemplate, stream.getKey(), stream.getGroup());
for (int i = 0; i < 5; 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()), consumerCandyCouponOrderBackRdsReceiver
));
listenerContainer.start();
}
return subscriptionList;
}
/* -------------------------------------------------------- ｜ */
}
liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-base/src/main/java/com/liquidnet/service/consumer/base/config/candy/ConsumerCandyCouponReceiveRedisStreamConfig.java
View file @ fd709cc4
package com.liquidnet.service.consumer.base.config.candy;
import com.liquidnet.common.cache.redis.config.RedisStreamConfig;
import com.liquidnet.service.base.constant.MQConst;
import com.liquidnet.service.consumer.base.receiver.candy.ConsumerCandyCouponReceiveRdsReceiver;
import lombok.var;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
......@@ -11,55 +11,35 @@ 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 static com.liquidnet.service.base.constant.MQConst.CandyQueue.COUPON_RECEIVE;
import java.util.ArrayList;
import java.util.List;
@Configuration
public class ConsumerCandyCouponReceiveRedisStreamConfig extends RedisStreamConfig {
@Autowired
ConsumerCandyCouponReceiveRdsReceiver consumerCandyCouponReceiveRdsReceiver;
@Autowired
StringRedisTemplate stringRedisTemplate;
/**
* 缺票登记
*
* @param listenerContainer
* @param t
* @return
*/
private Subscription receiveSqlCandyCouponReceive(StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer, int t) {
return listenerContainer.receiveAutoAck(Consumer.from(COUPON_RECEIVE.getGroup(), getConsumerName(COUPON_RECEIVE.name() + t)),
StreamOffset.create(COUPON_RECEIVE.getKey(), ReadOffset.lastConsumed()), consumerCandyCouponReceiveRdsReceiver);
}
/* —————————————————————————— ｜ —————————————————————————— ｜ —————————————————————————— */
/* -------------------------------------------------------- ｜ 缺票登记 */
@Bean
public Subscription subscriptionSqlCandyCouponReceive0(RedisConnectionFactory factory) {
var listenerContainer = this.buildStreamMessageListenerContainer(factory);
var subscription = receiveSqlCandyCouponReceive(listenerContainer, 0);
listenerContainer.start();
return subscription;
}
@Bean
public Subscription subscriptionSqlCandyCouponReceive1(RedisConnectionFactory factory) {
var listenerContainer = this.buildStreamMessageListenerContainer(factory);
var subscription = receiveSqlCandyCouponReceive(listenerContainer, 1);
listenerContainer.start();
return subscription;
}
@Bean
public Subscription subscriptionSqlCandyCouponReceive2(RedisConnectionFactory factory) {
var listenerContainer = this.buildStreamMessageListenerContainer(factory);
var subscription = receiveSqlCandyCouponReceive(listenerContainer, 2);
listenerContainer.start();
return subscription;
public List<Subscription> couponReceive(RedisConnectionFactory factory) {
List<Subscription> subscriptionList = new ArrayList<>();
MQConst.CandyQueue stream = MQConst.CandyQueue.COUPON_RECEIVE;
this.initStream(stringRedisTemplate, stream.getKey(), stream.getGroup());
for (int i = 0; i < 5; 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()), consumerCandyCouponReceiveRdsReceiver
));
listenerContainer.start();
}
return subscriptionList;
}
/* -------------------------------------------------------- ｜ */
}
liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-base/src/main/java/com/liquidnet/service/consumer/base/config/candy/ConsumerCandyCouponUseRedisStreamConfig.java
View file @ fd709cc4
package com.liquidnet.service.consumer.base.config.candy;
import com.liquidnet.common.cache.redis.config.RedisStreamConfig;
import com.liquidnet.service.base.constant.MQConst;
import com.liquidnet.service.consumer.base.receiver.candy.ConsumerCandyCouponUseRdsReceiver;
import lombok.var;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
......@@ -11,56 +11,35 @@ 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 static com.liquidnet.service.base.constant.MQConst.CandyQueue.COUPON_USE;
import java.util.ArrayList;
import java.util.List;
@Configuration
public class ConsumerCandyCouponUseRedisStreamConfig extends RedisStreamConfig {
@Autowired
ConsumerCandyCouponUseRdsReceiver consumerCandyCouponUseRdsReceiver;
@Autowired
StringRedisTemplate stringRedisTemplate;
/**
* 缺票登记
*
* @param listenerContainer
* @param t
* @return
*/
private Subscription receiveSqlCandyCouponUse(StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer, int t) {
return listenerContainer.receiveAutoAck(Consumer.from(COUPON_USE.getGroup(), getConsumerName(COUPON_USE.name() + t)),
StreamOffset.create(COUPON_USE.getKey(), ReadOffset.lastConsumed()), consumerCandyCouponUseRdsReceiver
);
}
/* —————————————————————————— ｜ —————————————————————————— ｜ —————————————————————————— */
/* -------------------------------------------------------- ｜ 缺票登记 */
@Bean
public Subscription subscriptionSqlCandyCouponUse0(RedisConnectionFactory factory) {
var listenerContainer = this.buildStreamMessageListenerContainer(factory);
var subscription = receiveSqlCandyCouponUse(listenerContainer, 0);
listenerContainer.start();
return subscription;
}
@Bean
public Subscription subscriptionSqlCandyCouponUse1(RedisConnectionFactory factory) {
var listenerContainer = this.buildStreamMessageListenerContainer(factory);
var subscription = receiveSqlCandyCouponUse(listenerContainer, 1);
listenerContainer.start();
return subscription;
}
@Bean
public Subscription subscriptionSqlCandyCouponUse2(RedisConnectionFactory factory) {
var listenerContainer = this.buildStreamMessageListenerContainer(factory);
var subscription = receiveSqlCandyCouponUse(listenerContainer, 2);
listenerContainer.start();
return subscription;
public List<Subscription> couponUse(RedisConnectionFactory factory) {
List<Subscription> subscriptionList = new ArrayList<>();
MQConst.CandyQueue stream = MQConst.CandyQueue.COUPON_USE;
this.initStream(stringRedisTemplate, stream.getKey(), stream.getGroup());
for (int i = 0; i < 5; 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()), consumerCandyCouponUseRdsReceiver
));
listenerContainer.start();
}
return subscriptionList;
}
/* -------------------------------------------------------- ｜ */
}
liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-base/src/main/java/com/liquidnet/service/consumer/base/config/goblin/ConsumerGoblinOrderAGRedisStreamConfig.java 0 → 100644
View file @ fd709cc4
package com.liquidnet.service.consumer.base.config.goblin;
import com.liquidnet.common.cache.redis.config.RedisStreamConfig;
import com.liquidnet.service.base.constant.MQConst;
import com.liquidnet.service.consumer.base.receiver.goblin.ConsumerGoblinOrderAGRdsReceiver;
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 ConsumerGoblinOrderAGRedisStreamConfig extends RedisStreamConfig {
@Autowired
ConsumerGoblinOrderAGRdsReceiver consumerGoblinOrderAGRdsReceiver;
@Autowired
StringRedisTemplate stringRedisTemplate;
@Bean
public List<Subscription> orderAg(RedisConnectionFactory factory) {
List<Subscription> subscriptionList = new ArrayList<>();
MQConst.GoblinQueue stream = MQConst.GoblinQueue.GOBLIN_ORDER_AGAIN;
this.initStream(stringRedisTemplate, stream.getKey(), stream.getGroup());
for (int i = 0; i < 5; 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()), consumerGoblinOrderAGRdsReceiver
));
listenerContainer.start();
}
return subscriptionList;
}
}
liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-base/src/main/java/com/liquidnet/service/consumer/base/config/goblin/ConsumerGoblinOrderCPRedisStreamConfig.java 0 → 100644
View file @ fd709cc4
package com.liquidnet.service.consumer.base.config.goblin;
import com.liquidnet.common.cache.redis.config.RedisStreamConfig;
import com.liquidnet.service.base.constant.MQConst;
import com.liquidnet.service.consumer.base.receiver.goblin.ConsumerGoblinOrderCPRdsReceiver;
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 ConsumerGoblinOrderCPRedisStreamConfig extends RedisStreamConfig {
@Autowired
ConsumerGoblinOrderCPRdsReceiver consumerGoblinOrderCPRdsReceiver;
@Autowired
StringRedisTemplate stringRedisTemplate;
@Bean
public List<Subscription> orderCP(RedisConnectionFactory factory) {
List<Subscription> subscriptionList = new ArrayList<>();
MQConst.GoblinQueue stream = MQConst.GoblinQueue.GOBLIN_ORDER_CREATE_PAY;
this.initStream(stringRedisTemplate, stream.getKey(), stream.getGroup());
for (int i = 0; i < 5; 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()), consumerGoblinOrderCPRdsReceiver
));
listenerContainer.start();
}
return subscriptionList;
}
}
liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-base/src/main/java/com/liquidnet/service/consumer/base/receiver/goblin/ConsumerGoblinOrderAGRdsReceiver.java 0 → 100644
View file @ fd709cc4
package com.liquidnet.service.consumer.base.receiver.goblin;
import com.liquidnet.service.base.constant.MQConst;
import com.liquidnet.service.consumer.base.receiver.AbstractSqlRedisReceiver;
import org.springframework.stereotype.Component;
@Component
public class ConsumerGoblinOrderAGRdsReceiver extends AbstractSqlRedisReceiver {
@Override
protected String getRedisStreamKey() {
return MQConst.GoblinQueue.GOBLIN_ORDER_AGAIN.getKey();
}
@Override
protected String getRedisStreamGroup() {
return MQConst.GoblinQueue.GOBLIN_ORDER_AGAIN.getGroup();
}
}
liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-base/src/main/java/com/liquidnet/service/consumer/base/receiver/goblin/ConsumerGoblinOrderCPRdsReceiver.java 0 → 100644
View file @ fd709cc4
package com.liquidnet.service.consumer.base.receiver.goblin;
import com.liquidnet.service.base.constant.MQConst;
import com.liquidnet.service.consumer.base.receiver.AbstractSqlRedisReceiver;
import org.springframework.stereotype.Component;
@Component
public class ConsumerGoblinOrderCPRdsReceiver extends AbstractSqlRedisReceiver {
@Override
protected String getRedisStreamKey() {
return MQConst.GoblinQueue.GOBLIN_ORDER_CREATE_PAY.getKey();
}
@Override
protected String getRedisStreamGroup() {
return MQConst.GoblinQueue.GOBLIN_ORDER_CREATE_PAY.getGroup();
}
}
liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-goblin/src/main/java/com/liquidnet/service/consumer/goblin/config/ConsumerGoblinOrderAGRedisStreamConfig.java
View file @ fd709cc4
package com.liquidnet.service.consumer.goblin.config;
import com.liquidnet.common.cache.redis.config.RedisStreamConfig;
import com.liquidnet.service.base.constant.MQConst;
import com.liquidnet.service.consumer.goblin.receiver.ConsumerGoblinOrderAGRdsReceiver;
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.stream.StreamMessageListenerContainer;
import org.springframework.data.redis.stream.Subscription;
@Configuration
public class ConsumerGoblinOrderAGRedisStreamConfig extends RedisStreamConfig {
@Autowired
ConsumerGoblinOrderAGRdsReceiver consumerGoblinOrderAGRdsReceiver;
/**
* 缺票登记
*
* @param listenerContainer
* @param t
* @return
*/
private Subscription receiveGoblinOrderAG(StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer, int t) {
return listenerContainer.receiveAutoAck(Consumer.from(MQConst.GoblinQueue.GOBLIN_ORDER_AGAIN.getGroup(), getConsumerName(MQConst.GoblinQueue.GOBLIN_ORDER_AGAIN.name() + t)),
StreamOffset.create(MQConst.GoblinQueue.GOBLIN_ORDER_AGAIN.getKey(), ReadOffset.lastConsumed()), consumerGoblinOrderAGRdsReceiver);
}
/* —————————————————————————— ｜ —————————————————————————— ｜ —————————————————————————— */
/* -------------------------------------------------------- ｜ 缺票登记 */
@Bean
public Subscription subscriptionGoblinOrderAG0(RedisConnectionFactory factory) {
var listenerContainer = this.buildStreamMessageListenerContainer(factory);
var subscription = receiveGoblinOrderAG(listenerContainer, 0);
listenerContainer.start();
return subscription;
}
@Bean
public Subscription subscriptionGoblinOrderAG1(RedisConnectionFactory factory) {
var listenerContainer = this.buildStreamMessageListenerContainer(factory);
var subscription = receiveGoblinOrderAG(listenerContainer, 1);
listenerContainer.start();
return subscription;
}
@Bean
public Subscription subscriptionGoblinOrderAG2(RedisConnectionFactory factory) {
var listenerContainer = this.buildStreamMessageListenerContainer(factory);
var subscription = receiveGoblinOrderAG(listenerContainer, 2);
listenerContainer.start();
return subscription;
}
/* -------------------------------------------------------- ｜ */
}
//package com.liquidnet.service.consumer.goblin.config;
//
//import com.liquidnet.common.cache.redis.config.RedisStreamConfig;
//import com.liquidnet.service.base.constant.MQConst;
//import com.liquidnet.service.consumer.goblin.receiver.ConsumerGoblinOrderAGRdsReceiver;
//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.stream.StreamMessageListenerContainer;
//import org.springframework.data.redis.stream.Subscription;
//
//@Configuration
//public class ConsumerGoblinOrderAGRedisStreamConfig extends RedisStreamConfig {
// @Autowired
// ConsumerGoblinOrderAGRdsReceiver consumerGoblinOrderAGRdsReceiver;
//
// /**
// * 缺票登记
// *
// * @param listenerContainer
// * @param t
// * @return
// */
// private Subscription receiveGoblinOrderAG(StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer, int t) {
// return listenerContainer.receiveAutoAck(Consumer.from(MQConst.GoblinQueue.GOBLIN_ORDER_AGAIN.getGroup(), getConsumerName(MQConst.GoblinQueue.GOBLIN_ORDER_AGAIN.name() + t)),
// StreamOffset.create(MQConst.GoblinQueue.GOBLIN_ORDER_AGAIN.getKey(), ReadOffset.lastConsumed()), consumerGoblinOrderAGRdsReceiver);
// }
//
// /* —————————————————————————— ｜ —————————————————————————— ｜ —————————————————————————— */
//
// /* -------------------------------------------------------- ｜ 缺票登记 */
//
// @Bean
// public Subscription subscriptionGoblinOrderAG0(RedisConnectionFactory factory) {
// var listenerContainer = this.buildStreamMessageListenerContainer(factory);
// var subscription = receiveGoblinOrderAG(listenerContainer, 0);
// listenerContainer.start();
// return subscription;
// }
//
// @Bean
// public Subscription subscriptionGoblinOrderAG1(RedisConnectionFactory factory) {
// var listenerContainer = this.buildStreamMessageListenerContainer(factory);
// var subscription = receiveGoblinOrderAG(listenerContainer, 1);
// listenerContainer.start();
// return subscription;
// }
//
// @Bean
// public Subscription subscriptionGoblinOrderAG2(RedisConnectionFactory factory) {
// var listenerContainer = this.buildStreamMessageListenerContainer(factory);
// var subscription = receiveGoblinOrderAG(listenerContainer, 2);
// listenerContainer.start();
// return subscription;
// }
//
// /* -------------------------------------------------------- ｜ */
//}
liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-goblin/src/main/java/com/liquidnet/service/consumer/goblin/config/ConsumerGoblinOrderCPRedisStreamConfig.java
View file @ fd709cc4
package com.liquidnet.service.consumer.goblin.config;
import com.liquidnet.common.cache.redis.config.RedisStreamConfig;
import com.liquidnet.service.base.constant.MQConst;
import com.liquidnet.service.consumer.goblin.receiver.ConsumerGoblinOrderCPRdsReceiver;
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.stream.StreamMessageListenerContainer;
import org.springframework.data.redis.stream.Subscription;
@Configuration
public class ConsumerGoblinOrderCPRedisStreamConfig extends RedisStreamConfig {
@Autowired
ConsumerGoblinOrderCPRdsReceiver consumerGoblinOrderCPRdsReceiver;
/**
* 商城订单
*/
private Subscription receiveGoblinOrderCP(StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer, int t) {
return listenerContainer.receiveAutoAck(Consumer.from(MQConst.GoblinQueue.GOBLIN_ORDER_CREATE_PAY.getGroup(), getConsumerName(MQConst.GoblinQueue.GOBLIN_ORDER_CREATE_PAY.name() + t)),
StreamOffset.create(MQConst.GoblinQueue.GOBLIN_ORDER_CREATE_PAY.getKey(), ReadOffset.lastConsumed()), consumerGoblinOrderCPRdsReceiver);
}
/* —————————————————————————— ｜ —————————————————————————— ｜ —————————————————————————— */
/*-------------------------------------- 商城订单 --------------------------------------*/
@Bean
public Subscription subscriptionGoblinOrderCP0(RedisConnectionFactory factory) {
var listenerContainer = this.buildStreamMessageListenerContainer(factory);
var subscription = receiveGoblinOrderCP(listenerContainer, 0);
listenerContainer.start();
return subscription;
}
@Bean
public Subscription subscriptionGoblinOrderCP1(RedisConnectionFactory factory) {
var listenerContainer = this.buildStreamMessageListenerContainer(factory);
var subscription = receiveGoblinOrderCP(listenerContainer, 1);
listenerContainer.start();
return subscription;
}
@Bean
public Subscription subscriptionGoblinOrderCP2(RedisConnectionFactory factory) {
var listenerContainer = this.buildStreamMessageListenerContainer(factory);
var subscription = receiveGoblinOrderCP(listenerContainer, 2);
listenerContainer.start();
return subscription;
}
/* -------------------------------------------------------- ｜ */
}
//package com.liquidnet.service.consumer.goblin.config;
//
//import com.liquidnet.common.cache.redis.config.RedisStreamConfig;
//import com.liquidnet.service.base.constant.MQConst;
//import com.liquidnet.service.consumer.goblin.receiver.ConsumerGoblinOrderCPRdsReceiver;
//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.stream.StreamMessageListenerContainer;
//import org.springframework.data.redis.stream.Subscription;
//
//@Configuration
//public class ConsumerGoblinOrderCPRedisStreamConfig extends RedisStreamConfig {
// @Autowired
// ConsumerGoblinOrderCPRdsReceiver consumerGoblinOrderCPRdsReceiver;
//
// /**
// * 商城订单
// */
// private Subscription receiveGoblinOrderCP(StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer, int t) {
// return listenerContainer.receiveAutoAck(Consumer.from(MQConst.GoblinQueue.GOBLIN_ORDER_CREATE_PAY.getGroup(), getConsumerName(MQConst.GoblinQueue.GOBLIN_ORDER_CREATE_PAY.name() + t)),
// StreamOffset.create(MQConst.GoblinQueue.GOBLIN_ORDER_CREATE_PAY.getKey(), ReadOffset.lastConsumed()), consumerGoblinOrderCPRdsReceiver);
// }
//
// /* —————————————————————————— ｜ —————————————————————————— ｜ —————————————————————————— */
//
// /*-------------------------------------- 商城订单 --------------------------------------*/
//
// @Bean
// public Subscription subscriptionGoblinOrderCP0(RedisConnectionFactory factory) {
// var listenerContainer = this.buildStreamMessageListenerContainer(factory);
// var subscription = receiveGoblinOrderCP(listenerContainer, 0);
// listenerContainer.start();
// return subscription;
// }
//
// @Bean
// public Subscription subscriptionGoblinOrderCP1(RedisConnectionFactory factory) {
// var listenerContainer = this.buildStreamMessageListenerContainer(factory);
// var subscription = receiveGoblinOrderCP(listenerContainer, 1);
// listenerContainer.start();
// return subscription;
// }
//
// @Bean
// public Subscription subscriptionGoblinOrderCP2(RedisConnectionFactory factory) {
// var listenerContainer = this.buildStreamMessageListenerContainer(factory);
// var subscription = receiveGoblinOrderCP(listenerContainer, 2);
// listenerContainer.start();
// return subscription;
// }
//
// /* -------------------------------------------------------- ｜ */
//}
liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-goblin/src/main/java/com/liquidnet/service/consumer/goblin/receiver/AbstractBizRedisReceiver.java
View file @ fd709cc4
package com.liquidnet.service.consumer.goblin.receiver;
import com.liquidnet.service.base.constant.MQConst;
import com.liquidnet.service.consumer.goblin.service.IBaseDao;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.stream.StreamListener;
/**
* 公共的业务队列消息监听器，具体业务消费逻辑通过`consumerMessageHandler`实现
*
* @author zhanggb
* Created by IntelliJ IDEA at 2022/3/31
*/
@Slf4j
public abstract class AbstractBizRedisReceiver implements StreamListener<String, MapRecord<String, String, String>> {
@Autowired
public IBaseDao baseDao;
@Autowired
public StringRedisTemplate stringRedisTemplate;
@Override
public void onMessage(MapRecord<String, String, String> message) {
String redisStreamKey = this.getRedisStreamKey();
log.debug("CONSUMER MSG[streamKey:{},messageId:{},stream:{},body:{}]", redisStreamKey, message.getId(), message.getStream(), message.getValue());
boolean result = this.consumerMessageHandler(message.getValue().get(MQConst.QUEUE_MESSAGE_KEY));
log.info("CONSUMER MSG RESULT:{} ==> [{}]MESSAGE_ID:{}", result, redisStreamKey, message.getId());
try {
stringRedisTemplate.opsForStream().acknowledge(getRedisStreamGroup(), message);
} catch (Exception e) {
log.error("#CONSUMER MSG EX_ACK ==> [{}]RESULT:{},MESSAGE:{}", redisStreamKey, result, message.getValue(), e);
}
try {
stringRedisTemplate.opsForStream().delete(redisStreamKey, message.getId());
} catch (Exception e) {
log.error("#CONSUMER MSG EX_DEL ==> [{}]RESULT:{},MESSAGE:{}", redisStreamKey, result, message.getValue(), e);
}
}
protected abstract boolean consumerMessageHandler(String msg);
protected abstract String getRedisStreamKey();
protected abstract String getRedisStreamGroup();
}
\ No newline at end of file
//package com.liquidnet.service.consumer.goblin.receiver;
//
//import com.liquidnet.service.base.constant.MQConst;
//import com.liquidnet.service.consumer.goblin.service.IBaseDao;
//import lombok.extern.slf4j.Slf4j;
//import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.data.redis.connection.stream.MapRecord;
//import org.springframework.data.redis.core.StringRedisTemplate;
//import org.springframework.data.redis.stream.StreamListener;
//
///**
// * 公共的业务队列消息监听器，具体业务消费逻辑通过`consumerMessageHandler`实现
// *
// * @author zhanggb
// * Created by IntelliJ IDEA at 2022/3/31
// */
//@Slf4j
//public abstract class AbstractBizRedisReceiver implements StreamListener<String, MapRecord<String, String, String>> {
// @Autowired
// public IBaseDao baseDao;
// @Autowired
// public StringRedisTemplate stringRedisTemplate;
//
// @Override
// public void onMessage(MapRecord<String, String, String> message) {
// String redisStreamKey = this.getRedisStreamKey();
// log.debug("CONSUMER MSG[streamKey:{},messageId:{},stream:{},body:{}]", redisStreamKey, message.getId(), message.getStream(), message.getValue());
// boolean result = this.consumerMessageHandler(message.getValue().get(MQConst.QUEUE_MESSAGE_KEY));
// log.info("CONSUMER MSG RESULT:{} ==> [{}]MESSAGE_ID:{}", result, redisStreamKey, message.getId());
//
// try {
// stringRedisTemplate.opsForStream().acknowledge(getRedisStreamGroup(), message);
// } catch (Exception e) {
// log.error("#CONSUMER MSG EX_ACK ==> [{}]RESULT:{},MESSAGE:{}", redisStreamKey, result, message.getValue(), e);
// }
// try {
// stringRedisTemplate.opsForStream().delete(redisStreamKey, message.getId());
// } catch (Exception e) {
// log.error("#CONSUMER MSG EX_DEL ==> [{}]RESULT:{},MESSAGE:{}", redisStreamKey, result, message.getValue(), e);
// }
// }
//
// protected abstract boolean consumerMessageHandler(String msg);
//
// protected abstract String getRedisStreamKey();
//
// protected abstract String getRedisStreamGroup();
//}
\ No newline at end of file
liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-goblin/src/main/java/com/liquidnet/service/consumer/goblin/receiver/AbstractMdbRedisReceiver.java
View file @ fd709cc4
package com.liquidnet.service.consumer.goblin.receiver;
import com.liquidnet.common.cache.redis.util.RedisUtil;
import com.liquidnet.commons.lang.util.CollectionUtil;
import com.liquidnet.commons.lang.util.JsonUtils;
import com.liquidnet.service.base.constant.MQConst;
import com.liquidnet.service.goblin.dto.GoblinQueueBizMongoDto;
import com.mongodb.BasicDBObject;
import com.mongodb.client.result.UpdateResult;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.StreamRecords;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.stream.StreamListener;
import java.util.HashMap;
/**
* 公共的业务Mongo数据处理队列消息监听器，具体业务消费逻辑通过`consumerMessageHandler`实现
*
* @author zhanggb
* Created by IntelliJ IDEA at 2022/4/22
*/
@Slf4j
public abstract class AbstractMdbRedisReceiver implements StreamListener<String, MapRecord<String, String, String>> {
@Autowired
public StringRedisTemplate stringRedisTemplate;
@Autowired
private MongoTemplate mongoTemplate;
@Autowired
private RedisUtil redisUtil;
private static final BasicDBObject BASIC_DB_OBJECT = new BasicDBObject();
@Override
public void onMessage(MapRecord<String, String, String> message) {
String redisStreamKey = this.getRedisStreamKey();
log.debug("CONSUMER MSG[streamKey:{},messageId:{},stream:{},body:{}]", redisStreamKey, message.getId(), message.getStream(), message.getValue());
boolean result = this.consumerMessageHandler(message.getValue().get(MQConst.QUEUE_MESSAGE_KEY));
log.info("CONSUMER MSG RESULT:{} ==> [{}]MESSAGE_ID:{}", result, redisStreamKey, message.getId());
try {
stringRedisTemplate.opsForStream().acknowledge(getRedisStreamGroup(), message);
} catch (Exception e) {
log.error("#CONSUMER MSG EX_ACK ==> [{}]RESULT:{},MESSAGE:{}", redisStreamKey, result, message.getValue(), e);
}
try {
stringRedisTemplate.opsForStream().delete(redisStreamKey, message.getId());
} catch (Exception e) {
log.error("#CONSUMER MSG EX_DEL ==> [{}]RESULT:{},MESSAGE:{}", redisStreamKey, result, message.getValue(), e);
}
}
private boolean consumerMessageHandler(String msg) {
boolean aBoolean = true;
try {
GoblinQueueBizMongoDto queueBizMongoDto = JsonUtils.fromJson(msg, GoblinQueueBizMongoDto.class);
if (null != queueBizMongoDto) {
String collectName = queueBizMongoDto.getCollect(), columnName = queueBizMongoDto.getColumn();
String prefix = queueBizMongoDto.getPrefix(), bizId = queueBizMongoDto.getBizId();
Object o;
switch (queueBizMongoDto.getOpType()) {
case 1:// insert
o = redisUtil.get(prefix.concat(bizId));
if (null != o) {
mongoTemplate.insert(o, collectName);
}
break;
case 2:// update
o = redisUtil.get(prefix.concat(bizId));
if (null != o) {
BasicDBObject basicDBObject = (BasicDBObject) AbstractMdbRedisReceiver.BASIC_DB_OBJECT.clone();
UpdateResult updateResult = mongoTemplate.getCollection(collectName).updateOne(
Query.query(Criteria.where(columnName).is(bizId)).getQueryObject(),
basicDBObject.append("$set", mongoTemplate.getConverter().convertToMongoType(o))
);
}
break;
default:
log.error("CONSUMER MSG ERR_HANDLE[未知的操作类型,{}]", msg);
break;
}
}
} catch (Exception e) {
log.error("CONSUMER MSG EX_HANDLE ==> [{}]:{}", this.getRedisStreamKey(), msg, e);
aBoolean = false;
} finally {
if (!aBoolean) {
HashMap<String, String> map = CollectionUtil.mapStringString();
map.put(MQConst.QUEUE_MESSAGE_KEY, msg);
stringRedisTemplate.opsForStream().add(StreamRecords.mapBacked(map).withStreamKey(this.getRedisStreamKey()));
}
}
return aBoolean;
}
protected abstract String getRedisStreamKey();
protected abstract String getRedisStreamGroup();
}
\ No newline at end of file
//package com.liquidnet.service.consumer.goblin.receiver;
//
//import com.liquidnet.common.cache.redis.util.RedisUtil;
//import com.liquidnet.commons.lang.util.CollectionUtil;
//import com.liquidnet.commons.lang.util.JsonUtils;
//import com.liquidnet.service.base.constant.MQConst;
//import com.liquidnet.service.goblin.dto.GoblinQueueBizMongoDto;
//import com.mongodb.BasicDBObject;
//import com.mongodb.client.result.UpdateResult;
//import lombok.extern.slf4j.Slf4j;
//import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.data.mongodb.core.MongoTemplate;
//import org.springframework.data.mongodb.core.query.Criteria;
//import org.springframework.data.mongodb.core.query.Query;
//import org.springframework.data.redis.connection.stream.MapRecord;
//import org.springframework.data.redis.connection.stream.StreamRecords;
//import org.springframework.data.redis.core.StringRedisTemplate;
//import org.springframework.data.redis.stream.StreamListener;
//
//import java.util.HashMap;
//
///**
// * 公共的业务Mongo数据处理队列消息监听器，具体业务消费逻辑通过`consumerMessageHandler`实现
// *
// * @author zhanggb
// * Created by IntelliJ IDEA at 2022/4/22
// */
//@Slf4j
//public abstract class AbstractMdbRedisReceiver implements StreamListener<String, MapRecord<String, String, String>> {
// @Autowired
// public StringRedisTemplate stringRedisTemplate;
// @Autowired
// private MongoTemplate mongoTemplate;
// @Autowired
// private RedisUtil redisUtil;
//
// private static final BasicDBObject BASIC_DB_OBJECT = new BasicDBObject();
//
// @Override
// public void onMessage(MapRecord<String, String, String> message) {
// String redisStreamKey = this.getRedisStreamKey();
// log.debug("CONSUMER MSG[streamKey:{},messageId:{},stream:{},body:{}]", redisStreamKey, message.getId(), message.getStream(), message.getValue());
// boolean result = this.consumerMessageHandler(message.getValue().get(MQConst.QUEUE_MESSAGE_KEY));
// log.info("CONSUMER MSG RESULT:{} ==> [{}]MESSAGE_ID:{}", result, redisStreamKey, message.getId());
//
// try {
// stringRedisTemplate.opsForStream().acknowledge(getRedisStreamGroup(), message);
// } catch (Exception e) {
// log.error("#CONSUMER MSG EX_ACK ==> [{}]RESULT:{},MESSAGE:{}", redisStreamKey, result, message.getValue(), e);
// }
// try {
// stringRedisTemplate.opsForStream().delete(redisStreamKey, message.getId());
// } catch (Exception e) {
// log.error("#CONSUMER MSG EX_DEL ==> [{}]RESULT:{},MESSAGE:{}", redisStreamKey, result, message.getValue(), e);
// }
// }
//
// private boolean consumerMessageHandler(String msg) {
// boolean aBoolean = true;
// try {
// GoblinQueueBizMongoDto queueBizMongoDto = JsonUtils.fromJson(msg, GoblinQueueBizMongoDto.class);
// if (null != queueBizMongoDto) {
// String collectName = queueBizMongoDto.getCollect(), columnName = queueBizMongoDto.getColumn();
// String prefix = queueBizMongoDto.getPrefix(), bizId = queueBizMongoDto.getBizId();
//
// Object o;
// switch (queueBizMongoDto.getOpType()) {
// case 1:// insert
// o = redisUtil.get(prefix.concat(bizId));
// if (null != o) {
// mongoTemplate.insert(o, collectName);
// }
// break;
// case 2:// update
// o = redisUtil.get(prefix.concat(bizId));
// if (null != o) {
// BasicDBObject basicDBObject = (BasicDBObject) AbstractMdbRedisReceiver.BASIC_DB_OBJECT.clone();
// UpdateResult updateResult = mongoTemplate.getCollection(collectName).updateOne(
// Query.query(Criteria.where(columnName).is(bizId)).getQueryObject(),
// basicDBObject.append("$set", mongoTemplate.getConverter().convertToMongoType(o))
// );
// }
// break;
// default:
// log.error("CONSUMER MSG ERR_HANDLE[未知的操作类型,{}]", msg);
// break;
// }
// }
// } catch (Exception e) {
// log.error("CONSUMER MSG EX_HANDLE ==> [{}]:{}", this.getRedisStreamKey(), msg, e);
// aBoolean = false;
// } finally {
// if (!aBoolean) {
// HashMap<String, String> map = CollectionUtil.mapStringString();
// map.put(MQConst.QUEUE_MESSAGE_KEY, msg);
// stringRedisTemplate.opsForStream().add(StreamRecords.mapBacked(map).withStreamKey(this.getRedisStreamKey()));
// }
// }
// return aBoolean;
// }
//
// protected abstract String getRedisStreamKey();
//
// protected abstract String getRedisStreamGroup();
//}
\ No newline at end of file
liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-goblin/src/main/java/com/liquidnet/service/consumer/goblin/receiver/AbstractSqlRedisReceiver.java
View file @ fd709cc4
package com.liquidnet.service.consumer.goblin.receiver;
import com.liquidnet.commons.lang.util.CollectionUtil;
import com.liquidnet.commons.lang.util.JsonUtils;
import com.liquidnet.service.base.SqlMapping;
import com.liquidnet.service.consumer.goblin.service.IBaseDao;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.StreamRecords;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.stream.StreamListener;
import java.util.HashMap;
/**
* 公共的SQL队列消息监听器，具体SQL消费逻辑统一使用`consumerMessageHandler`
*
* @author zhanggb
* Created by IntelliJ IDEA at 2022/3/31
*/
@Slf4j
public abstract class AbstractSqlRedisReceiver implements StreamListener<String, MapRecord<String, String, String>> {
@Autowired
private IBaseDao baseDao;
@Autowired
StringRedisTemplate stringRedisTemplate;
@Override
public void onMessage(MapRecord<String, String, String> message) {
String redisStreamKey = this.getRedisStreamKey();
log.debug("CONSUMER MSG[streamKey:{},messageId:{},stream:{},body:{}]", redisStreamKey, message.getId(), message.getStream(), message.getValue());
boolean result = this.consumerMessageHandler(message.getValue().get("message"));
log.info("CONSUMER MSG RESULT:{} ==> [{}]MESSAGE_ID:{}", result, redisStreamKey, message.getId());
try {
stringRedisTemplate.opsForStream().acknowledge(getRedisStreamGroup(), message);
} catch (Exception e) {
log.error("#CONSUMER MSG EX_ACK ==> [{}]RESULT:{},MESSAGE:{}", redisStreamKey, result, message.getValue(), e);
}
try {
stringRedisTemplate.opsForStream().delete(redisStreamKey, message.getId());
} catch (Exception e) {
log.error("#CONSUMER MSG EX_DEL ==> [{}]RESULT:{},MESSAGE:{}", redisStreamKey, result, message.getValue(), e);
}
}
private boolean consumerMessageHandler(String msg) {
boolean aBoolean = false;
try {
SqlMapping.SqlMessage sqlMessage = JsonUtils.fromJson(msg, SqlMapping.SqlMessage.class);
if (sqlMessage == null) {
aBoolean = true;
} else {
aBoolean = baseDao.batchSqls(sqlMessage.getSqls(), sqlMessage.getArgs());
}
} catch (Exception e) {
log.error("CONSUMER MSG EX_HANDLE ==> [{}]:{}", this.getRedisStreamKey(), msg, e);
} finally {
if (!aBoolean) {
HashMap<String, String> map = CollectionUtil.mapStringString();
map.put("message", msg);
stringRedisTemplate.opsForStream().add(StreamRecords.mapBacked(map).withStreamKey(this.getRedisStreamKey()));
}
}
return aBoolean;
}
protected abstract String getRedisStreamKey();
protected abstract String getRedisStreamGroup();
}
\ No newline at end of file
//package com.liquidnet.service.consumer.goblin.receiver;
//
//import com.liquidnet.commons.lang.util.CollectionUtil;
//import com.liquidnet.commons.lang.util.JsonUtils;
//import com.liquidnet.service.base.SqlMapping;
//import com.liquidnet.service.consumer.goblin.service.IBaseDao;
//import lombok.extern.slf4j.Slf4j;
//import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.data.redis.connection.stream.MapRecord;
//import org.springframework.data.redis.connection.stream.StreamRecords;
//import org.springframework.data.redis.core.StringRedisTemplate;
//import org.springframework.data.redis.stream.StreamListener;
//
//import java.util.HashMap;
//
///**
// * 公共的SQL队列消息监听器，具体SQL消费逻辑统一使用`consumerMessageHandler`
// *
// * @author zhanggb
// * Created by IntelliJ IDEA at 2022/3/31
// */
//@Slf4j
//public abstract class AbstractSqlRedisReceiver implements StreamListener<String, MapRecord<String, String, String>> {
// @Autowired
// private IBaseDao baseDao;
// @Autowired
// StringRedisTemplate stringRedisTemplate;
//
// @Override
// public void onMessage(MapRecord<String, String, String> message) {
// String redisStreamKey = this.getRedisStreamKey();
// log.debug("CONSUMER MSG[streamKey:{},messageId:{},stream:{},body:{}]", redisStreamKey, message.getId(), message.getStream(), message.getValue());
// boolean result = this.consumerMessageHandler(message.getValue().get("message"));
// log.info("CONSUMER MSG RESULT:{} ==> [{}]MESSAGE_ID:{}", result, redisStreamKey, message.getId());
//
// try {
// stringRedisTemplate.opsForStream().acknowledge(getRedisStreamGroup(), message);
// } catch (Exception e) {
// log.error("#CONSUMER MSG EX_ACK ==> [{}]RESULT:{},MESSAGE:{}", redisStreamKey, result, message.getValue(), e);
// }
// try {
// stringRedisTemplate.opsForStream().delete(redisStreamKey, message.getId());
// } catch (Exception e) {
// log.error("#CONSUMER MSG EX_DEL ==> [{}]RESULT:{},MESSAGE:{}", redisStreamKey, result, message.getValue(), e);
// }
// }
//
// private boolean consumerMessageHandler(String msg) {
// boolean aBoolean = false;
// try {
// SqlMapping.SqlMessage sqlMessage = JsonUtils.fromJson(msg, SqlMapping.SqlMessage.class);
// if (sqlMessage == null) {
// aBoolean = true;
// } else {
// aBoolean = baseDao.batchSqls(sqlMessage.getSqls(), sqlMessage.getArgs());
// }
// } catch (Exception e) {
// log.error("CONSUMER MSG EX_HANDLE ==> [{}]:{}", this.getRedisStreamKey(), msg, e);
// } finally {
// if (!aBoolean) {
// HashMap<String, String> map = CollectionUtil.mapStringString();
// map.put("message", msg);
// stringRedisTemplate.opsForStream().add(StreamRecords.mapBacked(map).withStreamKey(this.getRedisStreamKey()));
// }
// }
// return aBoolean;
// }
//
// protected abstract String getRedisStreamKey();
//
// protected abstract String getRedisStreamGroup();
//}
\ No newline at end of file
liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-goblin/src/main/java/com/liquidnet/service/consumer/goblin/receiver/ConsumerGoblinOrderAGRdsReceiver.java
View file @ fd709cc4
package com.liquidnet.service.consumer.goblin.receiver;
import com.liquidnet.service.base.constant.MQConst;
import org.springframework.stereotype.Component;
@Component
public class ConsumerGoblinOrderAGRdsReceiver extends AbstractSqlRedisReceiver {
@Override
protected String getRedisStreamKey() {
return MQConst.GoblinQueue.GOBLIN_ORDER_AGAIN.getKey();
}
@Override
protected String getRedisStreamGroup() {
return MQConst.GoblinQueue.GOBLIN_ORDER_AGAIN.getGroup();
}
}
//package com.liquidnet.service.consumer.goblin.receiver;
//
//import com.liquidnet.service.base.constant.MQConst;
//import org.springframework.stereotype.Component;
//
//@Component
//public class ConsumerGoblinOrderAGRdsReceiver extends AbstractSqlRedisReceiver {
// @Override
// protected String getRedisStreamKey() {
// return MQConst.GoblinQueue.GOBLIN_ORDER_AGAIN.getKey();
// }
//
// @Override
// protected String getRedisStreamGroup() {
// return MQConst.GoblinQueue.GOBLIN_ORDER_AGAIN.getGroup();
// }
//}
liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-goblin/src/main/java/com/liquidnet/service/consumer/goblin/receiver/ConsumerGoblinOrderCPRdsReceiver.java
View file @ fd709cc4
package com.liquidnet.service.consumer.goblin.receiver;
import com.liquidnet.service.base.constant.MQConst;
import org.springframework.stereotype.Component;
@Component
public class ConsumerGoblinOrderCPRdsReceiver extends AbstractSqlRedisReceiver {
@Override
protected String getRedisStreamKey() {
return MQConst.GoblinQueue.GOBLIN_ORDER_CREATE_PAY.getKey();
}
@Override
protected String getRedisStreamGroup() {
return MQConst.GoblinQueue.GOBLIN_ORDER_CREATE_PAY.getGroup();
}
}
//package com.liquidnet.service.consumer.goblin.receiver;
//
//import com.liquidnet.service.base.constant.MQConst;
//import org.springframework.stereotype.Component;
//
//@Component
//public class ConsumerGoblinOrderCPRdsReceiver extends AbstractSqlRedisReceiver {
// @Override
// protected String getRedisStreamKey() {
// return MQConst.GoblinQueue.GOBLIN_ORDER_CREATE_PAY.getKey();
// }
//
// @Override
// protected String getRedisStreamGroup() {
// return MQConst.GoblinQueue.GOBLIN_ORDER_CREATE_PAY.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