| | |
|---|
| | | import com.moral.api.websocket.SingleDeviceServer; |
|---|
| | | import com.moral.util.UnitConvertUtils; |
|---|
| | | import lombok.AllArgsConstructor; |
|---|
| | | import lombok.extern.slf4j.Slf4j; |
|---|
| | | import org.apache.kafka.clients.consumer.Consumer; |
|---|
| | | import org.apache.kafka.clients.consumer.ConsumerRecord; |
|---|
| | | import org.apache.kafka.common.TopicPartition; |
|---|
| | |
|---|
| | | * @Version TODO |
|---|
| | | **/ |
|---|
| | | @Component |
|---|
| | | @Slf4j |
|---|
| | | public class SecondDataConsumer implements ConsumerSeekAware { |
|---|
| | | |
|---|
| | | @KafkaListener(containerFactory = "secondDataListenerFactory",topics = "second_data") |
|---|
| | | public void listen(ConsumerRecord<String, String> record, Consumer consumer) throws Exception { |
|---|
| | | String messageStr = record.value(); |
|---|
| | | Map<String, Object> message = (Map<String, Object>) JSON.parse(messageStr); |
|---|
| | | long serverStartTime = System.currentTimeMillis(); |
|---|
| | | //long serverStartTime = System.currentTimeMillis(); |
|---|
| | | CopyOnWriteArraySet<SingleDeviceServer> sockets = SingleDeviceServer.sockets; |
|---|
| | | for (SingleDeviceServer socket : sockets) { |
|---|
| | | //判断消息是否数据该socket |
|---|
| | |
|---|
| | | if (!mac.equalsIgnoreCase(messageMac)) |
|---|
| | | continue; |
|---|
| | | //取出基本信息 |
|---|
| | | Map<String, Device> devicesInfo = socket.getDevicesInfo(); |
|---|
| | | Device device = devicesInfo.get(mac); |
|---|
| | | Device device = socket.getDeviceAlarmInfo(); |
|---|
| | | //创建最终消息对象 |
|---|
| | | Map<String, Object> resultMessgae = new HashMap<>(); |
|---|
| | | //拼接单位 |
|---|
