| | |
|---|
| | | import com.moral.constant.KafkaConstants; |
|---|
| | | import com.moral.constant.RedisConstants; |
|---|
| | | |
|---|
| | | /* |
|---|
| | | * 普通设备消费者 |
|---|
| | | * */ |
|---|
| | | |
|---|
| | | |
|---|
| | | @Component |
|---|
| | | @Slf4j |
|---|
| | | public class DeviceConsumer { |
|---|
| | |
|---|
| | | |
|---|
| | | //秒数据，修改设备状态，缓存最新秒数据 |
|---|
| | | @KafkaListener(topics = KafkaConstants.TOPIC_SECOND, containerFactory = "stateListenerContainerFactory") |
|---|
| | | public void listenSecond(ConsumerRecord<String, String> record) { |
|---|
| | | public void listenSecond(ConsumerRecord<String, String> record, Acknowledgment ack) { |
|---|
| | | String msg = record.value(); |
|---|
| | | try { |
|---|
| | | Map<String, Object> data = JSON.parseObject(msg, Map.class); |
|---|
| | |
|---|
| | | //判断并修改设备状态 |
|---|
| | | data.put("mac", mac); |
|---|
| | | deviceService.judgeDeviceState(data); |
|---|
| | | ack.acknowledge(); |
|---|
| | | } catch (Exception e) { |
|---|
| | | log.error("param{}" + msg); |
|---|
| | | } |
|---|
