|
@@ -1,6 +1,8 @@
|
|
|
package com.persagy.dmp.starter.alarm.communication.mq;
|
|
|
|
|
|
+import com.persagy.dmp.starter.alarm.communication.mq.model.DmpMessage;
|
|
|
import com.persagy.dmp.starter.alarm.communication.netty.NettyAlarmMsgBaseHandler;
|
|
|
+import com.persagy.dmp.starter.alarm.service.OrderStateChangeService;
|
|
|
import com.persagy.dmp.starter.alarm.util.StringUtil;
|
|
|
import com.rabbitmq.client.Channel;
|
|
|
import lombok.extern.slf4j.Slf4j;
|
|
@@ -11,6 +13,9 @@ import org.springframework.beans.factory.annotation.Autowired;
|
|
|
import org.springframework.beans.factory.annotation.Value;
|
|
|
import org.springframework.context.annotation.Bean;
|
|
|
import org.springframework.context.annotation.Configuration;
|
|
|
+import com.persagy.dmp.alarm.jms.model.OrderStateMessage;
|
|
|
+
|
|
|
+import java.io.IOException;
|
|
|
|
|
|
/**
|
|
|
* @description:报警定义消息通知
|
|
@@ -22,63 +27,111 @@ import org.springframework.context.annotation.Configuration;
|
|
|
@Slf4j
|
|
|
@Configuration
|
|
|
public class JmsConfig {
|
|
|
+ @Autowired
|
|
|
+ OrderStateChangeService orderStateChangeService;
|
|
|
+
|
|
|
+ /**
|
|
|
+ * NettyAlarmMsgBaseHandler本身不进行ioc注入,这里注入的是他的子类。
|
|
|
+ * 子类在实际项目中创建,starter中没有实例。
|
|
|
+ */
|
|
|
+ @Autowired
|
|
|
+ private NettyAlarmMsgBaseHandler msgHandler;
|
|
|
+ /**
|
|
|
+ * 报警定义变化类型
|
|
|
+ */
|
|
|
+ private static final String ALARM_CONFIGS_CHANGE = "alarmConfigsChange";
|
|
|
+ /**
|
|
|
+ * 报警交换器
|
|
|
+ */
|
|
|
+ @Value("${dmp.alarm.exchange}")
|
|
|
+ private String dmpAlarmExchange;
|
|
|
+ /**
|
|
|
+ * 报警路由键
|
|
|
+ */
|
|
|
+ @Value("${dmp.alarm.routingKey}")
|
|
|
+ private String alarmRoutingKey;
|
|
|
+ /**
|
|
|
+ * 报警队列
|
|
|
+ */
|
|
|
+ @Value("${dmp.alarm.queue}")
|
|
|
+ private String alarmQueue;
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 工单状态变化exchange, 下一期升级为对接集成框架
|
|
|
+ */
|
|
|
+ private String orderStateExchange = "workorder_state_publish_exchange";
|
|
|
+ /**
|
|
|
+ * 工单状态变化queue
|
|
|
+ */
|
|
|
+ private String orderStateQueue = "order_state_queue";
|
|
|
+
|
|
|
+
|
|
|
+ @Bean
|
|
|
+ public Queue alarmQueue() {
|
|
|
+ return new Queue(alarmQueue, true);
|
|
|
+ }
|
|
|
+
|
|
|
+ @Bean
|
|
|
+ public TopicExchange alarmExchange() {
|
|
|
+ return new TopicExchange(dmpAlarmExchange);
|
|
|
+ }
|
|
|
|
|
|
- /**
|
|
|
- * NettyAlarmMsgBaseHandler本身不进行ioc注入,这里注入的是他的子类。
|
|
|
- * 子类在实际项目中创建,starter中没有实例。
|
|
|
- */
|
|
|
- @Autowired
|
|
|
- private NettyAlarmMsgBaseHandler msgHandler;
|
|
|
- /**
|
|
|
- * 报警定义变化类型
|
|
|
- */
|
|
|
- private static final String ALARM_CONFIGS_CHANGE = "alarmConfigsChange";
|
|
|
- /**
|
|
|
- * 报警交换器
|
|
|
- */
|
|
|
- @Value("${dmp.alarm.exchange}")
|
|
|
- private String exchange;
|
|
|
- /**
|
|
|
- * 报警路由键
|
|
|
- */
|
|
|
- @Value("${dmp.alarm.routingKey}")
|
|
|
- private String alarmRoutingKey;
|
|
|
- /**
|
|
|
- * 报警队列
|
|
|
- */
|
|
|
- @Value("${dmp.alarm.queue}")
|
|
|
- private String alarmQueue;
|
|
|
+ @Bean
|
|
|
+ public Binding alarmBinding() {
|
|
|
+ return BindingBuilder.bind(alarmQueue()).to(alarmExchange()).with(alarmRoutingKey);
|
|
|
+ }
|
|
|
|
|
|
+ @Bean
|
|
|
+ public FanoutExchange orderStateExchange() {
|
|
|
+ return new FanoutExchange(orderStateExchange);
|
|
|
+ }
|
|
|
|
|
|
- @Bean
|
|
|
- public Queue queue() {
|
|
|
- return new Queue(alarmQueue, true);
|
|
|
- }
|
|
|
+ @Bean
|
|
|
+ public Queue orderStateQueue() {
|
|
|
+ return new Queue(orderStateQueue, true);
|
|
|
+ }
|
|
|
|
|
|
- @Bean
|
|
|
- public TopicExchange exchange() {
|
|
|
- return new TopicExchange(exchange);
|
|
|
- }
|
|
|
+ @Bean
|
|
|
+ public Binding orderStateBinding() {
|
|
|
+ return BindingBuilder.bind(orderStateQueue()).to(orderStateExchange());
|
|
|
+ }
|
|
|
|
|
|
- @Bean
|
|
|
- public Binding alarmBinding() {
|
|
|
- return BindingBuilder.bind(queue()).to(exchange()).with(alarmRoutingKey);
|
|
|
- }
|
|
|
+ @RabbitListener(queues = "order_state_queue") //监听器监听指定的Queue
|
|
|
+ public void processOrderState(String message, Channel channel, Message msg) {
|
|
|
+ log.info("============================== Receive:" + message);
|
|
|
+ try {
|
|
|
+ OrderStateMessage orderStateMessage = StringUtil.transferItemToDTO(message, OrderStateMessage.class);
|
|
|
+ log.info("order_id: {}", orderStateMessage.getOrder_id());
|
|
|
+ log.info("order_state: {}", orderStateMessage.getOrder_state());
|
|
|
+ // 根据工单状态消息更新报警记录状态
|
|
|
+ orderStateChangeService.updateAlarmWhenOrderStateChange(orderStateMessage);
|
|
|
+ // 手动确认消息已消费
|
|
|
+ channel.basicAck(msg.getMessageProperties().getDeliveryTag(),false);
|
|
|
+ } catch (Exception e) {
|
|
|
+ log.error("工单状态消息消费失败: {}", e.getMessage());
|
|
|
+ try {
|
|
|
+ // 将消费失败的消息移除消息队列,避免重复消费
|
|
|
+ channel.basicReject(msg.getMessageProperties().getDeliveryTag(),false);
|
|
|
+ } catch (IOException ex) {
|
|
|
+ log.error("工单状态消息从队列中移除失败,{}", ex.getMessage());
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
|
|
|
- @RabbitHandler
|
|
|
- @RabbitListener(queues = "${dmp.alarm.queue}")
|
|
|
- public void planQueues(String msg, Channel channel, Message message) throws Exception {
|
|
|
- log.info("============================== Receive:" + msg);
|
|
|
- DmpMessage dmpMessage = StringUtil.transferItemToDTO(msg, DmpMessage.class);
|
|
|
- //报警定义变化
|
|
|
- if(ALARM_CONFIGS_CHANGE.equals(dmpMessage.getType())){
|
|
|
- log.info("================收到一条报警定义变化通知==============");
|
|
|
- log.info(msg.toString());
|
|
|
- try {
|
|
|
- msgHandler.incrementSyncAlarmConfig(dmpMessage);
|
|
|
- } catch (Exception e) {
|
|
|
- log.error("error",e);
|
|
|
- }
|
|
|
- }
|
|
|
- }
|
|
|
+ @RabbitHandler
|
|
|
+ @RabbitListener(queues = "${dmp.alarm.queue}")
|
|
|
+ public void planQueues(String msg, Channel channel, Message message) throws Exception {
|
|
|
+ log.info("============================== Receive:" + msg);
|
|
|
+ DmpMessage dmpMessage = StringUtil.transferItemToDTO(msg, DmpMessage.class);
|
|
|
+ //报警定义变化
|
|
|
+ if (ALARM_CONFIGS_CHANGE.equals(dmpMessage.getType())) {
|
|
|
+ log.info("================收到一条报警定义变化通知==============");
|
|
|
+ log.info(msg.toString());
|
|
|
+ try {
|
|
|
+ msgHandler.incrementSyncAlarmConfig(dmpMessage);
|
|
|
+ } catch (Exception e) {
|
|
|
+ log.error("error", e);
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
}
|