luozukai 7 лет назад
Родитель
Сommit
fdb7874406
3 измененных файлов: 23 добавлений и 7 удалений
  1. +1
    -1
      mallinkAdmin/src/main/java/com/iformall/controller/WxFlowAbleController.java
  2. +5
    -6
      mallinkService/src/main/java/com/iformall/mq/MqProducer.java
  3. +17
    -0
      mallinkService/src/main/java/com/iformall/mq/impl/RocketMqConsumer.java

+ 1
- 1
mallinkAdmin/src/main/java/com/iformall/controller/WxFlowAbleController.java Просмотреть файл

@@ -31,7 +31,7 @@ public class WxFlowAbleController extends BaseController {


@ApiOperation(value = "mqSend",notes = "") @ApiOperation(value = "mqSend",notes = "")
@PostMapping("/mqSend") @PostMapping("/mqSend")
public ResultData mqSend() {
public ResultData mqSend() throws Throwable{
mqProducer.sendMessage("aaaa","topic-1","tag-1","key1"); mqProducer.sendMessage("aaaa","topic-1","tag-1","key1");
return new ResultData(); return new ResultData();
} }


+ 5
- 6
mallinkService/src/main/java/com/iformall/mq/MqProducer.java Просмотреть файл

@@ -30,6 +30,8 @@ public class MqProducer {
public void start(){ public void start(){
try { try {
producer.setNamesrvAddr(namesrvAddr); producer.setNamesrvAddr(namesrvAddr);
//producer.setRetryTimesWhenSendFailed(5);//重试机制,默认2次
//producer.setSendMsgTimeout(6000); //默认是3000
producer.start(); producer.start();
log.info("Message Producer Start..."); log.info("Message Producer Start...");
}catch (Exception e){ }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 { try {
byte[] messageBody = data.getBytes(RemotingHelper.DEFAULT_CHARSET); byte[] messageBody = data.getBytes(RemotingHelper.DEFAULT_CHARSET);
Message mqMsg = new Message(topic, tags, keys, messageBody); Message mqMsg = new Message(topic, tags, keys, messageBody);
@@ -49,14 +51,11 @@ public class MqProducer {


@Override @Override
public void onException(Throwable throwable) { public void onException(Throwable throwable) {



log.error("Message Producer: Send Message Error ", 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();
} }
} }
} }

+ 17
- 0
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.MQConfig;
import com.iformall.mq.MqBaseConsumer; import com.iformall.mq.MqBaseConsumer;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.message.MessageExt;
import org.rocketmq.starter.annotation.RocketMQListener; import org.rocketmq.starter.annotation.RocketMQListener;
import org.rocketmq.starter.annotation.RocketMQMessage; import org.rocketmq.starter.annotation.RocketMQMessage;
import org.springframework.context.annotation.Profile; import org.springframework.context.annotation.Profile;
@@ -12,8 +13,24 @@ import org.springframework.context.annotation.Profile;
@RocketMQListener(topic = "topic-1") @RocketMQListener(topic = "topic-1")
public class RocketMqConsumer extends MqBaseConsumer { public class RocketMqConsumer extends MqBaseConsumer {


<<<<<<< HEAD:mallinkService/src/main/java/com/iformall/mq/impl/RocketMqConsumer.java
@RocketMQMessage(messageClass = String.class, tag = "tag-1") @RocketMQMessage(messageClass = String.class, tag = "tag-1")
public void onMessage(String message) { public void onMessage(String message) {
doMessage(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
} }
} }

Загрузка…
Отмена
Сохранить