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

Skip to content

L
liquidnet-bus-v1
Commit a30ba6c6 authored by anjiabin's avatar anjiabin
Browse files
Options

修改redis消费队列方式

parent 52a77436
Hide whitespace changes
Inline Side-by-side
Showing with 25 additions and 23 deletions
liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-dragon/src/main/java/com/liquidnet/service/consumer/dragon/receiver/AbstractRedisReceiver.java
View file @ a30ba6c6
...@@ -37,16 +37,18 @@ public abstract class AbstractRedisReceiver implements StreamListener<String, Ma ...@@ -37,16 +37,18 @@ public abstract class AbstractRedisReceiver implements StreamListener<String, Ma
log.info("body " + message.getValue()); log.info("body " + message.getValue());
boolean result = this.consumerSqlDaoHandler(message.getValue().get("message")); boolean result = this.consumerSqlDaoHandler(message.getValue().get("message"));
// 消费成功确认，消息删除和消息确认是一个事务 if(result){
log.info("consumer success delete message messageId:{} ",message.getId()); // 消费成功确认，消息删除和消息确认是一个事务
try { log.info("consumer success delete message messageId:{} ",message.getId());
try {
// stringRedisTemplate.multi(); // stringRedisTemplate.multi();
stringRedisTemplate.opsForStream().delete(this.getRedisStreamKey(), message.getId()); stringRedisTemplate.opsForStream().delete(this.getRedisStreamKey(), message.getId());
// stringRedisTemplate.opsForStream().acknowledge(getRedisStreamGroup(), message); // stringRedisTemplate.opsForStream().acknowledge(getRedisStreamGroup(), message);
// stringRedisTemplate.exec(); // stringRedisTemplate.exec();
} catch (Exception e) { } catch (Exception e) {
e.printStackTrace(); e.printStackTrace();
log.error("delete redis queue message error messageId:{} errMsg:{}",message.getId(),e.getMessage()); log.error("delete redis queue message error messageId:{} errMsg:{}",message.getId(),e.getMessage());
}
} }
} }
...@@ -59,8 +61,6 @@ public abstract class AbstractRedisReceiver implements StreamListener<String, Ma ...@@ -59,8 +61,6 @@ public abstract class AbstractRedisReceiver implements StreamListener<String, Ma
log.debug("CONSUMER SQL result of execution:{}", rstBatchSqls); log.debug("CONSUMER SQL result of execution:{}", rstBatchSqls);
if (rstBatchSqls) { if (rstBatchSqls) {
return true; return true;
}else{
sendMySqlRedis(msg);
} }
} catch (Exception e) { } catch (Exception e) {
e.printStackTrace(); e.printStackTrace();
......
liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-dragon/src/main/java/com/liquidnet/service/consumer/dragon/receiver/RedisRefundReceiver.java
View file @ a30ba6c6
...@@ -27,23 +27,25 @@ public class RedisRefundReceiver implements StreamListener<String, MapRecord<Str ...@@ -27,23 +27,25 @@ public class RedisRefundReceiver implements StreamListener<String, MapRecord<Str
@Override @Override
public void onMessage(MapRecord<String, String, String> message) { public void onMessage(MapRecord<String, String, String> message) {
log.info("接受到来自redis REFUND 的消息"); log.info("接受到来自redis REFUND 的消息");
System.out.println("message id " + message.getId()); log.info("message id " + message.getId());
System.out.println("stream " + message.getStream()); log.info("stream " + message.getStream());
System.out.println("body " + message.getValue()); log.info("body " + message.getValue());
consumerSqlDaoHandler(message.getValue().get("message")); boolean result = this.consumerSqlDaoHandler(message.getValue().get("message"));
log.error("consumer success delete message messageId:{} ",message.getId()); if(result){
try { log.error("consumer success delete message messageId:{} ",message.getId());
try {
// stringRedisTemplate.multi(); // stringRedisTemplate.multi();
stringRedisTemplate.opsForStream().delete(DragonConstant.MysqlRedisQueueEnum.DRAGON_REFUND_KEY.getCode(), message.getId()); stringRedisTemplate.opsForStream().delete(DragonConstant.MysqlRedisQueueEnum.DRAGON_REFUND_KEY.getCode(), message.getId());
// stringRedisTemplate.opsForStream().acknowledge(getRedisStreamGroup(), message); // stringRedisTemplate.opsForStream().acknowledge(getRedisStreamGroup(), message);
// stringRedisTemplate.exec(); // stringRedisTemplate.exec();
} catch (Exception e) { } catch (Exception e) {
log.error("delete redis queue message Exception error: {} ",e); log.error("delete redis queue message Exception error: {} ",e);
log.error("delete redis queue message error messageId:{} errMsg:{}",message.getId(),e.getMessage()); log.error("delete redis queue message error messageId:{} errMsg:{}",message.getId(),e.getMessage());
}
} }
} }
private void consumerSqlDaoHandler(String msg) { private boolean consumerSqlDaoHandler(String msg) {
try { try {
SqlMapping.SqlMessage sqlMessage = JsonUtils.fromJson(msg, SqlMapping.SqlMessage.class); SqlMapping.SqlMessage sqlMessage = JsonUtils.fromJson(msg, SqlMapping.SqlMessage.class);
log.debug("CONSUMER SQL ==> Preparing:{}", JsonUtils.toJson(sqlMessage.getSqls())); log.debug("CONSUMER SQL ==> Preparing:{}", JsonUtils.toJson(sqlMessage.getSqls()));
...@@ -51,13 +53,13 @@ public class RedisRefundReceiver implements StreamListener<String, MapRecord<Str ...@@ -51,13 +53,13 @@ public class RedisRefundReceiver implements StreamListener<String, MapRecord<Str
Boolean rstBatchSqls = baseDao.batchSqls(sqlMessage.getSqls(), sqlMessage.getArgs()); Boolean rstBatchSqls = baseDao.batchSqls(sqlMessage.getSqls(), sqlMessage.getArgs());
log.debug("CONSUMER SQL result of execution:{}", rstBatchSqls); log.debug("CONSUMER SQL result of execution:{}", rstBatchSqls);
if (rstBatchSqls) { if (rstBatchSqls) {
//应答 return true;
} else {
sendMySqlRedis(msg);
} }
} catch (Exception e) { } catch (Exception e) {
e.printStackTrace(); e.printStackTrace();
log.error("CONSUMER SQL Exception error:{}", e);
} }
return false;
} }
/** /**
......
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