device-link-ykc/src/main/java/com/tmzn/devicelinkykc/taskQueue/queue/MsgTranscationQueue.java

package com.tmzn.devicelinkykc.taskQueue.queue;

import com.tmzn.devicelinkykc.socket.DeviceConnectionMsg;
import com.tmzn.devicelinkykc.taskQueue.runner.MsgTranscationRunner;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;

import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

/**
 * @author xp
 * @date 2024/3/15
 * @explain "  "
 */
@Component
@Slf4j(topic = "MsgTranscationQueue")
public class MsgTranscationQueue extends Thread implements ApplicationRunner {

    private BlockingQueue<Map<String, DeviceConnectionMsg>> freeQueue = new ArrayBlockingQueue(1);


    @Autowired
    private MsgTranscationRunner msgTranscationRunner;

    public void add(Map<String, DeviceConnectionMsg> object){
        //            freeQueue.put(object);
        boolean success = freeQueue.offer(object);
        if (!success) {
            log.info("队列已满，丢弃transaction设备状态更新: size={}", object.size());
        }
    }

    @Override
    public void run() {
        while (true){
            try {
                Map<String, DeviceConnectionMsg> map = freeQueue.take();
                msgTranscationRunner.transactionMsg(map);
            } catch (Exception e) {
                e.printStackTrace();
                log.error("执行异常",e);
            }
        }
    }


    @Override
    public void run(ApplicationArguments args) throws Exception {
        this.start();
        log.info("处理器启动");
    }
}