|
|
|
@@ -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<String>{ |
|
|
|
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); |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
} |