记得上下班打卡 | git大法好,push需谨慎
Skip to content
Projects
Groups
Snippets
Help
Loading...
Help
Submit feedback
Contribute to GitLab
Sign in
Toggle navigation
L
liquidnet-bus-v1
Project
Project
Details
Activity
Releases
Cycle Analytics
Repository
Repository
Files
Commits
Branches
Tags
Contributors
Graph
Compare
Charts
Issues
0
Issues
0
List
Board
Labels
Milestones
Merge Requests
0
Merge Requests
0
CI / CD
CI / CD
Pipelines
Jobs
Schedules
Charts
Wiki
Wiki
Snippets
Snippets
Members
Members
Collapse sidebar
Close sidebar
Activity
Graph
Charts
Create a new issue
Jobs
Commits
Issue Boards
Open sidebar
董敬伟
liquidnet-bus-v1
Commits
aa46fb4e
Commit
aa46fb4e
authored
Jun 15, 2022
by
zhanggb
Browse files
Options
Browse Files
Download
Email Patches
Plain Diff
~queue:GalaxyQueue.SQL_NFT_ORDER_INFO;
parent
835f8e14
Changes
5
Hide whitespace changes
Inline
Side-by-side
Showing
5 changed files
with
162 additions
and
124 deletions
+162
-124
ConsumerCommonSqlRedisStreamConfig.java
...sumer/base/config/ConsumerCommonSqlRedisStreamConfig.java
+48
-30
ConsumerGalaxySqlNftOrderInfoReceiver.java
.../base/receiver/ConsumerGalaxySqlNftOrderInfoReceiver.java
+20
-0
ConsumerGalaxySqlNftOrderInfoReceiver.java
...kylin/receiver/ConsumerGalaxySqlNftOrderInfoReceiver.java
+20
-20
ConsumerGalaxySqlNftOrderInfoConfig.java
...er/slowly/config/ConsumerGalaxySqlNftOrderInfoConfig.java
+54
-54
ConsumerGalaxySqlNftOrderInfoReceiver.java
...lowly/receiver/ConsumerGalaxySqlNftOrderInfoReceiver.java
+20
-20
No files found.
liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-base/src/main/java/com/liquidnet/service/consumer/base/config/ConsumerCommonSqlRedisStreamConfig.java
View file @
aa46fb4e
...
...
@@ -74,6 +74,8 @@ public class ConsumerCommonSqlRedisStreamConfig extends RedisStreamConfig {
ConsumerGalaxySqlUserInfoReceiver
consumerGalaxySqlUserInfoReceiver
;
@Autowired
ConsumerGalaxySqlSeriesInfoReceiver
consumerGalaxySqlSeriesInfoReceiver
;
@Autowired
ConsumerGalaxySqlNftOrderInfoReceiver
consumerGalaxySqlNftOrderInfoReceiver
;
/*------sweet------*/
@Autowired
ConsumerSweetAnswerRdsReceiver
consumerSweetAnswerRdsReceiver
;
...
...
@@ -116,7 +118,7 @@ public class ConsumerCommonSqlRedisStreamConfig extends RedisStreamConfig {
/*------sweet------*/
@Bean
// 关注服务号的用户信息
@Bean
public
List
<
Subscription
>
subscriptionSweetSqlWechatUser
(
RedisConnectionFactory
factory
)
{
List
<
Subscription
>
subscriptionList
=
new
ArrayList
<>();
MQConst
.
SweetQueue
stream
=
MQConst
.
SweetQueue
.
SWEET_WECHAT_USER_INSERT
;
...
...
@@ -132,7 +134,7 @@ public class ConsumerCommonSqlRedisStreamConfig extends RedisStreamConfig {
return
subscriptionList
;
}
@Bean
// 用户-关系
@Bean
public
List
<
Subscription
>
subscriptionSweetSqlArtistsRelation
(
RedisConnectionFactory
factory
)
{
List
<
Subscription
>
subscriptionList
=
new
ArrayList
<>();
MQConst
.
SweetQueue
stream
=
MQConst
.
SweetQueue
.
ARTISTS_RELATION
;
...
...
@@ -148,7 +150,7 @@ public class ConsumerCommonSqlRedisStreamConfig extends RedisStreamConfig {
return
subscriptionList
;
}
@Bean
// 用户-关系
@Bean
public
List
<
Subscription
>
subscriptionSweetSqlUserMdskRelation
(
RedisConnectionFactory
factory
)
{
List
<
Subscription
>
subscriptionList
=
new
ArrayList
<>();
MQConst
.
SweetQueue
stream
=
MQConst
.
SweetQueue
.
ARTISTS_RELATION_MDSK
;
...
...
@@ -164,7 +166,7 @@ public class ConsumerCommonSqlRedisStreamConfig extends RedisStreamConfig {
return
subscriptionList
;
}
@Bean
// 用户-抽奖
@Bean
public
List
<
Subscription
>
subscriptionSweetSqlLuckDraw
(
RedisConnectionFactory
factory
)
{
List
<
Subscription
>
subscriptionList
=
new
ArrayList
<>();
MQConst
.
SweetQueue
stream
=
MQConst
.
SweetQueue
.
LUCK_DRAW
;
...
...
@@ -180,7 +182,7 @@ public class ConsumerCommonSqlRedisStreamConfig extends RedisStreamConfig {
return
subscriptionList
;
}
@Bean
// 提醒记录
@Bean
public
List
<
Subscription
>
subscriptionSweetSqlRemind
(
RedisConnectionFactory
factory
)
{
List
<
Subscription
>
subscriptionList
=
new
ArrayList
<>();
MQConst
.
SweetQueue
stream
=
MQConst
.
SweetQueue
.
SWEET_REMIND_INSERT
;
...
...
@@ -196,7 +198,7 @@ public class ConsumerCommonSqlRedisStreamConfig extends RedisStreamConfig {
return
subscriptionList
;
}
@Bean
// 用户喜欢的音乐人
@Bean
public
List
<
Subscription
>
subscriptionSweetSqlPoster
(
RedisConnectionFactory
factory
)
{
List
<
Subscription
>
subscriptionList
=
new
ArrayList
<>();
MQConst
.
SweetQueue
stream
=
MQConst
.
SweetQueue
.
SWEET_POSTER
;
...
...
@@ -212,7 +214,7 @@ public class ConsumerCommonSqlRedisStreamConfig extends RedisStreamConfig {
return
subscriptionList
;
}
@Bean
// 猫登天空
@Bean
public
List
<
Subscription
>
subscriptionSweetSqlMaoDeng
(
RedisConnectionFactory
factory
)
{
List
<
Subscription
>
subscriptionList
=
new
ArrayList
<>();
MQConst
.
SweetQueue
stream
=
MQConst
.
SweetQueue
.
SWEET_WECHAT_MAODENG_TEXT_INSERT
;
...
...
@@ -228,7 +230,7 @@ public class ConsumerCommonSqlRedisStreamConfig extends RedisStreamConfig {
return
subscriptionList
;
}
@Bean
// 积分抽奖
@Bean
public
List
<
Subscription
>
subscriptionSweetSqlIntegralActivityDraw
(
RedisConnectionFactory
factory
)
{
List
<
Subscription
>
subscriptionList
=
new
ArrayList
<>();
MQConst
.
SweetQueue
stream
=
MQConst
.
SweetQueue
.
SWEET_INTEGRAL_ACTIVITY_DRAW
;
...
...
@@ -244,7 +246,7 @@ public class ConsumerCommonSqlRedisStreamConfig extends RedisStreamConfig {
return
subscriptionList
;
}
@Bean
// 用户投票记录
@Bean
public
List
<
Subscription
>
subscriptionSweetSqlCityVote
(
RedisConnectionFactory
factory
)
{
List
<
Subscription
>
subscriptionList
=
new
ArrayList
<>();
MQConst
.
SweetQueue
stream
=
MQConst
.
SweetQueue
.
SWEET_CITY_VOTE
;
...
...
@@ -260,7 +262,7 @@ public class ConsumerCommonSqlRedisStreamConfig extends RedisStreamConfig {
return
subscriptionList
;
}
@Bean
// 防疫答题
@Bean
public
List
<
Subscription
>
subscriptionSweetSqlAntigenicQuestion
(
RedisConnectionFactory
factory
)
{
List
<
Subscription
>
subscriptionList
=
new
ArrayList
<>();
MQConst
.
SweetQueue
stream
=
MQConst
.
SweetQueue
.
SWEET_ANTIGENIC_QUESTION
;
...
...
@@ -276,7 +278,7 @@ public class ConsumerCommonSqlRedisStreamConfig extends RedisStreamConfig {
return
subscriptionList
;
}
@Bean
// 答案
@Bean
public
List
<
Subscription
>
subscriptionSweetSqlAnswer
(
RedisConnectionFactory
factory
)
{
List
<
Subscription
>
subscriptionList
=
new
ArrayList
<>();
MQConst
.
SweetQueue
stream
=
MQConst
.
SweetQueue
.
SWEET_ANSWER_INSERT
;
...
...
@@ -294,7 +296,7 @@ public class ConsumerCommonSqlRedisStreamConfig extends RedisStreamConfig {
/*------goblin------*/
@Bean
// NFT订单
@Bean
public
List
<
Subscription
>
subscriptionGoblinSqlNftOrder
(
RedisConnectionFactory
factory
)
{
List
<
Subscription
>
subscriptionList
=
new
ArrayList
<>();
MQConst
.
GoblinQueue
stream
=
MQConst
.
GoblinQueue
.
GOBLIN_NFT_ORDER
;
...
...
@@ -310,7 +312,7 @@ public class ConsumerCommonSqlRedisStreamConfig extends RedisStreamConfig {
return
subscriptionList
;
}
@Bean
// 店铺相关
@Bean
public
List
<
Subscription
>
subscriptionGoblinSqlStore
(
RedisConnectionFactory
factory
)
{
List
<
Subscription
>
subscriptionList
=
new
ArrayList
<>();
MQConst
.
GoblinQueue
stream
=
MQConst
.
GoblinQueue
.
SQL_STORE
;
...
...
@@ -326,7 +328,7 @@ public class ConsumerCommonSqlRedisStreamConfig extends RedisStreamConfig {
return
subscriptionList
;
}
@Bean
// 商品相关
@Bean
public
List
<
Subscription
>
subscriptionGoblinSqlGoods
(
RedisConnectionFactory
factory
)
{
List
<
Subscription
>
subscriptionList
=
new
ArrayList
<>();
MQConst
.
GoblinQueue
stream
=
MQConst
.
GoblinQueue
.
SQL_GOODS
;
...
...
@@ -342,7 +344,7 @@ public class ConsumerCommonSqlRedisStreamConfig extends RedisStreamConfig {
return
subscriptionList
;
}
@Bean
// 店铺优惠券相关
@Bean
public
List
<
Subscription
>
subscriptionGoblinSqlStoreCoupon
(
RedisConnectionFactory
factory
)
{
List
<
Subscription
>
subscriptionList
=
new
ArrayList
<>();
MQConst
.
GoblinQueue
stream
=
MQConst
.
GoblinQueue
.
SQL_COUPON
;
...
...
@@ -358,7 +360,7 @@ public class ConsumerCommonSqlRedisStreamConfig extends RedisStreamConfig {
return
subscriptionList
;
}
@Bean
// 藏品生成
@Bean
public
List
<
Subscription
>
subscriptionGoblinSqlArtworkGen
(
RedisConnectionFactory
factory
)
{
List
<
Subscription
>
subscriptionList
=
new
ArrayList
<>();
MQConst
.
GoblinQueue
stream
=
MQConst
.
GoblinQueue
.
SQL_ARTWORK_GEN
;
...
...
@@ -376,7 +378,7 @@ public class ConsumerCommonSqlRedisStreamConfig extends RedisStreamConfig {
/*------adam------*/
@Bean
// 用户注册
@Bean
public
List
<
Subscription
>
subscriptionAdamSqlURegister
(
RedisConnectionFactory
factory
)
{
List
<
Subscription
>
subscriptionList
=
new
ArrayList
<>();
MQConst
.
AdamQueue
stream
=
MQConst
.
AdamQueue
.
SQL_UREGISTER
;
...
...
@@ -392,7 +394,7 @@ public class ConsumerCommonSqlRedisStreamConfig extends RedisStreamConfig {
return
subscriptionList
;
}
@Bean
// 用户中心
@Bean
public
List
<
Subscription
>
subscriptionAdamSqlUCenter
(
RedisConnectionFactory
factory
)
{
List
<
Subscription
>
subscriptionList
=
new
ArrayList
<>();
MQConst
.
AdamQueue
stream
=
MQConst
.
AdamQueue
.
SQL_UCENTER
;
...
...
@@ -408,7 +410,7 @@ public class ConsumerCommonSqlRedisStreamConfig extends RedisStreamConfig {
return
subscriptionList
;
}
@Bean
// 购买会员
@Bean
public
List
<
Subscription
>
subscriptionAdamSqlUMember
(
RedisConnectionFactory
factory
)
{
List
<
Subscription
>
subscriptionList
=
new
ArrayList
<>();
MQConst
.
AdamQueue
stream
=
MQConst
.
AdamQueue
.
SQL_UMEMBER
;
...
...
@@ -426,7 +428,7 @@ public class ConsumerCommonSqlRedisStreamConfig extends RedisStreamConfig {
/*------kylin------*/
@Bean
// 验票更新
@Bean
public
List
<
Subscription
>
subscriptionKylinSqlStation
(
RedisConnectionFactory
factory
)
{
List
<
Subscription
>
subscriptionList
=
new
ArrayList
<>();
MQConst
.
KylinQueue
stream
=
MQConst
.
KylinQueue
.
SQL_STATION
;
...
...
@@ -442,7 +444,7 @@ public class ConsumerCommonSqlRedisStreamConfig extends RedisStreamConfig {
return
subscriptionList
;
}
@Bean
// 用户退款发快递
@Bean
public
List
<
Subscription
>
subscriptionKylinSqlOrderRefundExpress
(
RedisConnectionFactory
factory
)
{
List
<
Subscription
>
subscriptionList
=
new
ArrayList
<>();
MQConst
.
KylinQueue
stream
=
MQConst
.
KylinQueue
.
SQL_ORDER_REFUND_EXPRESS
;
...
...
@@ -460,7 +462,7 @@ public class ConsumerCommonSqlRedisStreamConfig extends RedisStreamConfig {
/*------slime------*/
@Bean
// 场地相关
@Bean
public
List
<
Subscription
>
subscriptionSlimeSqlField
(
RedisConnectionFactory
factory
)
{
List
<
Subscription
>
subscriptionList
=
new
ArrayList
<>();
MQConst
.
SlimeQueue
stream
=
MQConst
.
SlimeQueue
.
SQL_SLIME_FIELD
;
...
...
@@ -476,7 +478,7 @@ public class ConsumerCommonSqlRedisStreamConfig extends RedisStreamConfig {
return
subscriptionList
;
}
@Bean
// 主办相关
@Bean
public
List
<
Subscription
>
subscriptionSlimeSqlSponsor
(
RedisConnectionFactory
factory
)
{
List
<
Subscription
>
subscriptionList
=
new
ArrayList
<>();
MQConst
.
SlimeQueue
stream
=
MQConst
.
SlimeQueue
.
SQL_SLIME_SPONSOR
;
...
...
@@ -492,7 +494,7 @@ public class ConsumerCommonSqlRedisStreamConfig extends RedisStreamConfig {
return
subscriptionList
;
}
@Bean
// 票种相关
@Bean
public
List
<
Subscription
>
subscriptionSlimeSqlTicketUpdate
(
RedisConnectionFactory
factory
)
{
List
<
Subscription
>
subscriptionList
=
new
ArrayList
<>();
MQConst
.
SlimeQueue
stream
=
MQConst
.
SlimeQueue
.
SQL_SLIME_TICKET_UPDATE
;
...
...
@@ -508,7 +510,7 @@ public class ConsumerCommonSqlRedisStreamConfig extends RedisStreamConfig {
return
subscriptionList
;
}
@Bean
// 演出相关
@Bean
public
List
<
Subscription
>
subscriptionSlimeSqlPerformanceUpdate
(
RedisConnectionFactory
factory
)
{
List
<
Subscription
>
subscriptionList
=
new
ArrayList
<>();
MQConst
.
SlimeQueue
stream
=
MQConst
.
SlimeQueue
.
SQL_SLIME_PERFORMANCE_UPDATE
;
...
...
@@ -524,7 +526,7 @@ public class ConsumerCommonSqlRedisStreamConfig extends RedisStreamConfig {
return
subscriptionList
;
}
@Bean
// 修改演出第二部
@Bean
public
List
<
Subscription
>
subscriptionSlimeSqlStep2
(
RedisConnectionFactory
factory
)
{
List
<
Subscription
>
subscriptionList
=
new
ArrayList
<>();
MQConst
.
SlimeQueue
stream
=
MQConst
.
SlimeQueue
.
SQL_SLIME_STEP2
;
...
...
@@ -540,7 +542,7 @@ public class ConsumerCommonSqlRedisStreamConfig extends RedisStreamConfig {
return
subscriptionList
;
}
@Bean
// 上下线相关
@Bean
public
List
<
Subscription
>
subscriptionSlimeSqlLine
(
RedisConnectionFactory
factory
)
{
List
<
Subscription
>
subscriptionList
=
new
ArrayList
<>();
MQConst
.
SlimeQueue
stream
=
MQConst
.
SlimeQueue
.
SQL_SLIME_LINE
;
...
...
@@ -556,7 +558,7 @@ public class ConsumerCommonSqlRedisStreamConfig extends RedisStreamConfig {
return
subscriptionList
;
}
@Bean
// 创建演出
@Bean
public
List
<
Subscription
>
subscriptionSlimeSqlPerformanceInsert
(
RedisConnectionFactory
factory
)
{
List
<
Subscription
>
subscriptionList
=
new
ArrayList
<>();
MQConst
.
SlimeQueue
stream
=
MQConst
.
SlimeQueue
.
SQL_SLIME_PERFORMANCE_INSERT
;
...
...
@@ -572,7 +574,7 @@ public class ConsumerCommonSqlRedisStreamConfig extends RedisStreamConfig {
return
subscriptionList
;
}
@Bean
// 用户注册信息
@Bean
public
List
<
Subscription
>
subscriptionGalaxySqlUserInfo
(
RedisConnectionFactory
factory
)
{
List
<
Subscription
>
subscriptionList
=
new
ArrayList
<>();
MQConst
.
GalaxyQueue
stream
=
MQConst
.
GalaxyQueue
.
SQL_USER_INFO
;
...
...
@@ -588,7 +590,7 @@ public class ConsumerCommonSqlRedisStreamConfig extends RedisStreamConfig {
return
subscriptionList
;
}
@Bean
// 系列信息
@Bean
public
List
<
Subscription
>
subscriptionGalaxySqlSeriesInfo
(
RedisConnectionFactory
factory
)
{
List
<
Subscription
>
subscriptionList
=
new
ArrayList
<>();
MQConst
.
GalaxyQueue
stream
=
MQConst
.
GalaxyQueue
.
SQL_SERIES_INFO
;
...
...
@@ -603,4 +605,20 @@ public class ConsumerCommonSqlRedisStreamConfig extends RedisStreamConfig {
}
return
subscriptionList
;
}
@Bean
public
List
<
Subscription
>
subscriptionGalaxySqlNftOrderInfo
(
RedisConnectionFactory
factory
)
{
List
<
Subscription
>
subscriptionList
=
new
ArrayList
<>();
MQConst
.
GalaxyQueue
stream
=
MQConst
.
GalaxyQueue
.
SQL_NFT_ORDER_INFO
;
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
()),
consumerGalaxySqlNftOrderInfoReceiver
));
listenerContainer
.
start
();
}
return
subscriptionList
;
}
}
liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-base/src/main/java/com/liquidnet/service/consumer/base/receiver/ConsumerGalaxySqlNftOrderInfoReceiver.java
0 → 100644
View file @
aa46fb4e
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
ConsumerGalaxySqlNftOrderInfoReceiver
extends
AbstractSqlRedisReceiver
{
@Override
protected
String
getRedisStreamKey
()
{
return
MQConst
.
GalaxyQueue
.
SQL_NFT_ORDER_INFO
.
getKey
();
}
@Override
protected
String
getRedisStreamGroup
()
{
return
MQConst
.
GalaxyQueue
.
SQL_NFT_ORDER_INFO
.
getGroup
();
}
}
liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-kylin/src/main/java/com/liquidnet/service/consumer/kylin/receiver/ConsumerGalaxySqlNftOrderInfoReceiver.java
View file @
aa46fb4e
package
com
.
liquidnet
.
service
.
consumer
.
kylin
.
receiver
;
import
com.liquidnet.service.base.constant.MQConst
;
import
lombok.extern.slf4j.Slf4j
;
import
org.springframework.stereotype.Component
;
@Slf4j
@Component
public
class
ConsumerGalaxySqlNftOrderInfoReceiver
extends
AbstractSqlRedisReceiver
{
@Override
protected
String
getRedisStreamKey
()
{
return
MQConst
.
GalaxyQueue
.
SQL_NFT_ORDER_INFO
.
getKey
();
}
@Override
protected
String
getRedisStreamGroup
()
{
return
MQConst
.
GalaxyQueue
.
SQL_NFT_ORDER_INFO
.
getGroup
();
}
}
//
package com.liquidnet.service.consumer.kylin.receiver;
//
//
import com.liquidnet.service.base.constant.MQConst;
//
import lombok.extern.slf4j.Slf4j;
//
import org.springframework.stereotype.Component;
//
//
@Slf4j
//
@Component
//
public class ConsumerGalaxySqlNftOrderInfoReceiver extends AbstractSqlRedisReceiver {
//
//
@Override
//
protected String getRedisStreamKey() {
//
return MQConst.GalaxyQueue.SQL_NFT_ORDER_INFO.getKey();
//
}
//
//
@Override
//
protected String getRedisStreamGroup() {
//
return MQConst.GalaxyQueue.SQL_NFT_ORDER_INFO.getGroup();
//
}
//
}
liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-slowly/src/main/java/com/liquidnet/service/consumer/slowly/config/ConsumerGalaxySqlNftOrderInfoConfig.java
View file @
aa46fb4e
package
com
.
liquidnet
.
service
.
consumer
.
slowly
.
config
;
import
com.liquidnet.common.cache.redis.config.RedisStreamConfig
;
import
com.liquidnet.service.base.constant.MQConst
;
import
com.liquidnet.service.consumer.slowly.receiver.ConsumerGalaxySqlNftOrderInfoReceiver
;
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
;
/**
* @author AnJiabin <anjiabin@zhengzai.tv>
* @version V1.0
* @Description: 区块链nft订单相关
* @class: ConsumerSqlNftOrderInfoConfig
* @Package com.liquidnet.service.consumer.galaxy.config
* @Copyright: LightNet @ Copyright (c) 2022
* @date 2022/3/25 14:45
*/
@Configuration
public
class
ConsumerGalaxySqlNftOrderInfoConfig
extends
RedisStreamConfig
{
@Autowired
private
ConsumerGalaxySqlNftOrderInfoReceiver
sqlNftOrderInfoReceiver
;
@Autowired
StringRedisTemplate
stringRedisTemplate
;
@Bean
public
List
<
Subscription
>
subscriptionSqlNftOrderInfo
(
RedisConnectionFactory
factory
)
{
List
<
Subscription
>
subscriptionList
=
new
ArrayList
<>();
MQConst
.
GalaxyQueue
stream
=
MQConst
.
GalaxyQueue
.
SQL_NFT_ORDER_INFO
;
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
()),
sqlNftOrderInfoReceiver
));
listenerContainer
.
start
();
}
return
subscriptionList
;
}
}
//
package com.liquidnet.service.consumer.slowly.config;
//
//
import com.liquidnet.common.cache.redis.config.RedisStreamConfig;
//
import com.liquidnet.service.base.constant.MQConst;
//
import com.liquidnet.service.consumer.slowly.receiver.ConsumerGalaxySqlNftOrderInfoReceiver;
//
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;
//
/
//
**
//
* @author AnJiabin <anjiabin@zhengzai.tv>
//
* @version V1.0
//
* @Description: 区块链nft订单相关
//
* @class: ConsumerSqlNftOrderInfoConfig
//
* @Package com.liquidnet.service.consumer.galaxy.config
//
* @Copyright: LightNet @ Copyright (c) 2022
//
* @date 2022/3/25 14:45
//
*/
//
@Configuration
//
public class ConsumerGalaxySqlNftOrderInfoConfig extends RedisStreamConfig {
//
@Autowired
//
private ConsumerGalaxySqlNftOrderInfoReceiver sqlNftOrderInfoReceiver;
//
//
@Autowired
//
StringRedisTemplate stringRedisTemplate;
//
//
@Bean
//
public List<Subscription> subscriptionSqlNftOrderInfo(RedisConnectionFactory factory) {
//
List<Subscription> subscriptionList = new ArrayList<>();
//
MQConst.GalaxyQueue stream = MQConst.GalaxyQueue.SQL_NFT_ORDER_INFO;
//
//
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()), sqlNftOrderInfoReceiver
//
));
//
listenerContainer.start();
//
}
//
return subscriptionList;
//
}
//
}
liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-slowly/src/main/java/com/liquidnet/service/consumer/slowly/receiver/ConsumerGalaxySqlNftOrderInfoReceiver.java
View file @
aa46fb4e
package
com
.
liquidnet
.
service
.
consumer
.
slowly
.
receiver
;
import
com.liquidnet.service.base.constant.MQConst
;
import
lombok.extern.slf4j.Slf4j
;
import
org.springframework.stereotype.Component
;
@Slf4j
@Component
public
class
ConsumerGalaxySqlNftOrderInfoReceiver
extends
AbstractSqlRedisReceiver
{
@Override
protected
String
getRedisStreamKey
()
{
return
MQConst
.
GalaxyQueue
.
SQL_NFT_ORDER_INFO
.
getKey
();
}
@Override
protected
String
getRedisStreamGroup
()
{
return
MQConst
.
GalaxyQueue
.
SQL_NFT_ORDER_INFO
.
getGroup
();
}
}
//
package com.liquidnet.service.consumer.slowly.receiver;
//
//
import com.liquidnet.service.base.constant.MQConst;
//
import lombok.extern.slf4j.Slf4j;
//
import org.springframework.stereotype.Component;
//
//
@Slf4j
//
@Component
//
public class ConsumerGalaxySqlNftOrderInfoReceiver extends AbstractSqlRedisReceiver {
//
//
@Override
//
protected String getRedisStreamKey() {
//
return MQConst.GalaxyQueue.SQL_NFT_ORDER_INFO.getKey();
//
}
//
//
@Override
//
protected String getRedisStreamGroup() {
//
return MQConst.GalaxyQueue.SQL_NFT_ORDER_INFO.getGroup();
//
}
//
}
Write
Preview
Markdown
is supported
0%
Try again
or
attach a new file
Attach a file
Cancel
You are about to add
0
people
to the discussion. Proceed with caution.
Finish editing this message first!
Cancel
Please
register
or
sign in
to comment