diff --git a/mallinkMQConsumer/src/main/java/com/iformall/mq/MqBaseConsumer.java b/mallinkMQConsumer/src/main/java/com/iformall/mq/MqBaseConsumer.java index 33df70929..390debf64 100644 --- a/mallinkMQConsumer/src/main/java/com/iformall/mq/MqBaseConsumer.java +++ b/mallinkMQConsumer/src/main/java/com/iformall/mq/MqBaseConsumer.java @@ -101,9 +101,11 @@ public class MqBaseConsumer { UpdateCouponStockMsg msg = (UpdateCouponStockMsg)JsonUtil.readValue(message,UpdateCouponStockMsg.class); updateCouponStockMsgService.send(msg); } - baseMsg.setMsgStatus(EnumMsgRecordStatus.CONSUME_SUCC.getCode()); - wxMsgRecordMapper.updateByBaseMsg(baseMsg); + //baseMsg.setMsgStatus(EnumMsgRecordStatus.CONSUME_SUCC.getCode()); + //wxMsgRecordMapper.updateByBaseMsg(baseMsg); }catch (Throwable e){ + log.error("consum received error: ", e); + /** try { log.error("consum received error: ", e); baseMsg.setStatusMessage(e.getMessage()); @@ -129,7 +131,7 @@ public class MqBaseConsumer { } }catch (Throwable e1){ log.error("更新db状态error: ", e1); - } + }*/ } } diff --git a/mallinkMQConsumer/src/main/java/com/iformall/mq/impl/AliyunRocketMqConsumer.java b/mallinkMQConsumer/src/main/java/com/iformall/mq/impl/AliyunRocketMqConsumer.java index 77827c01e..c4eb3d6d5 100644 --- a/mallinkMQConsumer/src/main/java/com/iformall/mq/impl/AliyunRocketMqConsumer.java +++ b/mallinkMQConsumer/src/main/java/com/iformall/mq/impl/AliyunRocketMqConsumer.java @@ -41,8 +41,6 @@ public class AliyunRocketMqConsumer extends MqBaseConsumer { @Value("${spring.aliyunRocketmq.namesrvAddr}") private String namesrvAddr; - private static Consumer consumer = null; - @PostConstruct public void init() { Properties properties = new Properties(); @@ -58,18 +56,21 @@ public class AliyunRocketMqConsumer extends MqBaseConsumer { // properties.put(PropertyKeyConst.MessageModel, PropertyValueConst.CLUSTERING); // 广播订阅方式 // 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)); + } + } }