|
@@ -3,6 +3,7 @@ package com.persagy.dmp.alarm.jms;
|
|
|
import com.persagy.common.json.JacksonMapper;
|
|
|
import com.persagy.dmp.rwd.model.DmpMessage;
|
|
|
import com.rabbitmq.client.Channel;
|
|
|
+import lombok.SneakyThrows;
|
|
|
import lombok.extern.slf4j.Slf4j;
|
|
|
import org.springframework.amqp.core.*;
|
|
|
import org.springframework.amqp.rabbit.annotation.RabbitListener;
|
|
@@ -29,20 +30,13 @@ public class JmsConfig {
|
|
|
@Value("${persagy.dmp.alarm.routingKey}")
|
|
|
private String dmpAlarmRoutingKey;
|
|
|
|
|
|
- /**
|
|
|
- * 报警定义路由键
|
|
|
- */
|
|
|
- @Value("${persagy.dmp.alarm.config.routingKey}")
|
|
|
- private String dmpAlarmConfigRoutingKey;
|
|
|
-
|
|
|
@Value("${alarm.obj.routingKey}")
|
|
|
private String alarmObjRoutingKey;
|
|
|
|
|
|
@Value("${persagy.dmp.rwd.routingKey}")
|
|
|
private String rwdRoutingKey;
|
|
|
|
|
|
- @Value("${persagy.dmp.alarm.queue}")
|
|
|
- private String alarmQueue;
|
|
|
+ private String alarmQueue = "dmp-test-queue";
|
|
|
|
|
|
// @Value("${persagy.dmp.alarm.rwd.queue}")
|
|
|
// private String alarmRwdQueue;
|
|
@@ -90,14 +84,46 @@ public class JmsConfig {
|
|
|
return BindingBuilder.bind(alarmObjQueue()).to(exchange()).with(alarmObjRoutingKey);
|
|
|
}
|
|
|
|
|
|
- @RabbitListener(queues = {alarmObjQueue}) //监听器监听指定的Queue
|
|
|
- public void process(String message, Channel channel, Message msg) throws Exception{
|
|
|
+// @RabbitListener(queues = {alarmObjQueue}) //监听器监听指定的Queue
|
|
|
+// public void process(String message, Channel channel, Message msg) throws Exception{
|
|
|
+// // 手动确认消息已消费
|
|
|
+// try {
|
|
|
+// log.info("============================== Receive:" + message);
|
|
|
+// DmpMessage dmpMessage = JacksonMapper.toObject(message, DmpMessage.class);
|
|
|
+// if (dmpMessage != null) {
|
|
|
+// messageProcesser.listen(dmpMessage);
|
|
|
+// }
|
|
|
+//
|
|
|
+// channel.basicAck(msg.getMessageProperties().getDeliveryTag(), false);
|
|
|
+// } catch (Exception e) {
|
|
|
+// Boolean isRedeliveredFail = msg.getMessageProperties().getRedelivered();
|
|
|
+// // true表示消息已经重复处理失败
|
|
|
+// if (isRedeliveredFail) {
|
|
|
+// // 拒绝消息,requeue=false 表示不再重新入队,如果配置了死信队列则进入死信队列
|
|
|
+// log.error("重复消费消息失败,message: {}", message);
|
|
|
+// channel.basicReject(msg.getMessageProperties().getDeliveryTag(), false);
|
|
|
+// } else {
|
|
|
+// // 如果是第一次失败则再次放入队列
|
|
|
+// // requeue为是否重新回到队列,true重新入队
|
|
|
+// channel.basicNack(msg.getMessageProperties().getDeliveryTag(), false, true);
|
|
|
+// }
|
|
|
+// }
|
|
|
+// }
|
|
|
+
|
|
|
+// @RabbitListener(queues ="${persagy.dmp.alarm.rwd.queue}") //监听器监听指定的Queue
|
|
|
+// public void processRwdMsg(DmpMessage message, Channel channel, Message msg) {
|
|
|
+// consumeMessageThreadPool.consumeMessage(message, channel, msg);
|
|
|
+// }
|
|
|
+
|
|
|
+ @SneakyThrows
|
|
|
+ @RabbitListener(queues = "dmp-test-queue") //监听器监听指定的Queue
|
|
|
+ public void processAlarmMsg(DmpMessage message, Channel channel, Message msg) {
|
|
|
// 手动确认消息已消费
|
|
|
try {
|
|
|
log.info("============================== Receive:" + message);
|
|
|
- DmpMessage dmpMessage = JacksonMapper.toObject(message, DmpMessage.class);
|
|
|
- if (dmpMessage != null) {
|
|
|
- messageProcesser.listen(dmpMessage);
|
|
|
+ Thread.sleep(100);
|
|
|
+ if (message != null) {
|
|
|
+ consumeMessageThreadPool.consumeMessage(message);
|
|
|
}
|
|
|
|
|
|
channel.basicAck(msg.getMessageProperties().getDeliveryTag(), false);
|
|
@@ -115,14 +141,4 @@ public class JmsConfig {
|
|
|
}
|
|
|
}
|
|
|
}
|
|
|
-
|
|
|
-// @RabbitListener(queues ="${persagy.dmp.alarm.rwd.queue}") //监听器监听指定的Queue
|
|
|
-// public void processRwdMsg(DmpMessage message, Channel channel, Message msg) {
|
|
|
-// consumeMessageThreadPool.consumeMessage(message, channel, msg);
|
|
|
-// }
|
|
|
-
|
|
|
-// @RabbitListener(queues = "${persagy.dmp.alarm.queue}") //监听器监听指定的Queue
|
|
|
-// public void processAlarmMsg(DmpMessage message, Channel channel, Message msg) {
|
|
|
-// consumeMessageThreadPool.consumeMessage(message, channel, msg);
|
|
|
-// }
|
|
|
}
|