group-four
/
cloud-car
5
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/dev' into dev.operation

dev.operation
86191 2024-09-30 10:32:54 +08:00
parent a6521e46c9 1f072d4ca2
commit 7149ea8f1c
19 changed files with 411 additions and 344 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
cloud-modules/cloud-modules-vehiclegateway/pom.xml
View File

@ -125,6 +125,11 @@
<artifactId>tea-util</artifactId>
<version>0.2.21</version>
</dependency>
<dependency>
<groupId>com.aliyun</groupId>
<artifactId>aliyun-java-sdk-ecs</artifactId>
<version>4.2.0</version><!-- 请根据实际情况使用最新的版本 -->
</dependency>
</dependencies>

173
cloud-modules/cloud-modules-vehiclegateway/src/main/java/com/muyu/vehicle/ManageInstance.java 100644
View File

@ -0,0 +1,173 @@
package com.muyu.vehicle;
import com.aliyun.ecs20140526.Client;
import com.aliyun.ecs20140526.models.*;
import com.aliyun.tea.TeaException;
import com.aliyun.teaopenapi.models.Config;
import com.aliyun.teautil.Common;
import com.aliyun.teautil.models.RuntimeOptions;
import com.muyu.vehicle.domain.InstanceInfo;
import com.muyu.vehicle.service.OpenInstance;
import com.muyu.vehicle.service.SelectInstance;
import lombok.extern.log4j.Log4j2;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;
import java.util.ArrayList;
import java.util.List;
@Component
@Log4j2
public class ManageInstance implements ApplicationRunner {
/**
* ACCESS_KEY_ID
*/
public static final String ALIBABA_CLOUD_ACCESS_KEY_ID="LTAI5tGabdxedjfCh2uXHNrw";
/**
*ACCESS_KEY_SECRET
*/
public static final String ACCESS_KEY_SECRET="NHb7wHVpesLW6Axc0bFBs6ThhuNR10";
/**
* 镜像ID
*/
public static final String IMAGE_ID="m-uf6agr9i6g27gj23om34";
/**
* 实例类型
*/
public static final String INSTANCE_TYPE="ecs.e-c1m1.large";
/**
* 安全组ID
*/
public static final String SECURITY_GROUP_ID="sg-uf6glo8c4k17szhxu7sk";
/**
*交换机ID
*/
public static final String V_SWITCH_ID="vsw-uf6xy4rbt9ggcz93t6oib";
/**
* 实例付费类型
*/
public static final String INSTANCE_CHARGE_TY="PostPaid";
/**
* 使用AK&SK初始化账号Client
* @return Client
* @throws Exception
*/
public static Client createClient() throws Exception {
// 工程代码泄露可能会导致 AccessKey 泄露，并威胁账号下所有资源的安全性。以下代码示例仅供参考。
Config config = new Config()
// 必填，请确保代码运行环境设置了环境变量 ALIBABA_CLOUD_ACCESS_KEY_ID。
.setAccessKeyId(ALIBABA_CLOUD_ACCESS_KEY_ID)
// 必填，请确保代码运行环境设置了环境变量 ALIBABA_CLOUD_ACCESS_KEY_SECRET。
.setAccessKeySecret(ACCESS_KEY_SECRET);
// Endpoint 请参考 https://api.aliyun.com/product/Ecs
config.endpoint = "ecs-cn-hangzhou.aliyuncs.com";
return new com.aliyun.ecs20140526.Client(config);
}
public static void generateInstance() throws Exception {
// 创建阿里云ECS客户端
Client client = ManageInstance.createClient();
// 配置系统盘参数
RunInstancesRequest.RunInstancesRequestSystemDisk systemDisk=
new RunInstancesRequest.RunInstancesRequestSystemDisk()
.setSize("40")
.setCategory("cloud_essd");
// 创建创建实例请求对象并设置参数
RunInstancesRequest runInstancesRequest = new RunInstancesRequest()
.setRegionId("cn-shanghai") // 设置地域ID
.setImageId(IMAGE_ID) // 设置镜像ID
.setInstanceType(INSTANCE_TYPE) // 设置实例类型
.setSecurityGroupId(SECURITY_GROUP_ID) // 设置安全组ID
.setVSwitchId(V_SWITCH_ID) // 设置虚拟交换机ID
.setInstanceName("cloud-MQTT") // 设置实例名称
.setInstanceChargeType(INSTANCE_CHARGE_TY) // 设置实例付费类型为后付费按量付费
.setSystemDisk(systemDisk) // 设置系统盘配置
.setHostName("root") // 设置主机名
.setPassword("2112A-four") // 设置实例密码
.setAmount(2) // 设置创建实例的数量
.setInternetChargeType("PayByTraffic")
.setInternetMaxBandwidthOut(1);
//创建运行时选择对象
RuntimeOptions runTime=
new RuntimeOptions();
// 尝试执行创建实例请求
try {
ArrayList<String> list = new ArrayList<>();
// 复制代码运行请自行打印 API 的返回值
RunInstancesResponse runInstancesResponse = client.runInstancesWithOptions(runInstancesRequest, runTime);
RunInstancesResponseBody body = runInstancesResponse.getBody();
for (String instance : body.getInstanceIdSets().getInstanceIdSet()) {
list.add(instance);
}
log.info("ESC创建成功,实例ID为:" + list);
} catch (TeaException error) {
// 错误 message
log.info(error.getMessage());
// 诊断地址
log.info(error.getData().get("Recommend"));
Common.assertAsString(error.message);
} catch (Exception _error) {
TeaException error = new TeaException(_error.getMessage(), _error);
// 此处仅做打印展示，请谨慎对待异常处理，在工程项目中切勿直接忽略异常。
// 错误 message
log.info("实例创建失败:"+error.getMessage());
}
}
private static List<InstanceInfo> selectInstance() throws Exception {
Client client = ManageInstance.createClient();
ArrayList<InstanceInfo> instanceInfos = new ArrayList<>();// 实例基础信息
DescribeInstancesRequest describeInstancesRequest = new DescribeInstancesRequest()
.setRegionId("cn-shanghai")
.setInternetChargeType("PayByTraffic")
.setInstanceChargeType("PostPaid")
.setInstanceName("cloud-MQTT") // 设置实例名称
;
// 创建运行时选项对象
RuntimeOptions runtime = new RuntimeOptions();
//实例ID Instances.Instance.InstanceId
//实例IP Instances.Instance.PublicIpAddress.IpAddress
//状态 Instances.Instance.Status
DescribeInstancesResponse resp =client.describeInstancesWithOptions(describeInstancesRequest, runtime);
DescribeInstancesResponseBody body = resp.getBody();
for (DescribeInstancesResponseBody.DescribeInstancesResponseBodyInstancesInstance instance : body.getInstances().getInstance()){
InstanceInfo instanceInfo = new InstanceInfo();
instanceInfo.setInstanceId(instance.getInstanceId());
instanceInfo.setIpAddress(String.valueOf(instance.getPublicIpAddress().getIpAddress()));
instanceInfo.setStatus(instance.getStatus());
instanceInfos.add(instanceInfo);
}
log.info("实例信息为:"+Common.toJSONString(instanceInfos));
return instanceInfos;
}
@Override
public void run(ApplicationArguments args) throws Exception {
generateInstance();
selectInstance();
}
}

18
cloud-modules/cloud-modules-vehiclegateway/src/main/java/com/muyu/vehicle/domain/InstanceInfo.java 100644
View File

@ -0,0 +1,18 @@
package com.muyu.vehicle.domain;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
@Data
@AllArgsConstructor
@NoArgsConstructor
public class InstanceInfo {
//实例ID
private String InstanceId;
private String IpAddress;
private String status;
}

2
cloud-modules/cloud-modules-vehiclegateway/src/main/java/com/muyu/vehicle/CloseInstance.java → cloud-modules/cloud-modules-vehiclegateway/src/main/java/com/muyu/vehicle/service/CloseInstance.java
View File

@ -1,4 +1,4 @@
package com.muyu.vehicle;
package com.muyu.vehicle.service;
import com.aliyun.ecs20140526.Client;

29
cloud-modules/cloud-modules-vehiclegateway/src/main/java/com/muyu/vehicle/OpenInstance.java → cloud-modules/cloud-modules-vehiclegateway/src/main/java/com/muyu/vehicle/service/OpenInstance.java
View File

@ -1,20 +1,20 @@
package com.muyu.vehicle;
package com.muyu.vehicle.service;
import com.aliyun.ecs20140526.Client;
import com.aliyun.ecs20140526.models.RunInstancesRequest;
import com.aliyun.ecs20140526.models.RunInstancesResponse;
import com.aliyun.ecs20140526.models.RunInstancesResponseBody;
import com.aliyun.tea.TeaException;
import com.aliyun.teaopenapi.models.Config;
import com.aliyun.teautil.Common;
import com.aliyun.teautil.models.RuntimeOptions;
import com.muyu.vehicle.ManageInstance;
import lombok.extern.log4j.Log4j2;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;
@Component
import java.util.ArrayList;
@Log4j2
public class OpenInstance implements ApplicationRunner {
public class OpenInstance {
/**
* ACCESS_KEY_ID
*/
@ -95,7 +95,7 @@ public class OpenInstance implements ApplicationRunner {
.setSystemDisk(systemDisk) // 设置系统盘配置
.setHostName("root") // 设置主机名
.setPassword("2112A-four") // 设置实例密码
.setAmount(1) // 设置创建实例的数量
.setAmount(2) // 设置创建实例的数量
.setInternetChargeType("PayByTraffic")
.setInternetMaxBandwidthOut(1);
@ -105,9 +105,14 @@ public class OpenInstance implements ApplicationRunner {
new RuntimeOptions();
// 尝试执行创建实例请求
try {
ArrayList<String> list = new ArrayList<>();
// 复制代码运行请自行打印 API 的返回值
client.runInstancesWithOptions(runInstancesRequest, runTime);
log.info("实例创建成功");
RunInstancesResponse runInstancesResponse = client.runInstancesWithOptions(runInstancesRequest, runTime);
RunInstancesResponseBody body = runInstancesResponse.getBody();
for (String instance : body.getInstanceIdSets().getInstanceIdSet()) {
list.add(instance);
log.info("ESC创建成功,实例ID为:" + list);
}
} catch (TeaException error) {
// 错误 message
log.info(error.getMessage());
@ -121,8 +126,4 @@ public class OpenInstance implements ApplicationRunner {
log.info("实例创建失败:"+error.getMessage());
}
}
@Override
public void run(ApplicationArguments args) throws Exception {
generateInstance();
}
}

66
cloud-modules/cloud-modules-vehiclegateway/src/main/java/com/muyu/vehicle/service/SelectInstance.java 100644
View File

@ -0,0 +1,66 @@
package com.muyu.vehicle.service;
import com.aliyun.ecs20140526.Client;
import com.aliyun.ecs20140526.models.DescribeInstancesRequest;
import com.aliyun.ecs20140526.models.DescribeInstancesResponse;
import com.aliyun.ecs20140526.models.DescribeInstancesResponseBody;
import com.aliyun.teaopenapi.models.Config;
import com.aliyun.teautil.Common;
import com.aliyun.teautil.models.RuntimeOptions;
import com.muyu.vehicle.domain.InstanceInfo;
import lombok.extern.log4j.Log4j2;
import java.util.ArrayList;
@Log4j2
public class SelectInstance {
/**
* ACCESS_KEY_ID
*/
public static final String ALIBABA_CLOUD_ACCESS_KEY_ID="LTAI5tGabdxedjfCh2uXHNrw";
/**
*ACCESS_KEY_SECRET
*/
public static final String ACCESS_KEY_SECRET="NHb7wHVpesLW6Axc0bFBs6ThhuNR10";
public static Client createClient(String accessKeyId, String accessKeySecret) throws Exception {
Config config = new com.aliyun.teaopenapi.models.Config()
// 必填，您的 AccessKey ID
.setAccessKeyId(ALIBABA_CLOUD_ACCESS_KEY_ID)
// 必填，您的 AccessKey Secret
.setAccessKeySecret(ACCESS_KEY_SECRET);
// 访问的域名
config.endpoint = "ecs-cn-hangzhou.aliyuncs.com";
return new Client(config);
}
public static void main(String[] args_) throws Exception {
java.util.List<String> args = java.util.Arrays.asList(args_);
// 请确保代码运行环境设置了环境变量 ALIBABA_CLOUD_ACCESS_KEY_ID 和 ALIBABA_CLOUD_ACCESS_KEY_SECRET。
// 工程代码泄露可能会导致 AccessKey 泄露，并威胁账号下所有资源的安全性。以下代码示例仅供参考，建议使用更安全的 STS 方式
Client client = SelectInstance.createClient(ALIBABA_CLOUD_ACCESS_KEY_ID, ACCESS_KEY_SECRET);
DescribeInstancesRequest describeInstancesRequest = new DescribeInstancesRequest()
.setRegionId("cn-shanghai")
.setInternetChargeType("PayByTraffic")
.setInstanceChargeType("PostPaid")
.setInstanceName("cloud-MQTT") // 设置实例名称
;
//实例ID Instances.Instance.InstanceId
//实例IP Instances.Instance.PublicIpAddress.IpAddress
//状态 Instances.Instance.Status
RuntimeOptions runtime = new RuntimeOptions();
DescribeInstancesResponse resp = client.describeInstancesWithOptions(describeInstancesRequest, runtime);
DescribeInstancesResponseBody body = resp.getBody();
ArrayList<InstanceInfo> instanceInfos = new ArrayList<>();// 实例基础信息
for (DescribeInstancesResponseBody.DescribeInstancesResponseBodyInstancesInstance instance : body.getInstances().getInstance()){
InstanceInfo instanceInfo = new InstanceInfo();
instanceInfo.setInstanceId(instance.getInstanceId());
instanceInfo.setIpAddress(String.valueOf(instance.getPublicIpAddress().getIpAddress()));
instanceInfo.setStatus(instance.getStatus());
instanceInfos.add(instanceInfo);
}
log.info(Common.toJSONString(instanceInfos));
}
}

2
cloud-modules/cloud-modules-vehiclegateway/src/main/resources/bootstrap.yml
View File

@ -54,3 +54,5 @@ logging:
level:
com.muyu.system.mapper: DEBUG

6
cloud-modules/saas/saas-common/src/main/java/com/muyu/common/domain/MessageTemplateType.java
View File

@ -1,5 +1,6 @@
package com.muyu.common.domain;
import com.baomidou.mybatisplus.annotation.IdType;
import com.baomidou.mybatisplus.annotation.TableField;
import com.baomidou.mybatisplus.annotation.TableId;
import com.baomidou.mybatisplus.annotation.TableName;
import io.swagger.v3.oas.annotations.tags.Tag;
@ -7,9 +8,7 @@ import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.experimental.SuperBuilder;
import java.io.Serializable;
/**
* 模版对应的配置
* @Author：liuxinyue
@ -25,8 +24,6 @@ import java.io.Serializable;
@Tag(name = "报文模版表")
@TableName(value = "message_template_type",autoResultMap = true)
public class MessageTemplateType implements Serializable {
/**
* 主键
*/
@ -59,6 +56,7 @@ public class MessageTemplateType implements Serializable {
/**
* 数据类型名称
*/
@TableField(exist = false)
private String dataTypeName;
/**
*最小值

239
cloud-modules/saas/saas-common/src/main/java/com/muyu/common/util/KafkaCommonProperties.java
View File

@ -1,239 +0,0 @@
package com.muyu.common.util;
import java.util.Collection;
import java.util.Collections;
/**
* kafka通用配置
* @author liuxinyue
* @Package：com.muyu.common.util
* @name：KafkaCommonProperties
* @Date：2024/9/29 12:26
*/
public class KafkaCommonProperties{
/**
* Kafka主机
*/
private String kafkaHost = "47.101.53.251:9092";
/**
* 生产者：要求leader请求结束前收到的确认次数，来控制发送数据的持久化
* 消息确认：
* 0：生产者不等待服务器确认，此时retry参数不生效
* 1：leader写入记录到log，不会等待follower的确认即向生产者发送通知
* all：leader等待所有副本通知，然后向生产者发送通知，保证所有数据落盘到所有副本，功能同设置为-1
*/
private String ack = "all";
/**
* 生产者重试次数
*/
private Integer retryTimes = 1;
/**
* 生产者：向同一分区发送打包发送的数据量，单位：bytes，默认16384bytes=16K
*/
private Integer batchSize = 16384;
/**
* 生产者：批量发送消息的间隔时间（延迟时间），单位：毫秒
*/
private Integer lingerMs = 1;
/**
* 生产者：可以使用的最大缓存空间，单位：bytes，默认33554432bytes=32M.
*/
private Integer bufferMemory = 33554432;
/**
* 生产者：键编码器
*/
private String keyEncoder = "org.apache.kafka.common.serialization.StringSerializer";
/**
* 生产者：值编码器
*/
private String valueEncoder = "org.apache.kafka.common.serialization.StringSerializer";
/**
* 消费者：消费topic的组ID
*/
private String groupId = "my-group-id";
/**
* 消费者：后台定期提交offset
*/
private String autoCommit = "true";
/**
* 消费者提交offset的时间间隔：单位：毫秒，当enable.auto.commit为true时生效
*/
private String autoCommitIntervalMs = "1000";
/**
* 消费者：键解码器
*/
private String keyDecoder = "org.apache.kafka.common.serialization.StringDeserializer";
/**
* 消费者：值解码器
*/
private String valueDecoder = "org.apache.kafka.common.serialization.StringDeserializer";
/**
* 消费者：重启后配置offset
* earliest：消费者恢复到当前topic最早的offset
* latest：消费者从最新的offset开始消费
* none：如果消费者组没找到之前的offset抛出异常
* 其他任何值都会抛出异常
*/
private String autoOffsetReset = "latest";
/**
* TOPIC
*/
private Collection<String> topic = Collections.singleton("my-topic");
public KafkaCommonProperties() {
}
public KafkaCommonProperties(String kafkaHost, String ack, Integer retryTimes, Integer batchSize, Integer lingerMs, Integer bufferMemory, String keyEncoder, String valueEncoder, String groupId, String autoCommit, String autoCommitIntervalMs, String keyDecoder, String valueDecoder, String autoOffsetReset, Collection<String> topic) {
this.kafkaHost = kafkaHost;
this.ack = ack;
this.retryTimes = retryTimes;
this.batchSize = batchSize;
this.lingerMs = lingerMs;
this.bufferMemory = bufferMemory;
this.keyEncoder = keyEncoder;
this.valueEncoder = valueEncoder;
this.groupId = groupId;
this.autoCommit = autoCommit;
this.autoCommitIntervalMs = autoCommitIntervalMs;
this.keyDecoder = keyDecoder;
this.valueDecoder = valueDecoder;
this.autoOffsetReset = autoOffsetReset;
this.topic = topic;
}
public String getKafkaHost() {
return kafkaHost;
}
public void setKafkaHost(String kafkaHost) {
this.kafkaHost = kafkaHost;
}
public String getAck() {
return ack;
}
public void setAck(String ack) {
this.ack = ack;
}
public Integer getRetryTimes() {
return retryTimes;
}
public void setRetryTimes(Integer retryTimes) {
this.retryTimes = retryTimes;
}
public Integer getBatchSize() {
return batchSize;
}
public void setBatchSize(Integer batchSize) {
this.batchSize = batchSize;
}
public Integer getLingerMs() {
return lingerMs;
}
public void setLingerMs(Integer lingerMs) {
this.lingerMs = lingerMs;
}
public Integer getBufferMemory() {
return bufferMemory;
}
public void setBufferMemory(Integer bufferMemory) {
this.bufferMemory = bufferMemory;
}
public String getKeyEncoder() {
return keyEncoder;
}
public void setKeyEncoder(String keyEncoder) {
this.keyEncoder = keyEncoder;
}
public String getValueEncoder() {
return valueEncoder;
}
public void setValueEncoder(String valueEncoder) {
this.valueEncoder = valueEncoder;
}
public String getGroupId() {
return groupId;
}
public void setGroupId(String groupId) {
this.groupId = groupId;
}
public String getAutoCommit() {
return autoCommit;
}
public void setAutoCommit(String autoCommit) {
this.autoCommit = autoCommit;
}
public String getAutoCommitIntervalMs() {
return autoCommitIntervalMs;
}
public void setAutoCommitIntervalMs(String autoCommitIntervalMs) {
this.autoCommitIntervalMs = autoCommitIntervalMs;
}
public String getKeyDecoder() {
return keyDecoder;
}
public void setKeyDecoder(String keyDecoder) {
this.keyDecoder = keyDecoder;
}
public String getValueDecoder() {
return valueDecoder;
}
public void setValueDecoder(String valueDecoder) {
this.valueDecoder = valueDecoder;
}
public String getAutoOffsetReset() {
return autoOffsetReset;
}
public void setAutoOffsetReset(String autoOffsetReset) {
this.autoOffsetReset = autoOffsetReset;
}
public Collection<String> getTopic() {
return topic;
}
public void setTopic(Collection<String> topic) {
this.topic = topic;
}
}

16
cloud-modules/saas/saas-common/src/main/java/com/muyu/common/util/KafkaConstants.java 100644
View File

@ -0,0 +1,16 @@
package com.muyu.common.util;
/**
* @author liuxinyue
* @Package：com.muyu.common.util
* @name：KafkaConstants
* @Date：2024/9/29 20:22
*/
public class KafkaConstants {
public final static String KafkaTopic="kafka_topic_test";
public final static String KafkaGrop="kafka_group_test";
}

43
cloud-modules/saas/saas-common/src/main/java/com/muyu/common/util/KafkaConsumerConfig.java 100644
View File

@ -0,0 +1,43 @@
package com.muyu.common.util;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.Deserializer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.util.HashMap;
/**
* @author liuxinyue
* @Package：com.muyu.common.util
* @name：KafkaConsumerConfig
* @Date：2024/9/29 20:19
*/
@Configuration
public class KafkaConsumerConfig
{
@Bean
public KafkaConsumer kafkaConsumer(){
HashMap<String, Object> map = new HashMap<>();
map.put("bootstrap.servers", "47.101.53.251:9092");
map.put("enable.auto.commit",true);
map.put("auto.commit.interval", 5000);
map.put("auto.offset.reset", "latest");
map.put("fetch.max.wait", 500);
map.put("fetch.min.size", 1);
map.put("heartbeat-interval", 3000);
map.put("max.poll.records", 500);
map.put("group.id", KafkaConstants.KafkaGrop);
//指定key使用的反序列化类
Deserializer keyDeserializer = new StringDeserializer();
//指定value使用的反序列化类
Deserializer valueDeserializer = new StringDeserializer();
//创建Kafka消费者
KafkaConsumer kafkaConsumer = new KafkaConsumer(map, keyDeserializer, valueDeserializer);
return kafkaConsumer;
}
}

64
cloud-modules/saas/saas-common/src/main/java/com/muyu/common/util/KafkaProducerTest.java
View File

@ -1,64 +0,0 @@
package com.muyu.common.util;
import org.apache.kafka.clients.producer.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Objects;
import java.util.Properties;
/**
* kafka生产
* @author liuxinyue
* @Package：com.muyu.common.util
* @name：KafkaProducerTest
* @Date：2024/9/29 12:27
*/
public class KafkaProducerTest {
private static final Logger logger = LoggerFactory.getLogger(KafkaProducerTest.class);
public static KafkaProducer<String, String> getDefaultKafkaProducer(KafkaCommonProperties kafkaCommonProperties) {
Properties properties = new Properties();
properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaCommonProperties.getKafkaHost());
properties.put(ProducerConfig.ACKS_CONFIG, kafkaCommonProperties.getAck());
properties.put(ProducerConfig.RETRIES_CONFIG, kafkaCommonProperties.getRetryTimes());
properties.put(ProducerConfig.BATCH_SIZE_CONFIG, kafkaCommonProperties.getBatchSize());
properties.put(ProducerConfig.LINGER_MS_CONFIG, kafkaCommonProperties.getLingerMs());
properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG, kafkaCommonProperties.getBufferMemory());
properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, kafkaCommonProperties.getKeyEncoder());
properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, kafkaCommonProperties.getValueEncoder());
return new KafkaProducer<>(properties);
}
static class MyProducerCallback implements Callback {
@Override
public void onCompletion(RecordMetadata metadata, Exception exception) {
if (Objects.nonNull(exception)) {
logger.error(">>>>>>>>>>Producer生产消息异常：", exception);
}
if (Objects.nonNull(metadata)) {
logger.info(">>>>>>>>>>Producer生产消息：metadata：{}，partition:{}, offset：{}", metadata, metadata.partition(), metadata.offset());
}
}
}
public static void main(String[] args) {
KafkaCommonProperties kafkaCommonProperties = new KafkaCommonProperties();
KafkaProducer<String, String> producer = getDefaultKafkaProducer(kafkaCommonProperties);
String message = "hello world ";
try {
for (int i = 0; i < 10; i++) {
// 异步写入数据
String topic = kafkaCommonProperties.getTopic().toArray()[0].toString();
ProducerRecord<String, String> producerRecord = new ProducerRecord<>(topic, message + i);
producer.send(producerRecord, new MyProducerCallback());
}
} catch (Exception ex) {
logger.error(">>>>>>>>生产数据异常：", ex);
throw new RuntimeException(ex);
} finally {
producer.close();
}
}
}

38
cloud-modules/saas/saas-common/src/main/java/com/muyu/common/util/KafkaProviderConfig.java 100644
View File

@ -0,0 +1,38 @@
package com.muyu.common.util;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.common.serialization.Serializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.util.HashMap;
/**
* @author liuxinyue
* @Package：com.muyu.common.util
* @name：KafkaProviderConfig
* @Date：2024/9/29 20:15
*/
@Configuration
public class KafkaProviderConfig{
@Bean
public KafkaProducer KafkaProvider(){
HashMap<String, Object> map = new HashMap<>();
map.put("bootstrap.servers", "47.101.53.251:9092");
map.put("retries", "2");
map.put("batch.size", 16384);
map.put("buffer-memory", 33554432);
map.put("acks", "-1");
Serializer keySerializer = new StringSerializer();
//指定value使用的序列化类
Serializer valueSerializer = new StringSerializer();
//创建Kafka生产者
KafkaProducer kafkaProducer = new KafkaProducer(map, keySerializer, valueSerializer);
return kafkaProducer;
}
}

1
cloud-modules/saas/saas-server/src/main/java/com/muyu/server/SaasApplication.java
View File

@ -10,6 +10,7 @@ import org.springframework.boot.autoconfigure.SpringBootApplication;
* @Version 1.0
* @Data 2024-09-28 17:34:31
*/
@SpringBootApplication
@EnableMyFeignClients
public class SaasApplication {

5
cloud-modules/saas/saas-server/src/main/java/com/muyu/server/config/MqttConfigure.java
View File

@ -10,8 +10,7 @@ import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;
/**
*
* @author liuxinyue
* @Author：liuxinyue
* @Package：com.muyu.mqtt.configure
* @Project：cloud-server
* @name：MqttConfigure
@ -30,7 +29,7 @@ public class MqttConfigure {
String topic = "vehicle";
int qos = 2;
String broker = "tcp://47.101.53.251:1883";
String clientId = "测试mqtt";
String clientId = "hhhhhh";
try {
MqttClient sampleClient = new MqttClient(broker, clientId);
MqttConnectOptions connOpts = new MqttConnectOptions();

5
cloud-modules/saas/saas-server/src/main/java/com/muyu/server/controller/TemplateController.java
View File

@ -6,7 +6,6 @@ import com.muyu.server.service.TemplateService;
import io.swagger.v3.oas.annotations.Operation;
import io.swagger.v3.oas.annotations.tags.Tag;
import lombok.AllArgsConstructor;
import lombok.extern.log4j.Log4j2;
import org.apache.iotdb.rpc.IoTDBConnectionException;
import org.apache.iotdb.rpc.StatementExecutionException;
import org.springframework.beans.factory.annotation.Autowired;
@ -20,8 +19,7 @@ import java.util.List;
import java.util.concurrent.ExecutionException;
/**
* 报文模版管理
* @author liuxinyue
* @Author：liuxinyue
* @Package：com.template.controller
* @Project：cloud-server-c
* @name：TemplateController
@ -31,7 +29,6 @@ import java.util.concurrent.ExecutionException;
@RequestMapping("/template")
@AllArgsConstructor
@Tag(name = "报文模版管理",description = "报文模版管理")
@Log4j2
public class TemplateController {
@Autowired

13
cloud-modules/saas/saas-server/src/main/java/com/muyu/server/mapper/KafkaMapper.java 100644
View File

@ -0,0 +1,13 @@
package com.muyu.server.mapper;
import org.apache.ibatis.annotations.Mapper;
/**
* @author liuxinyue
* @Package：com.muyu.server.mapper
* @name：KafkaMapper
* @Date：2024/9/29 20:53
*/
@Mapper
public interface KafkaMapper {
}

14
cloud-modules/saas/saas-server/src/main/java/com/muyu/server/service/impl/MessageTemplateTypeServiceImpl.java
View File

@ -1,5 +1,4 @@
package com.muyu.server.service.impl;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.muyu.common.domain.MessageTemplateType;
@ -7,9 +6,7 @@ import com.muyu.server.mapper.MessageTemplateTypeMapper;
import com.muyu.server.service.MessageTemplateTypeService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import java.util.List;
/**
* @Author：liuxinyue
* @Package：com.sheep.message.service.impl
@ -33,7 +30,7 @@ public class MessageTemplateTypeServiceImpl extends ServiceImpl<MessageTemplate
@Override
public List<MessageTemplateType> findvehicleFoundationData(Integer templatedId, String code) {
QueryWrapper<MessageTemplateType> messageTemplateTypeQueryWrapper = new QueryWrapper<>();
messageTemplateTypeQueryWrapper.eq("templated_id",templatedId);
messageTemplateTypeQueryWrapper.eq("template_id",templatedId);
messageTemplateTypeQueryWrapper.eq("message_class",code);
List<MessageTemplateType> messageTemplateTypes = messageTemplateTypeMapper.selectList(messageTemplateTypeQueryWrapper);
return messageTemplateTypes;
@ -42,7 +39,7 @@ public class MessageTemplateTypeServiceImpl extends ServiceImpl<MessageTemplate
@Override
public List<MessageTemplateType> findvehicleData(Integer templatedId, String code) {
QueryWrapper<MessageTemplateType> messageTemplateTypeQueryWrapper = new QueryWrapper<>();
messageTemplateTypeQueryWrapper.eq("templated_id",templatedId);
messageTemplateTypeQueryWrapper.eq("template_id",templatedId);
messageTemplateTypeQueryWrapper.eq("message_class",code);
List<MessageTemplateType> messageTemplateTypes = messageTemplateTypeMapper.selectList(messageTemplateTypeQueryWrapper);
return messageTemplateTypes;
@ -51,7 +48,7 @@ public class MessageTemplateTypeServiceImpl extends ServiceImpl<MessageTemplate
@Override
public List<MessageTemplateType> finddeviceStatusData(Integer templatedId, String code) {
QueryWrapper<MessageTemplateType> messageTemplateTypeQueryWrapper = new QueryWrapper<>();
messageTemplateTypeQueryWrapper.eq("templated_id",templatedId);
messageTemplateTypeQueryWrapper.eq("template_id",templatedId);
messageTemplateTypeQueryWrapper.eq("message_class",code);
List<MessageTemplateType> messageTemplateTypes = messageTemplateTypeMapper.selectList(messageTemplateTypeQueryWrapper);
return messageTemplateTypes;
@ -60,7 +57,7 @@ public class MessageTemplateTypeServiceImpl extends ServiceImpl<MessageTemplate
@Override
public List<MessageTemplateType> findMessageByTemplateName(Integer templatedId) {
QueryWrapper<MessageTemplateType> messageTemplateTypeQueryWrapper = new QueryWrapper<>();
messageTemplateTypeQueryWrapper.eq("templated_id",templatedId);
messageTemplateTypeQueryWrapper.eq("template_id",templatedId);
List<MessageTemplateType> messageTemplateTypes = messageTemplateTypeMapper.selectList(messageTemplateTypeQueryWrapper);
return messageTemplateTypes;
}
@ -69,6 +66,7 @@ public class MessageTemplateTypeServiceImpl extends ServiceImpl<MessageTemplate
public List<MessageTemplateType> findTemplateById(Integer templateId) {
QueryWrapper<MessageTemplateType> messageTemplateTypeQueryWrapper = new QueryWrapper<>();
messageTemplateTypeQueryWrapper.eq("template_id",templateId);
return List.of();
List<MessageTemplateType> messageTemplateTypes = messageTemplateTypeMapper.selectList(messageTemplateTypeQueryWrapper);
return messageTemplateTypes;
}
}

16
cloud-modules/saas/saas-server/src/main/java/com/muyu/server/service/impl/TemplateServiceImpl.java
View File

@ -69,38 +69,40 @@ public class TemplateServiceImpl extends ServiceImpl<TemplateMapper, Template> i
log.info("车辆信息为:" + carByVin);
//对应车辆所对应的报文模版
Integer templateId = carByVin.getTemplateId();
List<MessageTemplateType> templateTypeList;
//key
String redisKey = "messageTemplateType" + templateId;
log.info("key为:" + redisKey);
//key存在
if (redisTemplate.hasKey(redisKey)) {
List list = redisTemplate.opsForList().range(redisKey, 0, -1);
templateTypeList = list.stream().map(o -> JSON.parseObject(o.toString(), MessageTemplateType.class))
.toList();
} else {
List<MessageTemplateType> templateTypeList1 = messageTemplateTypeService.findTemplateById(templateId);
log.info("redis存入成功");
templateTypeList = templateTypeList1;
templateTypeList.forEach(
templateTypeList1.forEach(
templateType ->
redisTemplate.opsForList().rightPush(
redisKey, com.alibaba.fastjson.JSON.toJSONString(templateType)
)
);
}
log.info("哈哈哈哈哈哈哈"+templateTypeList);
//将模版里面有的配置进行循环
for (MessageTemplateType messageTemplateType : templateTypeList) {
//开始位置
Integer startIndex = messageTemplateType.getStartIndex() - 1;
//结束位置
Integer endIndex = messageTemplateType.getEndIndex();
String substring = result.substring(startIndex, endIndex);
log.info("截取后的字符1:" + substring);
//将每个解析后的字段都存入到JSON对象中
jsonObject.put(messageTemplateType.getMessageField(), result.substring(startIndex, endIndex));
jsonObject.put(messageTemplateType.getMessageField(),substring );
}
System.out.println("哈哈哈红红火火恍恍惚惚");