Procházet zdrojové kódy

更新

release_toaliyun_real
xiaohanzi před 6 roky
rodič
revize
0139657efe
2 změnil soubory, kde provedl 18 přidání a 15 odebrání
  1. +5
    -3
      mallinkMQConsumer/src/main/java/com/iformall/mq/MqBaseConsumer.java
  2. +13
    -12
      mallinkMQConsumer/src/main/java/com/iformall/mq/impl/AliyunRocketMqConsumer.java

+ 5
- 3
mallinkMQConsumer/src/main/java/com/iformall/mq/MqBaseConsumer.java Zobrazit soubor

@@ -101,9 +101,11 @@ public class MqBaseConsumer {
UpdateCouponStockMsg msg = (UpdateCouponStockMsg)JsonUtil.readValue(message,UpdateCouponStockMsg.class); UpdateCouponStockMsg msg = (UpdateCouponStockMsg)JsonUtil.readValue(message,UpdateCouponStockMsg.class);
updateCouponStockMsgService.send(msg); updateCouponStockMsgService.send(msg);
} }
baseMsg.setMsgStatus(EnumMsgRecordStatus.CONSUME_SUCC.getCode());
wxMsgRecordMapper.updateByBaseMsg(baseMsg);
//baseMsg.setMsgStatus(EnumMsgRecordStatus.CONSUME_SUCC.getCode());
//wxMsgRecordMapper.updateByBaseMsg(baseMsg);
}catch (Throwable e){ }catch (Throwable e){
log.error("consum received error: ", e);
/**
try { try {
log.error("consum received error: ", e); log.error("consum received error: ", e);
baseMsg.setStatusMessage(e.getMessage()); baseMsg.setStatusMessage(e.getMessage());
@@ -129,7 +131,7 @@ public class MqBaseConsumer {
} }
}catch (Throwable e1){ }catch (Throwable e1){
log.error("更新db状态error: ", e1); log.error("更新db状态error: ", e1);
}
}*/
} }


} }


+ 13
- 12
mallinkMQConsumer/src/main/java/com/iformall/mq/impl/AliyunRocketMqConsumer.java Zobrazit soubor

@@ -41,8 +41,6 @@ public class AliyunRocketMqConsumer extends MqBaseConsumer {
@Value("${spring.aliyunRocketmq.namesrvAddr}") @Value("${spring.aliyunRocketmq.namesrvAddr}")
private String namesrvAddr; private String namesrvAddr;
private static Consumer consumer = null;
@PostConstruct @PostConstruct
public void init() { public void init() {
Properties properties = new Properties(); Properties properties = new Properties();
@@ -58,18 +56,21 @@ public class AliyunRocketMqConsumer extends MqBaseConsumer {
// properties.put(PropertyKeyConst.MessageModel, PropertyValueConst.CLUSTERING); // properties.put(PropertyKeyConst.MessageModel, PropertyValueConst.CLUSTERING);
// 广播订阅方式 // 广播订阅方式
// properties.put(PropertyKeyConst.MessageModel, PropertyValueConst.BROADCASTING); // properties.put(PropertyKeyConst.MessageModel, PropertyValueConst.BROADCASTING);
consumer = ONSFactory.createConsumer(properties);
consumer.subscribe(EnumMsgMqTopic.DEFAULT.getCode(), "*", new MessageListener() { //订阅多个 Tag
public Action consume(Message message, ConsumeContext context) {
System.out.println("Receive: " + message);
doMessage(new String(message.getBody()));
return Action.CommitMessage;
}
});
consumer.start();
log.info(" aliyunrocketMq consumer start success! "+JSON.toJSONString(consumer));
for (EnumMsgMqTopic topic: EnumMsgMqTopic.values()) {
Consumer consumer = ONSFactory.createConsumer(properties);
consumer.subscribe(topic.getCode(), "*", new MessageListener() { //订阅多个 Tag
public Action consume(Message message, ConsumeContext context) {
System.out.println("Receive: " + message);
doMessage(new String(message.getBody()));
return Action.CommitMessage;
}
});
consumer.start();
log.info(" aliyunrocketMq consumer start success! "+JSON.toJSONString(consumer));
}
} }
} }

Načítá se…
Zrušit
Uložit