| | |
| | | package com.fzzy.gateway.hx2023.kafka; |
| | | |
| | | import com.alibaba.fastjson2.JSONObject; |
| | | import com.fzzy.gateway.hx2023.data.GrainData; |
| | | import com.fzzy.gateway.hx2023.data.KafaGrainData; |
| | | |
| | | import lombok.extern.slf4j.Slf4j; |
| | | |
| | | import org.apache.kafka.clients.consumer.ConsumerRecord; |
| | | import org.springframework.beans.factory.annotation.Autowired; |
| | | import org.springframework.kafka.annotation.KafkaListener; |
| | | import org.springframework.kafka.core.KafkaTemplate; |
| | | import org.springframework.stereotype.Component; |
| | |
| | | * @param data |
| | | * @return |
| | | */ |
| | | public String sendGrainData2Cloud(GrainData data) { |
| | | public String sendGrainData2Cloud(KafaGrainData data) { |
| | | |
| | | |
| | | String strData = JSONObject.toJSONString(data); |
| | | log.debug("----推送粮情信息到云端---{}", strData); |
| | | //log.debug("----推送粮情信息到云端---{}", strData); |
| | | |
| | | //推送信息 |
| | | kafkaTemplate.send("TOPIC_ZLJ_GRAIN_TEMPERATURE", strData).addCallback(success -> { |
| | |
| | | // 消息在分区内的offset |
| | | long offset = success.getRecordMetadata().offset(); |
| | | |
| | | log.info("粮情推送成功:{}-{}-{}-{}", topic, partition, offset, data.getDeviceID()); |
| | | log.info("---粮情-->>云端成功:{}-{}-{}-{}", topic, partition, offset, strData); |
| | | }, failure -> { |
| | | log.info("粮情推送失败:{}-{}", data.getDeviceID(), failure.getMessage()); |
| | | log.info("---粮情-->>云端失败:{}-{}", data.getDeviceID(), failure.getMessage()); |
| | | }); |
| | | |
| | | return "SUCCESS"; |