class CatalogResponseMessageHandler extends SIPRequestProcessorParent implements InitializingBean, IMessageHandler {

    private Logger logger = LoggerFactory.getLogger(CatalogResponseMessageHandler.class);

    private final String cmdType = "Catalog";

    @Autowired
    private ResponseMessageHandler responseMessageHandler;

    private final ConcurrentLinkedQueue<HandlerCatchData> taskQueue = new ConcurrentLinkedQueue<>();

    @Autowired
    private IVideoManagerStorage storager;

    @Autowired
    private CatalogDataCatch catalogDataCatch;

    @Qualifier("taskExecutor")
    @Autowired
    private ThreadPoolTaskExecutor taskExecutor;

    @Autowired
    private SipConfig sipConfig;
    private AtomicBoolean processing = new AtomicBoolean(false);

    @Override
    public void afterPropertiesSet() throws Exception {
        responseMessageHandler.addHandler(cmdType, this);
    }

    @Override
    public void handForDevice(RequestEvent evt, Device device, Element element) {<FILL_FUNCTION_BODY>}

    @Override
    public void handForPlatform(RequestEvent evt, ParentPlatform parentPlatform, Element rootElement) {

    }

    public SyncStatus getChannelSyncProgress(String deviceId) {
        if (catalogDataCatch.get(deviceId) == null) {
            return null;
        } else {
            return catalogDataCatch.getSyncStatus(deviceId);
        }
    }

    public boolean isSyncRunning(String deviceId) {
        if (catalogDataCatch.get(deviceId) == null) {
            return false;
        } else {
            return catalogDataCatch.isSyncRunning(deviceId);
        }
    }

    public void setChannelSyncReady(Device device, int sn) {
        catalogDataCatch.addReady(device, sn);
    }

    public void setChannelSyncEnd(String deviceId, String errorMsg) {
        catalogDataCatch.setChannelSyncEnd(deviceId, errorMsg);
    }
}

        taskQueue.offer(new HandlerCatchData(evt, device, element));
        // 回复200 OK
        try {
            responseAck((SIPRequest) evt.getRequest(), Response.OK);
        } catch (SipException | InvalidArgumentException | ParseException e) {
            logger.error("[命令发送失败] 目录查询回复: {}", e.getMessage());
        }
        // 已经开启消息处理则跳过
        if (processing.compareAndSet(false, true)) {
            taskExecutor.execute(() -> {
                while (!taskQueue.isEmpty()) {
                    // 全局异常捕获，保证下一条可以得到处理
                    try {
                        HandlerCatchData take = taskQueue.poll();
                        Element rootElement = null;
                        try {
                            rootElement = getRootElement(take.getEvt(), take.getDevice().getCharset());
                        } catch (DocumentException e) {
                            logger.error("[xml解析] 失败： ", e);
                            continue;
                        }
                        if (rootElement == null) {
                            logger.warn("[ 收到通道 ] content cannot be null, {}", evt.getRequest());
                            continue;
                        }
                        Element deviceListElement = rootElement.element("DeviceList");
                        Element sumNumElement = rootElement.element("SumNum");
                        Element snElement = rootElement.element("SN");
                        int sumNum = Integer.parseInt(sumNumElement.getText());

                        if (sumNum == 0) {
                            logger.info("[收到通道]设备:{}的: 0个", take.getDevice().getDeviceId());
                            // 数据已经完整接收
                            storager.cleanChannelsForDevice(take.getDevice().getDeviceId());
                            catalogDataCatch.setChannelSyncEnd(take.getDevice().getDeviceId(), null);
                        } else {
                            Iterator<Element> deviceListIterator = deviceListElement.elementIterator();
                            if (deviceListIterator != null) {
                                List<DeviceChannel> channelList = new ArrayList<>();
                                List<String> parentChannelIds = new ArrayList<>();
                                // 遍历DeviceList
                                while (deviceListIterator.hasNext()) {
                                    Element itemDevice = deviceListIterator.next();
                                    Element channelDeviceElement = itemDevice.element("DeviceID");
                                    if (channelDeviceElement == null) {
                                        continue;
                                    }
                                    DeviceChannel channel = XmlUtil.channelContentHandler(itemDevice, device, null);
                                    if (channel == null) {
                                        logger.info("[收到目录订阅]：但是解析失败 {}", new String(evt.getRequest().getRawContent()));
                                        continue;
                                    }
                                    if (channel.getParentId() != null && channel.getParentId().equals(sipConfig.getId())) {
                                        channel.setParentId(null);
                                    }
                                    SipUtils.updateGps(channel, device.getGeoCoordSys());
                                    channel.setDeviceId(take.getDevice().getDeviceId());

                                    channelList.add(channel);
                                }
                                int sn = Integer.parseInt(snElement.getText());
                                catalogDataCatch.put(take.getDevice().getDeviceId(), sn, sumNum, take.getDevice(), channelList);
                                logger.info("[收到通道]设备: {} -> {}个，{}/{}", take.getDevice().getDeviceId(), channelList.size(), catalogDataCatch.get(take.getDevice().getDeviceId()) == null ? 0 : catalogDataCatch.get(take.getDevice().getDeviceId()).size(), sumNum);
                                if (catalogDataCatch.get(take.getDevice().getDeviceId()).size() == sumNum) {
                                    // 数据已经完整接收， 此时可能存在某个设备离线变上线的情况，但是考虑到性能，此处不做处理，
                                    // 目前支持设备通道上线通知时和设备上线时向上级通知
                                    boolean resetChannelsResult = storager.resetChannels(take.getDevice().getDeviceId(), catalogDataCatch.get(take.getDevice().getDeviceId()));
                                    if (!resetChannelsResult) {
                                        String errorMsg = "接收成功，写入失败，共" + sumNum + "条，已接收" + catalogDataCatch.get(take.getDevice().getDeviceId()).size() + "条";
                                        catalogDataCatch.setChannelSyncEnd(take.getDevice().getDeviceId(), errorMsg);
                                    } else {
                                        catalogDataCatch.setChannelSyncEnd(take.getDevice().getDeviceId(), null);
                                    }
                                }
                            }

                        }
                    } catch (Exception e) {
                        logger.warn("[收到通道] 发现未处理的异常, \r\n{}", evt.getRequest());
                        logger.error("[收到通道] 异常内容： ", e);
                    }
                }
                processing.set(false);
            });
        }

    
<methods>public non-sealed void <init>() ,public HeaderFactory getHeaderFactory() ,public MessageFactory getMessageFactory() ,public Element getRootElement(RequestEvent) throws DocumentException,public Element getRootElement(RequestEvent, java.lang.String) throws DocumentException,public SIPResponse responseAck(SIPRequest, int) throws SipException, InvalidArgumentException, java.text.ParseException,public SIPResponse responseAck(SIPRequest, int, java.lang.String) throws SipException, InvalidArgumentException, java.text.ParseException,public SIPResponse responseAck(SIPRequest, int, java.lang.String, com.genersoft.iot.vmp.gb28181.transmit.event.request.SIPRequestProcessorParent.ResponseAckExtraParam) throws SipException, InvalidArgumentException, java.text.ParseException,public SIPResponse responseSdpAck(SIPRequest, java.lang.String, com.genersoft.iot.vmp.gb28181.bean.ParentPlatform) throws SipException, InvalidArgumentException, java.text.ParseException,public SIPResponse responseXmlAck(SIPRequest, java.lang.String, com.genersoft.iot.vmp.gb28181.bean.ParentPlatform, java.lang.Integer) throws SipException, InvalidArgumentException, java.text.ParseException<variables>private static final Logger logger,private com.genersoft.iot.vmp.gb28181.transmit.SIPSender sipSender