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

package com.coffee.aliyun;

import cn.hutool.core.text.CharSequenceUtil;
import cn.hutool.extra.spring.SpringUtil;
import cn.hutool.json.JSONUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.coffee.aliyun.utils.Constants;
import com.coffee.aliyun.utils.EnumUtils;
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.BusHospitalLogEntity;
import com.coffee.bus.enums.HospitalLogEnum;
import com.coffee.bus.listener.event.bean.DeviceInfoEvent;
import com.coffee.bus.service.LocalBusDeviceService;
import com.coffee.bus.service.LocalBusHospitalLogService;
import com.coffee.bus.websocket.listener.DeviceInfoListener;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Lazy;
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.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @Author 龙三郎
 * @Date 2022-4-06 16:22:13
 * @Version 1.0
 * @Description 阿里云物联网平台服务端订阅
 */
@Service
@Slf4j
public class AliyunConsumerGroupService {
    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";

    @Autowired
    @Lazy
    private AliyunIotSubscribeClient client;
    @Autowired
    @Lazy
    private LocalBusDeviceService deviceService;

    @Autowired
    private DeviceInfoListener deviceInfoListener;
    @Autowired
    @Lazy
    private LocalBusHospitalLogService hospitalLogService;
    @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),
            new AliYunThreadFactory());

    static class AliYunThreadFactory implements ThreadFactory {
        static final AtomicInteger poolNumber = new AtomicInteger(1);
        final ThreadGroup group;
        final AtomicInteger threadNumber = new AtomicInteger(1);
        final String namePrefix;

        AliYunThreadFactory() {
            SecurityManager s = System.getSecurityManager();
            group = (s != null)? s.getThreadGroup() :
                    Thread.currentThread().getThreadGroup();
            namePrefix = "ali-iot-" +
                    poolNumber.getAndIncrement();
        }
        @Override
        public Thread newThread(Runnable r) {
            Thread t = new Thread(group, r,
                    namePrefix + threadNumber.getAndIncrement(),
                    0);
            if (t.isDaemon()){
                t.setDaemon(false);
            }
            if (t.getPriority() != Thread.NORM_PRIORITY){
                t.setPriority(Thread.NORM_PRIORITY);
            }
            return t;
        }
    }

    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);
        }
    };

    private void processMessage(Message message) {


        BusHospitalLogEntity hospitalLog = new BusHospitalLogEntity();
        hospitalLog.setType(HospitalLogEnum.ALI);
        long startTime = System.currentTimeMillis();
        String deviceName=null;
        try {
            // 获取主题，消息id和内容
            String topic = message.getStringProperty(TOPIC);
            String messageId = message.getStringProperty(MESSAGEID);
            hospitalLog.setMsgId(messageId);
            JSONObject content = JSON.parseObject(new String(message.getBody(byte[].class)));
            log.info("阿里云物联网发送的数据："+JSON.toJSONString(content));

            // 设备名称
            deviceName = content.getString(DEVICENAME);

            hospitalLog.setIdentityCode(deviceName);
            hospitalLog.setInput(content.toJSONString());
            // 创建设备对象
            Date now = new Date();
            // 根据topic判断数据类型,设备上下线状态
            if (topic.matches("^/as/mqtt/status/[\\w\\/]*")){
                BusDeviceEntity device = new BusDeviceEntity();
                device.setDeviceId(deviceName);
                String status = content.getString(STATUS);
                device.setStatus(EnumUtils.getDeviceStatusEnum2(status));
                if ("online".equalsIgnoreCase(status)){
                    log.info(deviceName+"设备【{}】上线",deviceName);
                }else if ("offline".equalsIgnoreCase(status)){
                    log.info(deviceName+"设备【{}】下线",deviceName);
                }else {
                    log.info(deviceName+"设备【{}】未激活",deviceName);
                }
                // 更新设备状态
                deviceService.updateDevice(device);
            }else if (topic.matches("[\\w\\/]*event/property/post$")){//设备属性上报
                // 设备属性集合
                Items items = new Items(content.getJSONObject("items"));
                BusDeviceRunningEntity deviceRunning = new BusDeviceRunningEntity();
                deviceRunning.updateFieldsByItems(deviceName,items);
                deviceInfoListener.deviceInfoDetail(new DeviceInfoEvent(this,deviceRunning,deviceName));
            }else if(topic.matches("[\\w\\/]+thing/lifecycle$")){// 设备生命周期
                // 获取生命周期类型
                String action = content.getString("action");
                if ("create".equals(action)){
                    // 创建设备
                    BusDeviceEntity device = new BusDeviceEntity();
                    device.setDeviceId(deviceName);
                    device.setCreateTime(now);
                    device.setCreateBy(Constants.DefaultCreateBy);
                    device.setUpdateTime(now);
                    device.setUpdateBy(Constants.DefaultUpdateBy);
                    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.saveDevice(device);
                }else if ("delete".equals(action)){
                    // 删除设备
                    deviceService.removeByDeviceId(deviceName);
                }
            }else {
                log.warn("阿里云数据【{}】，未知的topic:【{}】",content.toJSONString(),topic);
            }
            hospitalLog.setSuccess(true);
        } catch (Exception e) {
            hospitalLog.setSuccess(false);
            hospitalLog.setMessage(e.toString());
            log.error("阿里云数据【{}】数据处理失败 ", JSONUtil.toJsonStr(message), e);
        }finally {
            if(CharSequenceUtil.isNotBlank(deviceName)){
                BusDeviceEntity device = deviceService.getByDeviceId(deviceName);
                long entTime = System.currentTimeMillis();
                hospitalLog.setTenantId(device.getTenantId());
                hospitalLog.setUseTime(entTime-startTime);
                hospitalLogService.save(hospitalLog);
            }
        }

    }





}