NB_NetPump/coffee-system/src/main/java/com/coffee/aliyun/AliyunConsumerGroupService.java

package com.coffee.aliyun;

import cn.hutool.extra.spring.SpringUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.coffee.aliyun.utils.Constants;
import com.coffee.aliyun.utils.Items;
import com.coffee.bus.bean.AliIotConfig;
import com.coffee.bus.entity.BusDeviceEntity;
import com.coffee.bus.entity.BusDeviceRunningEntity;
import com.coffee.bus.entity.BusInfusionHistoryEntity;
import com.coffee.bus.listener.event.bean.DeviceInfoEvent;
import com.coffee.bus.service.LocalBusDeviceService;
import com.coffee.bus.service.LocalBusInfusionHistoryService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.jms.Message;
import javax.jms.MessageListener;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
* @Author longsanlang
* @Date 2022-4-06 16:22:13
* @Version 1.0
* @Description 阿里云物联网平台服务端订阅
*/
@Service
@Slf4j
public class AliyunConsumerGroupService {

    // 初始化订阅客户端
    private final AliyunIotSubscribeClient client;

    // 设备Service
    private final LocalBusDeviceService deviceService;

    // 输注Service
    private final LocalBusInfusionHistoryService infusionHistoryService;

    public AliyunConsumerGroupService(AliyunIotSubscribeClient client,
                                      LocalBusDeviceService deviceService,
                                      LocalBusInfusionHistoryService infusionHistoryService) {
        this.client = client;
        this.deviceService = deviceService;
        this.infusionHistoryService = infusionHistoryService;
    }

    @Value("${aliyun.server-subscription.enable:false}")
    private boolean isEnable;

    // 开启服务端订阅
    @PostConstruct
    public void subscribe(){
        if (!isEnable){
            log.info("订阅禁止");
            return;
        }
        log.info("允许开启订阅");
        try {
            // 开启订阅
            client.start(messageListener);
        } catch (Exception e) {
            e.printStackTrace();
        }
        log.info("阿里云物联网订阅成功。。。。。。。。。。。");
    }



    //业务处理异步线程池，线程池参数可以根据您的业务特点调整，或者您也可以用其他异步方式处理接收到的消息。
    private final static ExecutorService executorService = new ThreadPoolExecutor(
            Runtime.getRuntime().availableProcessors(),
            Runtime.getRuntime().availableProcessors() * 2, 60, TimeUnit.SECONDS,
            new LinkedBlockingQueue(50000));

    private MessageListener messageListener = (message) -> {
            try {
                //1.收到消息之后一定要ACK。
                // 推荐做法：创建Session选择Session.AUTO_ACKNOWLEDGE，这里会自动ACK。
                // 其他做法：创建Session选择Session.CLIENT_ACKNOWLEDGE，这里一定要调message.acknowledge()来ACK。
                // message.acknowledge();
                //2.建议异步处理收到的消息，确保onMessage函数里没有耗时逻辑。
                // 如果业务处理耗时过程过长阻塞住线程，可能会影响SDK收到消息后的正常回调。
                executorService.submit(()-> processMessage(message));
            } catch (Exception e) {
                log.error("submit task occurs exception ", e);
            }
    };

    /**
     * 处理逻辑，加上锁
     * @param message
     */
    private static final String IOTID = "iotId";
    private static final String PRODUCTKEY = "productKey";
    private static final String DEVICENAME = "deviceName";
    private static final String TOPIC = "topic";
    private static final String MESSAGEID = "messageId";
    private static final String ITEMS = "items";
    private static final String VALUE = "value";
    private static final String STATUS = "status";
    private static final String CONTENT = "content";

    private void processMessage(Message message) {

        PlatformLog platformLog = new PlatformLog();
        Map<String, Object> platformData = new HashMap<>();
        try {
            // 获取主题，消息id和内容
            String topic = message.getStringProperty(TOPIC);
            String messageId = message.getStringProperty(MESSAGEID);
            JSONObject content = JSON.parseObject(new String(message.getBody(byte[].class)));
            // 平台数据
            platformData.put(TOPIC,topic);
            platformData.put(MESSAGEID,messageId);
            platformData.put(CONTENT,content);
            log.info("阿里云物联网发送的数据："+JSON.toJSONString(platformData));

            // 设备名称
            String deviceName = content.getString(DEVICENAME);
            // 创建设备对象
            Date now = new Date();
            // 根据topic判断数据类型
            if (topic.matches("^/as/mqtt/status/[\\w\\/]*")){//设备上下线状态
                String status = content.getString(STATUS);
                if (status.equals("online")){
                    log.info(deviceName+"设备上线");
                }else if (status.equals("offline")){
                    log.info(deviceName+"设备下线");
                }else {
                    log.info(deviceName+"设备未激活");
                }
            }else if (topic.matches("[\\w\\/]*event/property/post$")){//设备属性上报
                /**
                 * platformData:
                 * {
                 * "topic":"/a1M7k1TAECc/ceshi001/thing/event/property/post",
                 * "messageId":"1496410088574460416",
                 * "content":"{\"deviceType\":\"CustomCategory\",
                 * \"iotId\":\"0syOzMqwkGebZBGKa08d000000\",
                 * \"requestId\":\"254\",\"checkFailedData\":{},
                 * \"productKey\":\"a1M7k1TAECc\",\"gmtCreate\":1645606941721,
                 * \"deviceName\":\"ceshi001\",
                 * \"items\":{
                 * \"total\":{\"value\":100,\"time\":1645606941717},
                 * \"pumpCode\":{\"value\":\"5719512336330299\",\"time\":1645606941717},
                 * \"dataNumber\":{\"value\":254,\"time\":1645606941717},
                 * \"PCAInvalid\":{\"value\":0,\"time\":1645606941717},
                 * \"alarm\":{\"value\":0,\"time\":1645606941717},
                 * \"flowRate\":{\"value\":2,\"time\":1645606941717},
                 * \"PCAValid\":{\"value\":0,\"time\":1645606941717},
                 * \"forecast\":{\"value\":0,\"time\":1645606941717},
                 * \"finished\":{\"value\":17,\"time\":1645606941717},
                 * \"battery\":{\"value\":65,\"time\":1645606941717},
                 * \"runStatus\":{\"value\":2,\"time\":1645606941717},
                 * \"patientCode\":{\"value\":111110000,\"time\":1645606941717}}}"
                 * }
                 */
                // 设备属性集合
                Items items = new Items(content.getJSONObject("items"));
                BusInfusionHistoryEntity infusionHistory = infusionHistoryService.saveInfusion(deviceName,items);


                // 发布设备信息事件。处理完毕，传向下一步处理
                DeviceInfoEvent deviceInfoEvent = new DeviceInfoEvent(this,new BusDeviceRunningEntity(),"123456");
                SpringUtil.publishEvent(deviceInfoEvent);

            }else if(topic.matches("[\\w\\/]+thing/lifecycle$")){// 设备生命周期
                // 获取生命周期类型
                String action = content.getString("action");
                if ("create".equals(action)){
                    // 创建设备
                    BusDeviceEntity device = new BusDeviceEntity();
                    device.setDeviceId(deviceName);
                    device.setCreateBy(Constants.DefaultCreateBy);
                    device.setUpdateBy(Constants.DefaultUpdateBy);
                    device.setUpdateTime(now);
                    device.setTenantId(Constants.DefaultHospital);

                    // 配置信息
                    AliIotConfig config = new AliIotConfig();
                    config.setDeviceName(deviceName);
                    config.setDeviceSecret(content.getString("deviceSecret"));
                    config.setIotId(content.getString("iotId"));
                    config.setProductKey(content.getString("productKey"));

                    // 设置配置信息
                    device.setConfig(config);
                    deviceService.saveByDeviceId(device);
                }else if ("delete".equals(action)){
                    // 删除设备
                    deviceService.removeByDeviceId(deviceName);
                }



            }else {
                log.error("未知的topic："+topic);
            }

        } catch (Exception e) {
            // 错误标志
            platformLog.setIsError("1");
            platformLog.setErrorLog(e.getMessage());
            log.error("数据处理失败 ", e);
        }
        // 存储日志

    }







}