From fdb7874406aaf4c2c5d7cf8d7dc571873e7665b4 Mon Sep 17 00:00:00 2001 From: luozukai Date: Thu, 7 Mar 2019 18:49:57 +0800 Subject: [PATCH] aa --- .../controller/WxFlowAbleController.java | 2 +- .../main/java/com/iformall/mq/MqProducer.java | 11 +++++------ .../com/iformall/mq/impl/RocketMqConsumer.java | 17 +++++++++++++++++ 3 files changed, 23 insertions(+), 7 deletions(-) diff --git a/mallinkAdmin/src/main/java/com/iformall/controller/WxFlowAbleController.java b/mallinkAdmin/src/main/java/com/iformall/controller/WxFlowAbleController.java index a391d8d3d..83a5ac38a 100644 --- a/mallinkAdmin/src/main/java/com/iformall/controller/WxFlowAbleController.java +++ b/mallinkAdmin/src/main/java/com/iformall/controller/WxFlowAbleController.java @@ -31,7 +31,7 @@ public class WxFlowAbleController extends BaseController { @ApiOperation(value = "mqSend",notes = "") @PostMapping("/mqSend") - public ResultData mqSend() { + public ResultData mqSend() throws Throwable{ mqProducer.sendMessage("aaaa","topic-1","tag-1","key1"); return new ResultData(); } diff --git a/mallinkService/src/main/java/com/iformall/mq/MqProducer.java b/mallinkService/src/main/java/com/iformall/mq/MqProducer.java index 7906984cf..bb8c68801 100644 --- a/mallinkService/src/main/java/com/iformall/mq/MqProducer.java +++ b/mallinkService/src/main/java/com/iformall/mq/MqProducer.java @@ -30,6 +30,8 @@ public class MqProducer { public void start(){ try { producer.setNamesrvAddr(namesrvAddr); + //producer.setRetryTimesWhenSendFailed(5);//重试机制,默认2次 + //producer.setSendMsgTimeout(6000); //默认是3000 producer.start(); log.info("Message Producer Start..."); }catch (Exception e){ @@ -37,7 +39,7 @@ public class MqProducer { } } - public void sendMessage(String data, String topic, String tags, String keys) { + public void sendMessage(String data, String topic, String tags, String keys) throws Throwable{ try { byte[] messageBody = data.getBytes(RemotingHelper.DEFAULT_CHARSET); Message mqMsg = new Message(topic, tags, keys, messageBody); @@ -49,14 +51,11 @@ public class MqProducer { @Override public void onException(Throwable throwable) { - - - log.error("Message Producer: Send Message Error ", throwable); } }); - } catch (Exception e) { - log.error("Message Producer: Send Message Error ", e); + } catch (Throwable e) { + e.printStackTrace(); } } } diff --git a/mallinkService/src/main/java/com/iformall/mq/impl/RocketMqConsumer.java b/mallinkService/src/main/java/com/iformall/mq/impl/RocketMqConsumer.java index b480a4906..61a7e67a1 100644 --- a/mallinkService/src/main/java/com/iformall/mq/impl/RocketMqConsumer.java +++ b/mallinkService/src/main/java/com/iformall/mq/impl/RocketMqConsumer.java @@ -3,6 +3,7 @@ package com.iformall.mq.impl; import com.iformall.mq.MQConfig; import com.iformall.mq.MqBaseConsumer; import lombok.extern.slf4j.Slf4j; +import org.apache.rocketmq.common.message.MessageExt; import org.rocketmq.starter.annotation.RocketMQListener; import org.rocketmq.starter.annotation.RocketMQMessage; import org.springframework.context.annotation.Profile; @@ -12,8 +13,24 @@ import org.springframework.context.annotation.Profile; @RocketMQListener(topic = "topic-1") public class RocketMqConsumer extends MqBaseConsumer { +<<<<<<< HEAD:mallinkService/src/main/java/com/iformall/mq/impl/RocketMqConsumer.java @RocketMQMessage(messageClass = String.class, tag = "tag-1") public void onMessage(String message) { doMessage(message); +======= + @RocketMQMessage(messageClass = MessageExt.class, tag = "tag-1") + public void onMessage(MessageExt message) { + try { + log.info("received message: {}", message); + +// System.out.println("------"+message.getProperties().get("PROPERTY_RECONSUME_TIME")); + +// System.out.println(1/0); +// return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; + }catch (Throwable e){ + e.printStackTrace(); + } +// return ConsumeConcurrentlyStatus.RECONSUME_LATER; +>>>>>>> [铜锣湾][修改][预账单]:mallinkService/src/main/java/com/iformall/mq/MqConsumer.java } }