diff --git a/couplet-modules/couplet-analyze/couplet-analyze-msg/src/main/java/com/couplet/analyze/msg/model/ModelsKafkaMessage.java b/couplet-modules/couplet-analyze/couplet-analyze-msg/src/main/java/com/couplet/analyze/msg/model/ModelsKafkaMessage.java index 8e64a8f..e73cb34 100644 --- a/couplet-modules/couplet-analyze/couplet-analyze-msg/src/main/java/com/couplet/analyze/msg/model/ModelsKafkaMessage.java +++ b/couplet-modules/couplet-analyze/couplet-analyze-msg/src/main/java/com/couplet/analyze/msg/model/ModelsKafkaMessage.java @@ -1,7 +1,5 @@ package com.couplet.analyze.msg.model; -import com.couplet.analyze.common.contents.AnalyzeEventContents; -import com.couplet.analyze.common.event.AnalyzeEventCache; import com.couplet.analyze.common.event.AnalyzeEventCache; import com.couplet.analyze.msg.domain.CoupletMsgData; import com.couplet.analyze.msg.service.IncidentService; @@ -11,14 +9,18 @@ import lombok.extern.slf4j.Slf4j; import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.consumer.ConsumerRecords; import org.apache.kafka.clients.consumer.KafkaConsumer; -import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Component; import javax.annotation.PostConstruct; import java.time.Duration; -import java.util.*; +import java.util.Collections; +import java.util.List; +import java.util.Properties; +import java.util.Set; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; import static com.couplet.analyze.msg.utils.MsgUtils.hexToString; import static com.couplet.analyze.msg.utils.MsgUtils.sendMsg; @@ -37,8 +39,38 @@ public class ModelsKafkaMessage { private static final String TOPIC_NAME = "online"; private static final String BOOTSTRAP_SERVERS = "39.103.133.136:9092"; + + //线程池，用于异步处理消息到来时的业务逻辑 + private ExecutorService executorService = Executors.newSingleThreadExecutor(); + + //kafka消费者实例化为类成员变量 + KafkaConsumer consumer; + @Autowired private AnalyzeEventCache analyzeEventCache; + + + + + + //kafka消费者初始化 + @PostConstruct + public void initKafkaConsumer() { + Properties props = new Properties(); + props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS); +// props.put(ConsumerConfig.GROUP_ID_CONFIG, "my-consumer-group"); + props.put(ConsumerConfig.GROUP_ID_CONFIG, "ddd"); + props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer"); + props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer"); + + //消费者 + consumer = new KafkaConsumer<>(props); + //订阅主题 + consumer.subscribe(Collections.singletonList(TOPIC_NAME)); + } + + + /** * 消费者配置 * @@ -46,17 +78,14 @@ public class ModelsKafkaMessage { */ @Scheduled(fixedDelay = 50) public void consumerMessages() { - Properties props = new Properties(); - props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS); - props.put(ConsumerConfig.GROUP_ID_CONFIG, "my-consumer-group"); - props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer"); - props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer"); + executorService.execute(this::consumer); - //创建消费者 - KafkaConsumer consumer = new KafkaConsumer<>(props); + + } + + public void consumer() { try { - //订阅主题 - consumer.subscribe(Collections.singletonList(TOPIC_NAME)); + //持续消费消息 while (true) { ConsumerRecords records = consumer.poll(Duration.ofMillis(100)); @@ -71,7 +100,7 @@ public class ModelsKafkaMessage { IncidentService incidentService = SpringUtils.getBean(string); incidentService.incident(msgData); } - // 发送消息 + // 发送消息 // rabbitTemplate.convertAndSend("couplet-code-queue",msgData,message -> { // message.getMessageProperties().setMessageId(UUID.randomUUID().toString()); // return message; @@ -80,9 +109,8 @@ public class ModelsKafkaMessage { }); } } catch (Exception e) { - e.printStackTrace(); - } finally { - consumer.close(); + log.error(e.getMessage()); } } + } diff --git a/couplet-modules/couplet-analyze/couplet-analyze-msg/src/main/resources/bootstrap.yml b/couplet-modules/couplet-analyze/couplet-analyze-msg/src/main/resources/bootstrap.yml index 9b0c2eb..bed4d26 100644 --- a/couplet-modules/couplet-analyze/couplet-analyze-msg/src/main/resources/bootstrap.yml +++ b/couplet-modules/couplet-analyze/couplet-analyze-msg/src/main/resources/bootstrap.yml @@ -48,3 +48,9 @@ logging: mybatis-plus: configuration: map-underscore-to-camel-case: true + +# RabbitMQ配置 +mq: + queueName: queue + exchangeName: exchange + routingKey: routingKey diff --git a/couplet-modules/couplet-business/src/main/java/com/couplet/business/server/controller/SysTroubleController.java b/couplet-modules/couplet-business/src/main/java/com/couplet/business/server/controller/SysTroubleController.java index b9f733b..8f79174 100644 --- a/couplet-modules/couplet-business/src/main/java/com/couplet/business/server/controller/SysTroubleController.java +++ b/couplet-modules/couplet-business/src/main/java/com/couplet/business/server/controller/SysTroubleController.java @@ -93,4 +93,13 @@ public class SysTroubleController extends BaseController { troubleService.newFaultData(code); return success(); } + + /** + * 查询故障的状态并处理 + */ + @PostMapping("/getUpdState") + public Result getUpdState(@RequestBody CoupletTroubleCode code) { + int updateState = troubleService.updateState(code); + return Result.success(updateState); + } } diff --git a/couplet-modules/couplet-business/src/main/java/com/couplet/business/server/mapper/SysTroubleMapper.java b/couplet-modules/couplet-business/src/main/java/com/couplet/business/server/mapper/SysTroubleMapper.java index bf95533..760a95a 100644 --- a/couplet-modules/couplet-business/src/main/java/com/couplet/business/server/mapper/SysTroubleMapper.java +++ b/couplet-modules/couplet-business/src/main/java/com/couplet/business/server/mapper/SysTroubleMapper.java @@ -29,4 +29,6 @@ public interface SysTroubleMapper extends BaseMapper { void newFaultData(CoupletTroubleCode code); void cleanTroubleCode(); + + int updateState(CoupletTroubleCode code); } diff --git a/couplet-modules/couplet-business/src/main/java/com/couplet/business/server/service/SysTroubleService.java b/couplet-modules/couplet-business/src/main/java/com/couplet/business/server/service/SysTroubleService.java index cb9b551..ca8ba18 100644 --- a/couplet-modules/couplet-business/src/main/java/com/couplet/business/server/service/SysTroubleService.java +++ b/couplet-modules/couplet-business/src/main/java/com/couplet/business/server/service/SysTroubleService.java @@ -26,4 +26,6 @@ public interface SysTroubleService extends IService { void newFaultData(CoupletTroubleCode code); void cleanTroubleCode(); + + int updateState(CoupletTroubleCode code); } diff --git a/couplet-modules/couplet-business/src/main/java/com/couplet/business/server/service/impl/SysTroubleServiceImpl.java b/couplet-modules/couplet-business/src/main/java/com/couplet/business/server/service/impl/SysTroubleServiceImpl.java index 56797cc..511ff60 100644 --- a/couplet-modules/couplet-business/src/main/java/com/couplet/business/server/service/impl/SysTroubleServiceImpl.java +++ b/couplet-modules/couplet-business/src/main/java/com/couplet/business/server/service/impl/SysTroubleServiceImpl.java @@ -58,4 +58,14 @@ public class SysTroubleServiceImpl extends ServiceImpl truncate table couplet_trouble_code + + update couplet_trouble_code set processing_state = 1 where thourble_id = #{troubleId} +