补充提交配置文件

src/main/java/com/yunfeiyun/agmp/iotm/device/service/impl/IotDeviceServiceImpl.java

@@ -6,9 +6,9 @@ import com.yunfeiyun.agmp.common.utils.SecurityUtils;
 import com.yunfeiyun.agmp.iot.common.domain.IotDevice;
 import com.yunfeiyun.agmp.iotm.device.mapper.IotDeviceMapper;
 import com.yunfeiyun.agmp.iotm.device.service.IIotDeviceService;
-import com.yunfeiyun.agmp.iotm.mq.provider.IotMqProviderService;
+
 import com.yunfeiyun.agmp.iotm.mq.service.AgmpTosMsgService;
-import com.yunfeiyun.agmp.iotm.mq.service.SendToIotsMsgService;
+
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
 
@@ -27,8 +27,7 @@ public class IotDeviceServiceImpl implements IIotDeviceService
     private IotDeviceMapper iotDeviceMapper;
     @Resource
     private AgmpTosMsgService agmpTosMsgService;
-    @Resource
-    private SendToIotsMsgService sendToIotsMsgService;
+
 
     /**
      * 查询设备基础
@@ -49,8 +48,7 @@ public class IotDeviceServiceImpl implements IIotDeviceService
      * @return 设备基础
      */
     @Override
-    public List<IotDevice> selectIotDeviceList(IotDevice iotDevice)
-    {
+    public List<IotDevice> selectIotDeviceList(IotDevice iotDevice) {
         return iotDeviceMapper.selectIotDeviceList(iotDevice);
     }
 
@@ -66,7 +64,7 @@ public class IotDeviceServiceImpl implements IIotDeviceService
         iotDevice.setTid(SecurityUtils.getTid());
         int result =  iotDeviceMapper.insertIotDevice(iotDevice);
         agmpTosMsgService.sendIotDeviceInsertMsg(iotDevice);
-        sendToIotsMsgService.sendIotDeviceInsertMsg(iotDevice);
+
         return result;
     }
 

src/main/java/com/yunfeiyun/agmp/iotm/mq/bussiness/AgmpMqBusConfig.java

@@ -0,0 +1,98 @@
+package com.yunfeiyun.agmp.iotm.mq.bussiness;
+
+import com.yunfeiyun.agmp.common.framework.mq.rabbitmq.consts.MqAgmpConsts;
+import com.yunfeiyun.agmp.iot.common.constant.mq.IotMqExchange;
+import com.yunfeiyun.agmp.iot.common.constant.mq.IotMqQueue;
+
+import com.yunfeiyun.agmp.iotm.mq.listener.AgmpChannelAwareMessageListener;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.amqp.core.*;
+import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
+import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+import javax.annotation.PostConstruct;
+
+/**
+ * 智慧农业SaaS系统的MQ配置类
+ */
+@Configuration
+@Slf4j
+@ConditionalOnBean(name = "agmpMqConfig")
+public class AgmpMqBusConfig {
+
+    /**
+     * 注入AMQP管理员，用于声明队列、交换机和绑定
+     */
+    @Autowired
+    @Qualifier("agmpAmqpAdmin")
+    private AmqpAdmin agmpAmqpAdmin;
+
+    /**
+     * 初始化方法，在Spring容器初始化时声明队列、交换机和绑定
+     */
+    @PostConstruct
+    public void init() {
+        log.info("加载Agmp Mq");
+        agmpAmqpAdmin.declareQueue(agmpQueue());
+        agmpAmqpAdmin.declareExchange(agmpExchange());
+        agmpAmqpAdmin.declareBinding(agmpBinding());
+    }
+
+    /**
+     * 定义一个持久化的队列
+     *
+     * @return 队列对象
+     */
+    @Bean("agmpQueue")
+    public Queue agmpQueue() {
+        return QueueBuilder.durable(IotMqQueue.IOTM_TO_IOTS_CMD_QUEUE).build();
+    }
+
+    /**
+     * 定义一个主题交换机
+     *
+     * @return 交换机对象
+     */
+    @Bean("agmpExchange")
+    public FanoutExchange agmpExchange() {
+        return new FanoutExchange(IotMqExchange.IOTM_TO_IOTS_EXCHANGE);
+    }
+
+    /**
+     * 定义队列与交换机的绑定关系
+     *
+     * @return 绑定对象
+     */
+    @Bean("agmpBinding")
+    public Binding agmpBinding() {
+        return BindingBuilder.bind(agmpQueue()).to(agmpExchange());
+    }
+
+
+    /**
+     * 配置一个具体的消息监听器容器，用于监听特定的队列并处理消息
+     *
+     * @param connectionFactory 连接工厂
+     * @param listener          消息监听器
+     * @return 消息监听器容器
+     */
+    @Bean("agmpSimpleMessageListenerContainer")
+    public SimpleMessageListenerContainer agmpSimpleMessageListenerContainer(
+            @Qualifier("agmpConnectionFactory") CachingConnectionFactory connectionFactory,
+            AgmpChannelAwareMessageListener listener) {
+        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
+        container.setQueues(agmpQueue());
+        container.setMessageListener(listener);
+        // 设置确认模式为自动确认
+        container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
+        container.setAmqpAdmin(agmpAmqpAdmin);
+
+        return container;
+    }
+
+}

src/main/java/com/yunfeiyun/agmp/iotm/mq/listener/AgmpChannelAwareMessageListener.java

@@ -0,0 +1,34 @@
+package com.yunfeiyun.agmp.iotm.mq.listener;
+
+import com.rabbitmq.client.Channel;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.amqp.core.Message;
+import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
+import org.springframework.stereotype.Component;
+
+import java.io.IOException;
+
+/**
+ * 负责处理AGMP 子系统的消息
+ */
+@Component
+@Slf4j
+@ConditionalOnBean(name = "agmpMqConfig")
+public class AgmpChannelAwareMessageListener implements ChannelAwareMessageListener {
+
+    @Override
+    public void onMessage(Message message, Channel channel) throws Exception {
+        try {
+            // 处理消息
+            byte[] body = message.getBody();
+            String content = new String(body);
+            log.info("【SAAS:】收到AGMP消息：{}", content);
+            // 手动确认消息
+            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
+        } catch (IOException e) {
+            // 处理异常，例如重新入队或拒绝消息
+            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
+        }
+    }
+}

src/main/java/com/yunfeiyun/agmp/iotm/mq/provider/IotMqProviderService.java

@@ -1,37 +0,0 @@
-package com.yunfeiyun.agmp.iotm.mq.provider;
-
-import com.alibaba.fastjson2.JSONObject;
-import com.yunfeiyun.agmp.common.framework.mq.rabbitmq.consts.MqTosConsts;
-import com.yunfeiyun.agmp.common.framework.mq.rabbitmq.model.SynGlobalTenantInfoDto;
-import com.yunfeiyun.agmp.common.utils.JSONUtils;
-import com.yunfeiyun.agmp.iot.common.constant.mq.IotActionEnums;
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.amqp.rabbit.core.RabbitTemplate;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.beans.factory.annotation.Qualifier;
-import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
-import org.springframework.stereotype.Service;
-
-@Service
-@Slf4j
-@ConditionalOnBean(name = "iotMqConfig")
-public class IotMqProviderService {
-    @Autowired
-    @Qualifier("iotRabbitTemplate")
-    private RabbitTemplate iotRabbitTemplate;
-
-
-    /**
-     * 往TOS发送
-     *
-     * @param
-     */
-    public void sendToIots(IotActionEnums iotActionEnums, Object data, String desc) {
-        SynGlobalTenantInfoDto synGlobalTenantInfoDto = new SynGlobalTenantInfoDto();
-        synGlobalTenantInfoDto.setAction(iotActionEnums.getCode());
-        synGlobalTenantInfoDto.setDesc(desc);
-        synGlobalTenantInfoDto.setData(JSONObject.from(data));
-        iotRabbitTemplate.convertAndSend(MqTosConsts.ExchangeConsts.AGMP_TO_TOS_MSG_EXCHANGE, "", JSONUtils.toJSONString(synGlobalTenantInfoDto));
-    }
-
-}

src/main/java/com/yunfeiyun/agmp/iotm/mq/service/SendToIotsMsgService.java

@@ -1,50 +0,0 @@
-package com.yunfeiyun.agmp.iotm.mq.service;
-
-
-import com.yunfeiyun.agmp.iot.common.constant.mq.IotActionEnums;
-import com.yunfeiyun.agmp.iot.common.domain.IotDevice;
-import com.yunfeiyun.agmp.iotm.mq.provider.IotMqProviderService;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
-import org.springframework.stereotype.Service;
-
-/**
- * 往Iots发送
- *
- * @author liuyaowen
- */
-@Service
-@ConditionalOnBean(name = "iotMqConfig")
-public class SendToIotsMsgService {
-    @Autowired
-    private IotMqProviderService mqService;
-
-    /**
-     * 物联网新增设备，同步到iots ok
-     *
-     * @param iotDevice
-     */
-    public void sendIotDeviceInsertMsg(IotDevice iotDevice) {
-        mqService.sendToIots( IotActionEnums.IOT_DEVICE_CREATE, iotDevice, "【Iotm】to【Iots】物联网新增设备，同步到Iotm");
-
-    }
-
-    /**
-     * 物联网编辑设备，同步到iots  ok
-     *
-     * @param iotDevice
-     */
-    public void sendIotDeviceUpdateMsg(IotDevice iotDevice) {
-        mqService.sendToIots( IotActionEnums.IOT_DEVICE_UPDATE, iotDevice, "【Iotm】to【Iots】物联网编辑设备，同步到Iotm");
-
-    }
-
-    /**
-     * 物联网删除设备，同步iots  ok
-     *
-     * @param iotDevice
-     */
-    public void sendIotDeviceDeleteMsg(IotDevice iotDevice) {
-        mqService.sendToIots( IotActionEnums.IOT_DEVICE_DELETE, iotDevice, "【Iotm】to【Iots】物联网删除设备，同步到Iotm");
-    }
-}

src/main/resources/application-dev.yml

@@ -140,7 +140,7 @@ spring:
       connection-timeout: 15000
       publisher-returns: true
       enabled: true
-    iot:
+    agmp:
       host: 192.168.1.228
       port: 5672
       username: admin