package cn.iocoder.yudao.module.iot.service.rule.data.action;
|
|
import cn.hutool.core.util.StrUtil;
|
import cn.iocoder.yudao.framework.common.util.json.JsonUtils;
|
import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
|
import cn.iocoder.yudao.module.iot.dal.dataobject.rule.config.IotDataSinkRedisConfig;
|
import cn.iocoder.yudao.module.iot.enums.rule.IotDataSinkTypeEnum;
|
import cn.iocoder.yudao.module.iot.enums.rule.IotRedisDataStructureEnum;
|
import lombok.extern.slf4j.Slf4j;
|
import org.redisson.Redisson;
|
import org.redisson.api.RedissonClient;
|
import org.redisson.config.Config;
|
import org.redisson.config.SingleServerConfig;
|
import org.redisson.spring.data.connection.RedissonConnectionFactory;
|
import org.springframework.data.redis.connection.RedisConnectionFactory;
|
import org.springframework.data.redis.connection.stream.ObjectRecord;
|
import org.springframework.data.redis.connection.stream.StreamRecords;
|
import org.springframework.data.redis.core.RedisTemplate;
|
import org.springframework.data.redis.serializer.RedisSerializer;
|
import org.springframework.stereotype.Component;
|
|
import java.util.Map;
|
|
/**
|
* Redis 的 {@link IotDataRuleAction} 实现类
|
* 支持多种 Redis 数据结构：Stream、Hash、List、Set、ZSet、String
|
*
|
* @author HUIHUI
|
*/
|
@Component
|
@Slf4j
|
public class IotRedisRuleAction extends
|
IotDataRuleCacheableAction<IotDataSinkRedisConfig, RedisTemplate<String, Object>> {
|
|
@Override
|
public Integer getType() {
|
return IotDataSinkTypeEnum.REDIS.getType();
|
}
|
|
@Override
|
public void execute(IotDeviceMessage message, IotDataSinkRedisConfig config) throws Exception {
|
// 1. 获取 RedisTemplate
|
RedisTemplate<String, Object> redisTemplate = getProducer(config);
|
|
// 2. 根据数据结构类型执行不同的操作
|
String messageJson = JsonUtils.toJsonString(message);
|
IotRedisDataStructureEnum dataStructure = getDataStructureByType(config.getDataStructure());
|
switch (dataStructure) {
|
case STREAM:
|
executeStream(redisTemplate, config, messageJson);
|
break;
|
case HASH:
|
executeHash(redisTemplate, config, message, messageJson);
|
break;
|
case LIST:
|
executeList(redisTemplate, config, messageJson);
|
break;
|
case SET:
|
executeSet(redisTemplate, config, messageJson);
|
break;
|
case ZSET:
|
executeZSet(redisTemplate, config, message, messageJson);
|
break;
|
case STRING:
|
executeString(redisTemplate, config, messageJson);
|
break;
|
default:
|
throw new IllegalArgumentException("不支持的 Redis 数据结构类型: " + dataStructure);
|
}
|
|
log.info("[execute][消息发送成功] dataStructure: {}, config: {}", dataStructure.getName(), config);
|
}
|
|
/**
|
* 执行 Stream 操作
|
*/
|
private void executeStream(RedisTemplate<String, Object> redisTemplate, IotDataSinkRedisConfig config, String messageJson) {
|
ObjectRecord<String, ?> record = StreamRecords.newRecord()
|
.ofObject(messageJson).withStreamKey(config.getTopic());
|
redisTemplate.opsForStream().add(record);
|
}
|
|
/**
|
* 执行 Hash 操作
|
*/
|
private void executeHash(RedisTemplate<String, Object> redisTemplate, IotDataSinkRedisConfig config,
|
IotDeviceMessage message, String messageJson) {
|
String hashField = StrUtil.isNotBlank(config.getHashField()) ?
|
config.getHashField() : String.valueOf(message.getDeviceId());
|
redisTemplate.opsForHash().put(config.getTopic(), hashField, messageJson);
|
}
|
|
/**
|
* 执行 List 操作
|
*/
|
private void executeList(RedisTemplate<String, Object> redisTemplate, IotDataSinkRedisConfig config, String messageJson) {
|
redisTemplate.opsForList().rightPush(config.getTopic(), messageJson);
|
}
|
|
/**
|
* 执行 Set 操作
|
*/
|
private void executeSet(RedisTemplate<String, Object> redisTemplate, IotDataSinkRedisConfig config, String messageJson) {
|
redisTemplate.opsForSet().add(config.getTopic(), messageJson);
|
}
|
|
/**
|
* 执行 ZSet 操作
|
*/
|
private void executeZSet(RedisTemplate<String, Object> redisTemplate, IotDataSinkRedisConfig config,
|
IotDeviceMessage message, String messageJson) {
|
double score;
|
if (StrUtil.isNotBlank(config.getScoreField())) {
|
// 尝试从消息中获取分数字段
|
try {
|
Map<String, Object> messageMap = JsonUtils.parseObject(messageJson, Map.class);
|
Object scoreValue = messageMap.get(config.getScoreField());
|
score = scoreValue instanceof Number ? ((Number) scoreValue).doubleValue() : System.currentTimeMillis();
|
} catch (Exception e) {
|
score = System.currentTimeMillis();
|
}
|
} else {
|
// 使用当前时间戳作为分数
|
score = System.currentTimeMillis();
|
}
|
redisTemplate.opsForZSet().add(config.getTopic(), messageJson, score);
|
}
|
|
/**
|
* 执行 String 操作
|
*/
|
private void executeString(RedisTemplate<String, Object> redisTemplate, IotDataSinkRedisConfig config, String messageJson) {
|
redisTemplate.opsForValue().set(config.getTopic(), messageJson);
|
}
|
|
@Override
|
protected RedisTemplate<String, Object> initProducer(IotDataSinkRedisConfig config) {
|
// 1.1 创建 Redisson 配置
|
Config redissonConfig = new Config();
|
SingleServerConfig serverConfig = redissonConfig.useSingleServer()
|
.setAddress("redis://" + config.getHost() + ":" + config.getPort())
|
.setDatabase(config.getDatabase());
|
// 1.2 设置密码（如果有）
|
if (StrUtil.isNotBlank(config.getPassword())) {
|
serverConfig.setPassword(config.getPassword());
|
}
|
|
// 2.1 创建 RedisTemplate 并配置
|
RedissonClient redisson = Redisson.create(redissonConfig);
|
RedisTemplate<String, Object> template = new RedisTemplate<>();
|
template.setConnectionFactory(new RedissonConnectionFactory(redisson));
|
// 2.2 设置序列化器
|
template.setKeySerializer(RedisSerializer.string());
|
template.setHashKeySerializer(RedisSerializer.string());
|
template.setValueSerializer(RedisSerializer.json());
|
template.setHashValueSerializer(RedisSerializer.json());
|
template.afterPropertiesSet();
|
return template;
|
}
|
|
@Override
|
protected void closeProducer(RedisTemplate<String, Object> producer) throws Exception {
|
RedisConnectionFactory factory = producer.getConnectionFactory();
|
if (factory != null) {
|
((RedissonConnectionFactory) factory).destroy();
|
}
|
}
|
|
/**
|
* 根据类型值获取数据结构枚举
|
*/
|
private IotRedisDataStructureEnum getDataStructureByType(Integer type) {
|
for (IotRedisDataStructureEnum dataStructure : IotRedisDataStructureEnum.values()) {
|
if (dataStructure.getType().equals(type)) {
|
return dataStructure;
|
}
|
}
|
throw new IllegalArgumentException("不支持的 Redis 数据结构类型: " + type);
|
}
|
|
}
|
