From b831782a4f15d4c87c99095c0bc7592f0cf5befa Mon Sep 17 00:00:00 2001 From: luozukai Date: Thu, 7 Mar 2019 19:08:49 +0800 Subject: [PATCH] =?UTF-8?q?[=E6=B6=88=E6=81=AF=E7=BB=84=E4=BB=B6][?= =?UTF-8?q?=E6=B7=BB=E5=8A=A0][=E6=94=B9=E9=80=A0,=E7=BB=9F=E4=B8=80?= =?UTF-8?q?=E6=9B=BF=E6=8D=A2=E5=8E=9F=E6=9D=A5=E5=8F=91=E9=80=81=E6=96=B9?= =?UTF-8?q?=E5=BC=8F]?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../iformall/mq/impl/RocketMqConsumer.java | 47 +++++-------------- pom.xml | 4 +- 2 files changed, 13 insertions(+), 38 deletions(-) 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