//package com.muyu.parsing.test;
//
//import com.alibaba.fastjson.JSONObject;
//import com.muyu.common.kafka.constants.KafkaConstants;
//import com.muyu.parsing.domain.KafKaData;
//import com.muyu.parsing.domain.SysCarMessage;
//import com.muyu.parsing.manager.TaskManager;
//import com.muyu.parsing.service.impl.SysCarMessageServiceImpl;
//import org.apache.kafka.clients.producer.KafkaProducer;
//import org.apache.kafka.clients.producer.ProducerRecord;
//import org.eclipse.paho.client.mqttv3.*;
//import org.slf4j.Logger;
//import org.slf4j.LoggerFactory;
//import org.springframework.stereotype.Component;
//
//import javax.annotation.PostConstruct;
//import javax.annotation.Resource;
//import java.util.ArrayList;
//import java.util.List;
//
//@Component
//public class MqttTest {
//    private static final Logger log = LoggerFactory.getLogger(MqttTest.class);
//
//    private static final Integer ID = 1;
//    private static final String TOPIC = "vehicle";
//    private static final String BROKER = "tcp://106.15.136.7:1883";
//
//    @Resource
//    private KafkaProducer<String, String> kafkaProducer;
//    @Resource
//    private SysCarMessageServiceImpl sysCarMessageService;
//    @Resource
//    private TaskManager taskManager;  // 注入 TaskManager
//
//    @PostConstruct
//    public void init() {
//        String clientId = "JavaSample";
//        String userName = "";
//        String password = "";
//
//        try {
//            MqttClient sampleClient = new MqttClient(BROKER, clientId);
//            MqttConnectOptions connOpts = new MqttConnectOptions();
//            connOpts.setCleanSession(true);
//            System.out.println("Connecting to broker: " + BROKER);
//            sampleClient.connect(connOpts);
//            sampleClient.subscribe(TOPIC, 0);
//            sampleClient.setCallback(new MqttCallback() {
//                @Override
//                public void connectionLost(Throwable throwable) {
//                    // 处理连接丢失
//                }
//
//                @Override
//                public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
//                    taskManager.execute(() -> processMessage(mqttMessage));
//                }
//
//                @Override
//                public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
//                    // 处理消息发送完成
//                }
//            });
//        } catch (MqttException me) {
//            me.printStackTrace();
//        }
//    }
//
//
//}
