five-groups
/
five-groups-couplet
4
0
Fork 0
Code Issues Pull Requests 2 Packages Projects Releases Wiki Activity

Compare commits

base: five-groups:server_five_dongxiaodong
Branches Tags
five-groups:server_five_fufanrui
five-groups:server_five_dongxiaodong
five-groups:server_five
five-groups:server_five_liuyunhu
..
compare: five-groups:server_five
Branches Tags
five-groups:server_five_fufanrui
five-groups:server_five_dongxiaodong
five-groups:server_five
five-groups:server_five_liuyunhu

No commits in common. "server_five_dongxiaodong" and "server_five" have entirely different histories.
server_fiv ... server_fiv

16 changed files with 133 additions and 182 deletions
Show Stats Expand all files Collapse all files

8
couplet-common/couplet-common-business/src/main/java/com/couplet/remote/RemoteTroubleService.java
View File

@ -8,7 +8,6 @@ import org.springframework.cloud.openfeign.FeignClient;
import org.springframework.web.bind.annotation.PostMapping; import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody; import org.springframework.web.bind.annotation.RequestBody;
@FeignClient(contextId = "remoteTroubleService" , @FeignClient(contextId = "remoteTroubleService" ,
value = ServiceNameConstants.BUSINESS_SERVICE, value = ServiceNameConstants.BUSINESS_SERVICE,
fallbackFactory = RemoteTroubleFallbackFactory.class fallbackFactory = RemoteTroubleFallbackFactory.class
@ -23,11 +22,4 @@ public interface RemoteTroubleService {
@PostMapping("/trouble/newFaultData") @PostMapping("/trouble/newFaultData")
public Result<?> newFaultData(@RequestBody CoupletTroubleCode code); public Result<?> newFaultData(@RequestBody CoupletTroubleCode code);
/**
* 修改故障码
* @param trouble
* @return
*/
@PostMapping("/trouble/updEndTime")
public int updEndTime(@RequestBody CoupletTroubleCode trouble);
} }

7
couplet-common/couplet-common-business/src/main/java/com/couplet/remote/factory/RemoteTroubleFallbackFactory.java
View File

@ -8,7 +8,6 @@ import org.slf4j.LoggerFactory;
import org.springframework.cloud.openfeign.FallbackFactory; import org.springframework.cloud.openfeign.FallbackFactory;
import org.springframework.stereotype.Component; import org.springframework.stereotype.Component;
/** /**
* @author fufanrui * @author fufanrui
* @version 1.0 * @version 1.0
@ -27,13 +26,9 @@ public class RemoteTroubleFallbackFactory implements FallbackFactory<RemoteTroub
@Override @Override
public Result<?> newFaultData(CoupletTroubleCode code) { public Result<?> newFaultData(CoupletTroubleCode code) {
return Result.error("调用失败...." + cause.getMessage()); return Result.error("调用失败...."+cause.getMessage());
} }
@Override
public int updEndTime(CoupletTroubleCode trouble) {
return 0;
}
}; };
} }
} }

2
couplet-modules/couplet-analyze/couplet-analyze-msg/pom.xml
View File

@ -80,7 +80,6 @@
<groupId>com.couplet</groupId> <groupId>com.couplet</groupId>
<artifactId>couplet-common-swagger</artifactId> <artifactId>couplet-common-swagger</artifactId>
</dependency> </dependency>
<!-- 事件核心配置 --> <!-- 事件核心配置 -->
<dependency> <dependency>
<groupId>com.couplet</groupId> <groupId>com.couplet</groupId>
@ -99,6 +98,7 @@
<artifactId>couplet-common-business</artifactId> <artifactId>couplet-common-business</artifactId>
</dependency> </dependency>
</dependencies> </dependencies>
</project> </project>

2
couplet-modules/couplet-analyze/couplet-analyze-msg/src/main/java/com/couplet/analyze/msg/CoupletMsgApplication.java
View File

@ -20,7 +20,7 @@ public class CoupletMsgApplication {
public static void main(String[] args) { public static void main(String[] args) {
SpringApplication.run(CoupletMsgApplication.class); SpringApplication.run(CoupletMsgApplication.class);
System.out.println("解析系统启动成功"); System.out.println("解析系统启动成功");
// new ModelsKafkaMessage().initKafkaConsumer(); new ModelsKafkaMessage().initKafkaConsumer();
} }

13
couplet-modules/couplet-analyze/couplet-analyze-msg/src/main/java/com/couplet/analyze/msg/model/ModelsKafkaMessage.java
View File

@ -38,7 +38,7 @@ import static java.lang.Thread.sleep;
@Component @Component
@Slf4j @Slf4j
public class ModelsKafkaMessage { public class ModelsKafkaMessage {
private static final String TOPIC_NAME = "topic_dxd"; private static final String TOPIC_NAME = "topic_lyh";
private static final String BOOTSTRAP_SERVERS = "39.103.133.136:9092"; private static final String BOOTSTRAP_SERVERS = "39.103.133.136:9092";
@ -52,20 +52,23 @@ public class ModelsKafkaMessage {
private AnalyzeEventCache analyzeEventCache; private AnalyzeEventCache analyzeEventCache;
//kafka消费者初始化 //kafka消费者初始化
@PostConstruct @PostConstruct
public void initKafkaConsumer() { public void initKafkaConsumer() {
Properties props = new Properties(); Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS); props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
// props.put(ConsumerConfig.GROUP_ID_CONFIG, "my-consumer-group"); // props.put(ConsumerConfig.GROUP_ID_CONFIG, "my-consumer-group");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "fbab"); props.put(ConsumerConfig.GROUP_ID_CONFIG, "group");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer"); 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"); props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
//消费者 //消费者
consumer = new KafkaConsumer<>(props); consumer = new KafkaConsumer<>(props);
//订阅主题
this.consumerMessages(); consumer.subscribe(Collections.singletonList(TOPIC_NAME));
} }
@ -83,8 +86,6 @@ public class ModelsKafkaMessage {
public void consumer() { public void consumer() {
try { try {
//订阅主题
consumer.subscribe(Collections.singletonList(TOPIC_NAME));
//持续消费消息 //持续消费消息
while (true) { while (true) {

17
couplet-modules/couplet-analyze/couplet-analyze-msg/src/main/java/com/couplet/analyze/msg/service/impl/BreakdownServiceImpl.java
View File

@ -1,5 +1,6 @@
package com.couplet.analyze.msg.service.impl; package com.couplet.analyze.msg.service.impl;
import com.alibaba.fastjson.JSON;
import com.couplet.analyze.common.contents.AnalyzeEventContents; import com.couplet.analyze.common.contents.AnalyzeEventContents;
import com.couplet.analyze.msg.contents.StateConstant; import com.couplet.analyze.msg.contents.StateConstant;
import com.couplet.analyze.msg.domain.CoupletMsgData; import com.couplet.analyze.msg.domain.CoupletMsgData;
@ -79,18 +80,13 @@ public class BreakdownServiceImpl extends KeyExpirationEventMessageListener impl
long expireTime = 30; long expireTime = 30;
redisService.expire(key, expireTime, TimeUnit.MINUTES); redisService.expire(key, expireTime, TimeUnit.MINUTES);
scheduledRedis(); scheduledRedis();
// long timeMillis = System.currentTimeMillis();
// log.info("故障事件结束时间："+timeMillis);
// log.info("故障事件检测结束.....");
}else {
long timeMillis = System.currentTimeMillis(); long timeMillis = System.currentTimeMillis();
CoupletTroubleCode troubleCode = new CoupletTroubleCode();
troubleCode.setTroubleEndTime(new Date());
troubleCode.setTroubleVin(coupletMsgData.getVin());
remoteTroubleService.updEndTime(troubleCode);
log.info("故障事件结束时间："+timeMillis); log.info("故障事件结束时间："+timeMillis);
log.info("故障事件结束....."); log.info("故障事件检测结束.....");
} }
long timeMillis = System.currentTimeMillis();
log.info("故障事件结束时间："+timeMillis);
log.info("故障事件结束.....");
} }
/** /**
@ -109,7 +105,6 @@ public class BreakdownServiceImpl extends KeyExpirationEventMessageListener impl
for (CoupletMsgData member : members) { for (CoupletMsgData member : members) {
Set<String> breakdownIds = redisService.getCacheSet(member.getVin()+":"+key); Set<String> breakdownIds = redisService.getCacheSet(member.getVin()+":"+key);
if (breakdownIds.size()==0){ if (breakdownIds.size()==0){
//异步执行任务的方法，它可以在后台执行一个无返回值的异步操作。该方法接受一个 Runnable 类型的参数，用于指定要执行的任务
CompletableFuture.runAsync(() -> { CompletableFuture.runAsync(() -> {
CoupletTroubleCode troubleCode = new CoupletTroubleCode(); CoupletTroubleCode troubleCode = new CoupletTroubleCode();
troubleCode.setTroubleStartTime(new Date()); troubleCode.setTroubleStartTime(new Date());
@ -256,7 +251,7 @@ public class BreakdownServiceImpl extends KeyExpirationEventMessageListener impl
break; break;
} }
remoteTroubleService.newFaultData(troubleCode); remoteTroubleService.newFaultData(troubleCode);
}); });
redisService.setCacheSet(member.getVin()+":"+key, member.getVin()+":"+member); redisService.setCacheSet(member.getVin()+":"+key, member.getVin()+":"+member);
long expireTime = 30; long expireTime = 30;
redisService.expire(member.getVin()+":"+key, expireTime, TimeUnit.MINUTES); redisService.expire(member.getVin()+":"+key, expireTime, TimeUnit.MINUTES);

2
couplet-modules/couplet-analyze/couplet-analyze-msg/src/main/java/com/couplet/analyze/msg/service/impl/ElectronicFenceServiceImpl.java
View File

@ -49,7 +49,7 @@ public class ElectronicFenceServiceImpl implements IncidentService {
log.info("电子围栏事件redis存在......."); log.info("电子围栏事件redis存在.......");
for (Fence fence : cacheSet) { for (Fence fence : cacheSet) {
String fenceLongitudeLatitude = fence.getFenceLongitudeLatitude(); String fenceLongitudeLatitude = fence.getFenceLongitudeLatitude();
if (!StringUtils.isEmpty(fenceLongitudeLatitude)){ if (StringUtils.isEmpty(fenceLongitudeLatitude)){
/** /**
* 先通过；后切割 * 先通过；后切割
*/ */

13
couplet-modules/couplet-analyze/couplet-analyze-msg/src/main/java/com/couplet/analyze/msg/utils/MsgUtils.java
View File

@ -5,9 +5,6 @@ import lombok.extern.slf4j.Slf4j;
import java.math.BigDecimal; import java.math.BigDecimal;
import java.nio.charset.StandardCharsets; import java.nio.charset.StandardCharsets;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.Date; import java.util.Date;
import java.util.List; import java.util.List;
@ -22,16 +19,6 @@ import java.util.Random;
@Slf4j @Slf4j
public class MsgUtils { public class MsgUtils {
public static Date convertMillisToDate(long millis) {
Instant instant = Instant.ofEpochMilli(millis);
LocalDateTime localDateTime = LocalDateTime.ofInstant(instant, ZoneId.systemDefault());
return Date.from(localDateTime.atZone(ZoneId.systemDefault()).toInstant());
}
/**
* 生成GTA
* @return
*/
public static String generateGTA() { public static String generateGTA() {
// 生成四位以"GTA"开头的字符串 // 生成四位以"GTA"开头的字符串
String prefix = "GTA"; String prefix = "GTA";

26
couplet-modules/couplet-analyze/couplet-analyze-msg/src/test/java/com/couplet/msg/Main.java
View File

@ -1,10 +1,5 @@
package com.couplet.msg; package com.couplet.msg;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.util.Date;
import java.util.Random; import java.util.Random;
import java.util.regex.Matcher; import java.util.regex.Matcher;
import java.util.regex.Pattern; import java.util.regex.Pattern;
@ -17,14 +12,17 @@ import java.util.regex.Pattern;
*/ */
public class Main { public class Main {
public static void main(String[] args) { public static void main(String[] args) {
long timeMillis = System.currentTimeMillis(); // 创建一个字符串数组存储三个状态
Date date = new Date(timeMillis); String[] statuses = {"电池故障", "车体故障", "车尾故障","抽轮故障"};
System.out.println(date);
}
// private static Date convertMillisToDate(long millis) { // 生成随机数对象
// Instant instant = Instant.ofEpochMilli(millis); Random rand = new Random();
// LocalDateTime localDateTime = LocalDateTime.ofInstant(instant, ZoneId.systemDefault());
// return Date.from(localDateTime.atZone(ZoneId.systemDefault()).toInstant()); // 生成一个范围在 0 到 2 之间的随机整数
// } int randomIndex = rand.nextInt(4);
// 随机选择一个字符串并输出
String randomStatus = statuses[randomIndex];
System.out.println("随机输出的字符串：" + randomStatus);
}
} }

9
couplet-modules/couplet-business/src/main/java/com/couplet/business/server/controller/SysTroubleController.java
View File

@ -14,7 +14,6 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.validation.annotation.Validated; import org.springframework.validation.annotation.Validated;
import org.springframework.web.bind.annotation.*; import org.springframework.web.bind.annotation.*;
import java.util.Date;
import java.util.List; import java.util.List;
/** /**
@ -98,9 +97,9 @@ public class SysTroubleController extends BaseController {
/** /**
* 查询故障的状态并处理 * 查询故障的状态并处理
*/ */
@PostMapping("/updEndTime") @PostMapping("/getUpdState")
public int updEndTime(@RequestBody CoupletTroubleCode trouble) { public Result<?> getUpdState(@RequestBody CoupletTroubleCode code) {
int i = troubleService.updEndTime(trouble); int updateState = troubleService.updateState(code);
return i; return Result.success(updateState);
} }
} }

6
couplet-modules/couplet-business/src/main/java/com/couplet/business/server/mapper/SysTroubleMapper.java
View File

@ -6,9 +6,7 @@ import com.couplet.common.domain.CoupletTroubleGrade;
import com.couplet.common.domain.request.TroubleResp; import com.couplet.common.domain.request.TroubleResp;
import org.apache.ibatis.annotations.Mapper; import org.apache.ibatis.annotations.Mapper;
import org.apache.ibatis.annotations.Param;
import java.util.Date;
import java.util.List; import java.util.List;
/** /**
@ -32,7 +30,5 @@ public interface SysTroubleMapper extends BaseMapper<CoupletTroubleCode> {
void cleanTroubleCode(); void cleanTroubleCode();
int updEndTime(CoupletTroubleCode trouble); int updateState(CoupletTroubleCode code);
List<CoupletTroubleCode> selectMaxTime();
} }

3
couplet-modules/couplet-business/src/main/java/com/couplet/business/server/service/SysTroubleService.java
View File

@ -6,7 +6,6 @@ import com.couplet.common.domain.CoupletTroubleCode;
import com.couplet.common.domain.CoupletTroubleGrade; import com.couplet.common.domain.CoupletTroubleGrade;
import com.couplet.common.domain.request.TroubleResp; import com.couplet.common.domain.request.TroubleResp;
import java.util.Date;
import java.util.List; import java.util.List;
/** /**
@ -28,5 +27,5 @@ public interface SysTroubleService extends IService<CoupletTroubleCode> {
void cleanTroubleCode(); void cleanTroubleCode();
int updEndTime(CoupletTroubleCode trouble); int updateState(CoupletTroubleCode code);
} }

11
couplet-modules/couplet-business/src/main/java/com/couplet/business/server/service/impl/SysTroubleServiceImpl.java
View File

@ -12,7 +12,6 @@ import com.github.pagehelper.PageInfo;
import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
import java.util.Date;
import java.util.List; import java.util.List;
/** /**
@ -61,17 +60,11 @@ public class SysTroubleServiceImpl extends ServiceImpl<SysTroubleMapper, Couplet
/** /**
* 获取故障状态 * 获取故障状态
* @param trouble * @param code
* @return * @return
*/ */
@Override @Override
public int updEndTime(CoupletTroubleCode trouble) { public int updateState(CoupletTroubleCode code) {
// 查询表中最大时间
List<CoupletTroubleCode> maxTime = sysTroubleMapper.selectMaxTime();
// 然后将结束时间改成当前时间
if (maxTime != null) {
sysTroubleMapper.updEndTime(trouble);
}
return 0; return 0;
} }

8
couplet-modules/couplet-business/src/main/resources/mapper/business/SysTroubleMapper.xml
View File

@ -26,11 +26,10 @@
<update id="cleanTroubleCode"> <update id="cleanTroubleCode">
truncate table couplet_trouble_code truncate table couplet_trouble_code
</update> </update>
<update id="updEndTime"> <update id="updateState">
update couplet_trouble_code set trouble_end_time = #{troubleEndTime} where trouble_vin = #{troubleVin} update couplet_trouble_code set processing_state = 1 where thourble_id = #{troubleId}
</update> </update>
<select id="selectTroubleList" parameterType="com.couplet.business.server.mapper.SysTroubleMapper" resultMap="SysTroubleResult"> <select id="selectTroubleList" parameterType="com.couplet.business.server.mapper.SysTroubleMapper" resultMap="SysTroubleResult">
<include refid="selectTroubleVo"/> <include refid="selectTroubleVo"/>
<where> <where>
@ -46,8 +45,5 @@
<select id="selectTroubleListByGrade" resultType="com.couplet.common.domain.CoupletTroubleGrade"> <select id="selectTroubleListByGrade" resultType="com.couplet.common.domain.CoupletTroubleGrade">
select * from couplet_trouble_grade select * from couplet_trouble_grade
</select> </select>
<select id="selectMaxTime" resultType="com.couplet.common.domain.CoupletTroubleCode">
select max(trouble_start_time) from couplet_trouble_code
</select>
</mapper> </mapper>

186
couplet-modules/couplet-modules-mq/src/main/java/com/couplet/mq/controller/KafkaTest.java
View File

@ -1,93 +1,93 @@
package com.couplet.mq.controller; //package com.couplet.mq.controller;
//
import lombok.extern.slf4j.Slf4j; //import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerConfig; //import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords; //import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer; //import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer; //import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig; //import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord; //import org.apache.kafka.clients.producer.ProducerRecord;
//
import javax.annotation.PostConstruct; //import javax.annotation.PostConstruct;
import java.time.Duration; //import java.time.Duration;
import java.util.Collections; //import java.util.Collections;
import java.util.Properties; //import java.util.Properties;
//
/** ///**
* @ProjectName: five-groups-couplet // * @ProjectName: five-groups-couplet
* @Author: LiuYunHu // * @Author: LiuYunHu
* @CreateTime: 2024/4/4 // * @CreateTime: 2024/4/4
* @Description: kafka测试类 // * @Description: kafka测试类
*/ // */
//
@Slf4j //@Slf4j
public class KafkaTest { //public class KafkaTest {
private static final String TOPIC_NAME = "topic_dxd"; // private static final String TOPIC_NAME = "online";
private static final String BOOTSTRAP_SERVERS = "39.103.133.136:9092"; // private static final String BOOTSTRAP_SERVERS = "39.103.133.136:9092";
//
public static void main(String[] args) { // public static void main(String[] args) {
//生产者示例 // //生产者示例
// produceMessage(); //// produceMessage();
//
//消费者示例 // //消费者示例
consumerMessages(); //// consumerMessages();
//
} // }
//
//生产者 // //生产者
@PostConstruct // @PostConstruct
private static void produceMessage() { // private static void produceMessage() {
Properties props = new Properties(); // Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS); // props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer"); // props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer"); // props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
//
KafkaProducer<String, String> producer = new KafkaProducer<>(props); // KafkaProducer<String, String> producer = new KafkaProducer<>(props);
//创建生产者 // //创建生产者
try { // try {
//
//发送消息 // //发送消息
for (int i = 0; i < 10000; i++) { // for (int i = 0; i < 10000; i++) {
String message = "佳佳来喽" + (i + 1); // String message = "佳佳来喽" + (i + 1);
producer.send(new ProducerRecord<>(TOPIC_NAME, message)); // producer.send(new ProducerRecord<>(TOPIC_NAME, message));
//
System.out.println("发送消息：" + message); // System.out.println("发送消息：" + message);
} // }
} catch (Exception e) { // } catch (Exception e) {
e.printStackTrace(); // e.printStackTrace();
} finally { // } finally {
producer.close(); // producer.close();
} // }
} // }
//
//消费者 // //消费者
private static void consumerMessages() { //// private static void consumerMessages() {
Properties props = new Properties(); //// Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS); //// props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
props.put(ConsumerConfig.GROUP_ID_CONFIG, "my-consumer-group"); //// 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.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer"); //// props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
////
//创建消费者 //// //创建消费者
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props); //// KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
////
try { //// try {
////
//订阅主题 //// //订阅主题
consumer.subscribe(Collections.singletonList(TOPIC_NAME)); //// consumer.subscribe(Collections.singletonList(TOPIC_NAME));
////
//持续消费消息 //// //持续消费消息
while (true) { //// while (true) {
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100)); //// ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
records.forEach(record -> { //// records.forEach(record -> {
System.out.println("消费者接受到的消息值：" + record.value()); //// System.out.println("消费者接受到的消息值：" + record.value());
}); //// });
} //// }
} catch (Exception e) { //// } catch (Exception e) {
e.printStackTrace(); //// e.printStackTrace();
} finally { //// } finally {
consumer.close(); //// consumer.close();
} //// }
} //// }
//
} //}

2
couplet-modules/couplet-modules-onLine/src/main/java/com/couplet/online/utils/MqttMonitor.java
View File

@ -76,7 +76,7 @@ public class MqttMonitor {
//Kafka生产者配置 //Kafka生产者配置
private static final String TOPIC_NAME = "topic_dxd"; private static final String TOPIC_NAME = "topic_lhy";
private static final String BOOTSTRAP_SERVERS = "39.103.133.136:9092"; private static final String BOOTSTRAP_SERVERS = "39.103.133.136:9092";
//线程池，用于异步处理消息到来时的业务逻辑 //线程池，用于异步处理消息到来时的业务逻辑