diff --git a/mallinkMQConsumer/src/main/java/com/iformall/mq/impl/RocketMqConsumer.java b/mallinkMQConsumer/src/main/java/com/iformall/mq/impl/RocketMqConsumer.java index a100c8dd6..0508f1df7 100644 --- a/mallinkMQConsumer/src/main/java/com/iformall/mq/impl/RocketMqConsumer.java +++ b/mallinkMQConsumer/src/main/java/com/iformall/mq/impl/RocketMqConsumer.java @@ -1,44 +1,19 @@ -package com.iformall.mq; +package com.iformall.mq.impl; -import com.alibaba.druid.support.json.JSONUtils; -import com.iformall.domain.po.WxMsgRecord; -import com.iformall.enums.EnumMsgRecordStatus; -import com.iformall.mapper.WxMsgRecordMapper; +import com.iformall.mq.MqBaseConsumer; +import com.iformall.mq.MQConfig; import lombok.extern.slf4j.Slf4j; -import org.apache.rocketmq.spring.annotation.RocketMQMessageListener; -import org.apache.rocketmq.spring.core.RocketMQListener; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; -import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.stereotype.Service; +import org.rocketmq.starter.annotation.RocketMQListener; +import org.rocketmq.starter.annotation.RocketMQMessage; +import org.springframework.context.annotation.Profile; @Slf4j -@Service -@RocketMQMessageListener(topic = "topic-1",consumerGroup = "DefaultConsumer") -public class MqConsumer implements RocketMQListener{ - private final Logger log = LoggerFactory.getLogger(this.getClass()); - @Autowired - private WxMsgRecordMapper wxMsgRecordMapper; +@Profile(MQConfig.Impl.ROCKET_MQ) +@RocketMQListener(topic = "topic-1") +public class RocketMqConsumer extends MqBaseConsumer { + @RocketMQMessage(messageClass = String.class, tag = "tag-1") public void onMessage(String message) { - WxMsgRecord wxMsgRecord = null; - try { - log.info("received message: {}", message); - wxMsgRecord = (WxMsgRecord)JSONUtils.parse(message); - - - wxMsgRecord.setStatus(EnumMsgRecordStatus.CONSUME_SUCC.getCode()); - wxMsgRecordMapper.update(wxMsgRecord); - }catch (Throwable e){ - e.printStackTrace(); - if(null != wxMsgRecord){ - wxMsgRecord.setStatusMessage(e.getMessage()); - wxMsgRecord.setStatus(EnumMsgRecordStatus.CONSUME_FAIL.getCode()); - wxMsgRecordMapper.update(wxMsgRecord); - } - } + doMessage(message); } - - - } diff --git a/pom.xml b/pom.xml index 4c0b37338..612e9dcda 100644 --- a/pom.xml +++ b/pom.xml @@ -122,9 +122,9 @@ - org.apache.rocketmq + org.rocketmq.spring.boot rocketmq-spring-boot-starter - 2.0.1 + 1.0.0.RELEASE org.apache.rocketmq