feat: 新增根据vin将实时数据存入redis
parent
9459bb2576
commit
cb1abfa949
|
|
@ -0,0 +1,134 @@
|
|||
package com.muyu.eventdriven.constants;
|
||||
|
||||
/**
|
||||
* 通用常量信息
|
||||
*
|
||||
* @author muyu
|
||||
*/
|
||||
public class Constants {
|
||||
/**
|
||||
* UTF-8 字符集
|
||||
*/
|
||||
public static final String UTF8 = "UTF-8";
|
||||
|
||||
/**
|
||||
* GBK 字符集
|
||||
*/
|
||||
public static final String GBK = "GBK";
|
||||
|
||||
/**
|
||||
* www主域
|
||||
*/
|
||||
public static final String WWW = "www.";
|
||||
|
||||
/**
|
||||
* RMI 远程方法调用
|
||||
*/
|
||||
public static final String LOOKUP_RMI = "rmi:";
|
||||
|
||||
/**
|
||||
* LDAP 远程方法调用
|
||||
*/
|
||||
public static final String LOOKUP_LDAP = "ldap:";
|
||||
|
||||
/**
|
||||
* LDAPS 远程方法调用
|
||||
*/
|
||||
public static final String LOOKUP_LDAPS = "ldaps:";
|
||||
|
||||
/**
|
||||
* http请求
|
||||
*/
|
||||
public static final String HTTP = "http://";
|
||||
|
||||
/**
|
||||
* https请求
|
||||
*/
|
||||
public static final String HTTPS = "https://";
|
||||
|
||||
/**
|
||||
* 成功标记
|
||||
*/
|
||||
public static final Integer SUCCESS = 200;
|
||||
|
||||
/**
|
||||
* 失败标记
|
||||
*/
|
||||
public static final Integer FAIL = 500;
|
||||
|
||||
/**
|
||||
* 登录成功状态
|
||||
*/
|
||||
public static final String LOGIN_SUCCESS_STATUS = "0";
|
||||
|
||||
/**
|
||||
* 登录失败状态
|
||||
*/
|
||||
public static final String LOGIN_FAIL_STATUS = "1";
|
||||
|
||||
/**
|
||||
* 登录成功
|
||||
*/
|
||||
public static final String LOGIN_SUCCESS = "Success";
|
||||
|
||||
/**
|
||||
* 注销
|
||||
*/
|
||||
public static final String LOGOUT = "Logout";
|
||||
|
||||
/**
|
||||
* 注册
|
||||
*/
|
||||
public static final String REGISTER = "Register";
|
||||
|
||||
/**
|
||||
* 登录失败
|
||||
*/
|
||||
public static final String LOGIN_FAIL = "Error";
|
||||
|
||||
/**
|
||||
* 当前记录起始索引
|
||||
*/
|
||||
public static final String PAGE_NUM = "pageNum";
|
||||
|
||||
/**
|
||||
* 每页显示记录数
|
||||
*/
|
||||
public static final String PAGE_SIZE = "pageSize";
|
||||
|
||||
/**
|
||||
* 排序列
|
||||
*/
|
||||
public static final String ORDER_BY_COLUMN = "orderByColumn";
|
||||
|
||||
/**
|
||||
* 排序的方向 "desc" 或者 "asc".
|
||||
*/
|
||||
public static final String IS_ASC = "isAsc";
|
||||
|
||||
/**
|
||||
* 验证码有效期（分钟）
|
||||
*/
|
||||
public static final long CAPTCHA_EXPIRATION = 2;
|
||||
|
||||
/**
|
||||
* 资源映射路径 前缀
|
||||
*/
|
||||
public static final String RESOURCE_PREFIX = "/profile";
|
||||
|
||||
/**
|
||||
* 自动识别json对象白名单配置（仅允许解析的包名，范围越小越安全）
|
||||
*/
|
||||
public static final String[] JSON_WHITELIST_STR = {"org.springframework", "com.muyu"};
|
||||
|
||||
/**
|
||||
* 定时任务白名单配置（仅允许访问的包名，如其他需要可以自行添加）
|
||||
*/
|
||||
public static final String[] JOB_WHITELIST_STR = {"com.muyu"};
|
||||
|
||||
/**
|
||||
* 定时任务违规的字符
|
||||
*/
|
||||
public static final String[] JOB_ERROR_STR = {"java.net.URL", "javax.naming.InitialContext", "org.yaml.snakeyaml",
|
||||
"org.springframework", "org.apache", "com.muyu.common.core.utils.file"};
|
||||
}
|
||||
|
|
@ -0,0 +1,93 @@
|
|||
package com.muyu.eventdriven.constants;
|
||||
|
||||
/**
|
||||
* 返回状态码
|
||||
*
|
||||
* @author muyu
|
||||
*/
|
||||
public class HttpStatus {
|
||||
/**
|
||||
* 操作成功
|
||||
*/
|
||||
public static final int SUCCESS = 200;
|
||||
|
||||
/**
|
||||
* 对象创建成功
|
||||
*/
|
||||
public static final int CREATED = 201;
|
||||
|
||||
/**
|
||||
* 请求已经被接受
|
||||
*/
|
||||
public static final int ACCEPTED = 202;
|
||||
|
||||
/**
|
||||
* 操作已经执行成功，但是没有返回数据
|
||||
*/
|
||||
public static final int NO_CONTENT = 204;
|
||||
|
||||
/**
|
||||
* 资源已被移除
|
||||
*/
|
||||
public static final int MOVED_PERM = 301;
|
||||
|
||||
/**
|
||||
* 重定向
|
||||
*/
|
||||
public static final int SEE_OTHER = 303;
|
||||
|
||||
/**
|
||||
* 资源没有被修改
|
||||
*/
|
||||
public static final int NOT_MODIFIED = 304;
|
||||
|
||||
/**
|
||||
* 参数列表错误（缺少，格式不匹配）
|
||||
*/
|
||||
public static final int BAD_REQUEST = 400;
|
||||
|
||||
/**
|
||||
* 未授权
|
||||
*/
|
||||
public static final int UNAUTHORIZED = 401;
|
||||
|
||||
/**
|
||||
* 访问受限，授权过期
|
||||
*/
|
||||
public static final int FORBIDDEN = 403;
|
||||
|
||||
/**
|
||||
* 资源，服务未找到
|
||||
*/
|
||||
public static final int NOT_FOUND = 404;
|
||||
|
||||
/**
|
||||
* 不允许的http方法
|
||||
*/
|
||||
public static final int BAD_METHOD = 405;
|
||||
|
||||
/**
|
||||
* 资源冲突，或者资源被锁
|
||||
*/
|
||||
public static final int CONFLICT = 409;
|
||||
|
||||
/**
|
||||
* 不支持的数据，媒体类型
|
||||
*/
|
||||
public static final int UNSUPPORTED_TYPE = 415;
|
||||
|
||||
/**
|
||||
* 系统内部错误
|
||||
*/
|
||||
public static final int ERROR = 500;
|
||||
|
||||
/**
|
||||
* 接口未实现
|
||||
*/
|
||||
public static final int NOT_IMPLEMENTED = 501;
|
||||
|
||||
/**
|
||||
* 系统警告消息
|
||||
*/
|
||||
public static final int WARN = 601;
|
||||
}
|
||||
|
|
@ -1,14 +1,11 @@
|
|||
package com.muyu.eventdriven.consumer;
|
||||
|
||||
import com.alibaba.fastjson.JSON;
|
||||
import com.muyu.eventdriven.domain.VehicleKafka;
|
||||
import org.apache.kafka.clients.consumer.ConsumerConfig;
|
||||
import org.apache.kafka.clients.consumer.KafkaConsumer;
|
||||
import org.apache.kafka.common.TopicPartition;
|
||||
import org.apache.kafka.common.serialization.StringDeserializer;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.beans.factory.annotation.Value;
|
||||
import org.springframework.data.redis.core.RedisTemplate;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import java.util.ArrayList;
|
||||
|
|
@ -26,14 +23,11 @@ public class KafkaConsumers {
|
|||
@Value("${spring.kafka.bootstrap-servers}")
|
||||
private String bootstrapServers;
|
||||
|
||||
@Autowired
|
||||
private RedisTemplate<String,String> redisTemplate;
|
||||
|
||||
|
||||
|
||||
public KafkaConsumer kafkaConsumer(String vin){
|
||||
Object o = redisTemplate.opsForHash().get("vehicleKafka", vin);
|
||||
VehicleKafka vehicleKafka = JSON.parseObject(o.toString(), VehicleKafka.class);
|
||||
|
||||
public KafkaConsumer kafkaConsumer(String vin,VehicleKafka vehicleKafka){
|
||||
Properties properties = new Properties();
|
||||
properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
|
||||
properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
|
||||
|
|
@ -47,6 +41,8 @@ public class KafkaConsumers {
|
|||
List<TopicPartition> topicPartitions = new ArrayList<>();
|
||||
topicPartitions.add(new TopicPartition(vehicleKafka.getConsumerName(), vehicleKafka.getPartitions()));
|
||||
consumer.assign(topicPartitions);
|
||||
// 设置每个分区的位移为latest，即只消费启动后发布的消息
|
||||
consumer.seekToEnd(topicPartitions);
|
||||
return consumer;
|
||||
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,6 +1,11 @@
|
|||
package com.muyu.eventdriven.controller;
|
||||
|
||||
import com.muyu.eventdriven.domain.rest.Result;
|
||||
import com.muyu.eventdriven.server.EventInfoService;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.web.bind.annotation.GetMapping;
|
||||
import org.springframework.web.bind.annotation.RequestMapping;
|
||||
import org.springframework.web.bind.annotation.RequestParam;
|
||||
import org.springframework.web.bind.annotation.RestController;
|
||||
|
||||
/**
|
||||
|
|
@ -12,5 +17,16 @@ import org.springframework.web.bind.annotation.RestController;
|
|||
@RestController
|
||||
@RequestMapping("/eventInfo")
|
||||
public class EventInfoController {
|
||||
@Autowired
|
||||
private EventInfoService eventInfoService;
|
||||
|
||||
@GetMapping("/CreatKafkaConsumer")
|
||||
public void creatKafkaConsumer(@RequestParam("vin") String vin) {
|
||||
eventInfoService.creatKafkaConsumer(vin);
|
||||
}
|
||||
|
||||
@GetMapping("/CloseKafkaConsumer")
|
||||
public Result closeKafkaConsumer(@RequestParam("vin") String vin) {
|
||||
return eventInfoService.closeKafkaConsumer(vin);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,112 @@
|
|||
package com.muyu.eventdriven.domain.rest;
|
||||
|
||||
import com.muyu.eventdriven.constants.Constants;
|
||||
import com.muyu.eventdriven.constants.HttpStatus;
|
||||
import lombok.AllArgsConstructor;
|
||||
import lombok.Builder;
|
||||
import lombok.Data;
|
||||
import lombok.NoArgsConstructor;
|
||||
|
||||
import java.io.Serializable;
|
||||
|
||||
/**
|
||||
* 响应信息主体
|
||||
*
|
||||
* @author muyu
|
||||
*/
|
||||
@Data
|
||||
@Builder
|
||||
@NoArgsConstructor
|
||||
@AllArgsConstructor
|
||||
public class Result<T> implements Serializable {
|
||||
/**
|
||||
* 成功
|
||||
*/
|
||||
public static final int SUCCESS = Constants.SUCCESS;
|
||||
/**
|
||||
* 失败
|
||||
*/
|
||||
public static final int FAIL = Constants.FAIL;
|
||||
/**
|
||||
* 警告
|
||||
*/
|
||||
public static final int WARN = HttpStatus.WARN;
|
||||
|
||||
private static final long serialVersionUID = 1L;
|
||||
private int code;
|
||||
|
||||
private String msg;
|
||||
|
||||
private T data;
|
||||
|
||||
public static <T> Result<T> success () {
|
||||
return restResult(null, SUCCESS, null);
|
||||
}
|
||||
|
||||
public static <T> Result<T> success (T data) {
|
||||
return restResult(data, SUCCESS, null);
|
||||
}
|
||||
|
||||
public static <T> Result<T> success (T data, String msg) {
|
||||
return restResult(data, SUCCESS, msg);
|
||||
}
|
||||
|
||||
public static <T> Result<T> error () {
|
||||
return restResult(null, FAIL, null);
|
||||
}
|
||||
|
||||
public static <T> Result<T> error (String msg) {
|
||||
return restResult(null, FAIL, msg);
|
||||
}
|
||||
|
||||
public static <T> Result<T> error (T data) {
|
||||
return restResult(data, FAIL, null);
|
||||
}
|
||||
|
||||
public static <T> Result<T> error (T data, String msg) {
|
||||
return restResult(data, FAIL, msg);
|
||||
}
|
||||
|
||||
public static <T> Result<T> error (int code, String msg) {
|
||||
return restResult(null, code, msg);
|
||||
}
|
||||
|
||||
|
||||
|
||||
public static <T> Result<T> warn () {
|
||||
return restResult(null, WARN, null);
|
||||
}
|
||||
|
||||
public static <T> Result<T> warn (String msg) {
|
||||
return restResult(null, WARN, msg);
|
||||
}
|
||||
|
||||
public static <T> Result<T> warn (T data) {
|
||||
return restResult(data, WARN, null);
|
||||
}
|
||||
|
||||
public static <T> Result<T> warn (T data, String msg) {
|
||||
return restResult(data, WARN, msg);
|
||||
}
|
||||
|
||||
public static <T> Result<T> warn (int code, String msg) {
|
||||
return restResult(null, code, msg);
|
||||
}
|
||||
|
||||
private static <T> Result<T> restResult (T data, int code, String msg) {
|
||||
return Result.<T>builder()
|
||||
.code(code)
|
||||
.data(data)
|
||||
.msg(msg)
|
||||
.build();
|
||||
}
|
||||
|
||||
public static <T> Boolean isError (Result<T> ret) {
|
||||
return !isSuccess(ret);
|
||||
}
|
||||
|
||||
public static <T> Boolean isSuccess (Result<T> ret) {
|
||||
return Result.SUCCESS == ret.getCode();
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -0,0 +1,15 @@
|
|||
package com.muyu.eventdriven.server;
|
||||
|
||||
import com.muyu.eventdriven.domain.rest.Result;
|
||||
|
||||
/**
|
||||
* @ClassName EventInfo
|
||||
* @Description 事件接口
|
||||
* @Author Xin.Yao
|
||||
* @Date 2024/6/18 上午9:34
|
||||
*/
|
||||
public interface EventInfoService {
|
||||
void creatKafkaConsumer(String vin);
|
||||
|
||||
Result closeKafkaConsumer(String vin);
|
||||
}
|
||||
|
|
@ -0,0 +1,68 @@
|
|||
package com.muyu.eventdriven.server.impl;
|
||||
|
||||
import com.alibaba.fastjson.JSON;
|
||||
import com.muyu.eventdriven.consumer.KafkaConsumers;
|
||||
import com.muyu.eventdriven.domain.VehicleData;
|
||||
import com.muyu.eventdriven.domain.VehicleKafka;
|
||||
import com.muyu.eventdriven.domain.rest.Result;
|
||||
import com.muyu.eventdriven.mapper.EventDrivenMapper;
|
||||
import com.muyu.eventdriven.server.EventInfoService;
|
||||
import lombok.extern.log4j.Log4j2;
|
||||
import org.apache.kafka.clients.consumer.ConsumerRecord;
|
||||
import org.apache.kafka.clients.consumer.ConsumerRecords;
|
||||
import org.apache.kafka.clients.consumer.KafkaConsumer;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.data.redis.core.RedisTemplate;
|
||||
import org.springframework.stereotype.Service;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.util.HashMap;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
/**
|
||||
* @ClassName EventInfoServiceImpl
|
||||
* @Description 事件实现层
|
||||
* @Author Xin.Yao
|
||||
* @Date 2024/6/18 上午9:35
|
||||
*/
|
||||
@Service
|
||||
@Log4j2
|
||||
public class EventInfoServiceImpl implements EventInfoService {
|
||||
private static Map<String,KafkaConsumer> kafkaConsumerMap = new HashMap<>();
|
||||
|
||||
@Autowired
|
||||
private KafkaConsumers kafkaConsumers;
|
||||
@Autowired
|
||||
private RedisTemplate<String,String> redisTemplate;
|
||||
@Autowired
|
||||
private EventDrivenMapper eventDrivenMapper;
|
||||
|
||||
@Override
|
||||
public void creatKafkaConsumer(String vin) {
|
||||
Object o = redisTemplate.opsForHash().get("vehicleKafka", vin);
|
||||
VehicleKafka vehicleKafka = JSON.parseObject(o.toString(), VehicleKafka.class);
|
||||
redisTemplate.opsForList().rightPush(vin,"");
|
||||
redisTemplate.expire(vin, 10, TimeUnit.HOURS);
|
||||
if (!kafkaConsumerMap.containsKey(vehicleKafka.getConsumerName() + "-" + vehicleKafka.getPartitions())) {
|
||||
KafkaConsumer kafkaConsumer = kafkaConsumers.kafkaConsumer(vin,vehicleKafka);
|
||||
kafkaConsumerMap.put(vehicleKafka.getConsumerName()+"-"+vehicleKafka.getPartitions(),kafkaConsumer);
|
||||
while (kafkaConsumerMap.containsKey(vehicleKafka.getConsumerName()+"-"+vehicleKafka.getPartitions())){
|
||||
// 拉取消息
|
||||
ConsumerRecords<String, String> msg = kafkaConsumer.poll(Duration.ofSeconds(1));
|
||||
for (ConsumerRecord<String, String> consumerRecord : msg) {
|
||||
if (redisTemplate.hasKey(vin)){
|
||||
log.info("{}监听到的消息内容: {}", vin,consumerRecord.value());
|
||||
redisTemplate.opsForList().rightPush(vin,consumerRecord.value());
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public Result closeKafkaConsumer(String vin) {
|
||||
redisTemplate.delete(vin);
|
||||
return Result.success("释放消费者");
|
||||
}
|
||||
}
|
||||
|
|
@ -14,7 +14,7 @@ spring:
|
|||
#这个可以和config/consumer.properties里的group.id不同
|
||||
group-id: test-consumer-group
|
||||
redis:
|
||||
host: 127.0.0.1
|
||||
host: 101.34.248.9
|
||||
port: 6379
|
||||
password:
|
||||
datasource:
|
||||
|
|
|
|||
Loading…
Reference in New Issue