osc-group/osc
1
1
Fork 0
Code Pull Requests Projects Wiki Activity

Update StorageService.java

Browse Source 
入库MySQL
This commit is contained in:
yuxin-pc 2025-12-23 19:23:15 +08:00
parent afe6c34db7
commit 9f3eb9cf94
1 changed files with 33 additions and 67 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

100
dsp/src/main/java/com/jsc/dsp/service/StorageService.java
View File

@ -20,6 +20,7 @@ import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.common.xcontent.XContentType; import org.elasticsearch.common.xcontent.XContentType;
import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value; import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.json.JacksonJsonParser; import org.springframework.boot.json.JacksonJsonParser;
import org.springframework.cloud.stream.annotation.EnableBinding; import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener; import org.springframework.cloud.stream.annotation.StreamListener;
@ -33,6 +34,7 @@ import java.util.Base64.Decoder;
@Component @Component
@EnableBinding(StorageBinding.class) @EnableBinding(StorageBinding.class)
@ConditionalOnProperty(name = "switch.enable-storage-service", havingValue = "true", matchIfMissing = true)
public class StorageService extends StreamService { public class StorageService extends StreamService {
@Autowired @Autowired
@ -44,27 +46,15 @@ public class StorageService extends StreamService {
@Autowired @Autowired
JacksonJsonParser jsonParser; JacksonJsonParser jsonParser;
@Value("${es.ip}")
String esIp;
@Value("${es.port}")
Integer esPort;
@Value("${es.username}")
String esUsername;
@Value("${es.password}")
String esPassword;
@Value("${es.index}")
String esIndex;
@Value("${custom.dev-mode}") @Value("${custom.dev-mode}")
boolean devMode; boolean devMode;
@Value("${custom.local-file-storage-path}") @Value("${custom.local-file-storage-path}")
String localFileStoragePath; String localFileStoragePath;
@Value("${custom.websiteWhiteList}")
String websiteWhiteListString;
@Value("${db.driver}") @Value("${db.driver}")
String dbDriver; String dbDriver;
@ -80,6 +70,7 @@ public class StorageService extends StreamService {
private final Logger logger = LogManager.getLogger(StorageService.class.getName()); private final Logger logger = LogManager.getLogger(StorageService.class.getName());
@Override @Override
public void sendMessage(byte[] msg) { public void sendMessage(byte[] msg) {
source.StorageOutput().send(MessageBuilder.withPayload(msg).build()); source.StorageOutput().send(MessageBuilder.withPayload(msg).build());
@ -91,8 +82,8 @@ public class StorageService extends StreamService {
@Override @Override
@StreamListener(StorageBinding.STORAGE_PIPELINE_IN) @StreamListener(StorageBinding.STORAGE_PIPELINE_IN)
public void receiveMessage(Object payload) { public void receiveMessage(Object payload) {
List<String> websiteWhiteList = Arrays.asList(websiteWhiteListString.split(";"));
String tempString; String tempString;
ObjectMapper objectMapper = new ObjectMapper();
try { try {
tempString = new String(base64.decode(payload.toString()), StandardCharsets.UTF_8); tempString = new String(base64.decode(payload.toString()), StandardCharsets.UTF_8);
Map<String, Object> data = jsonParser.parseMap(tempString); Map<String, Object> data = jsonParser.parseMap(tempString);
@ -101,7 +92,6 @@ public class StorageService extends StreamService {
if ("public_info_data_".equals(protoName)) { if ("public_info_data_".equals(protoName)) {
EsSets.Builder esSetsBuilder = EsSets.newBuilder(); EsSets.Builder esSetsBuilder = EsSets.newBuilder();
EsSets esSets = EsSets.parseFrom(data.get("content").toString().getBytes(StandardCharsets.ISO_8859_1)); EsSets esSets = EsSets.parseFrom(data.get("content").toString().getBytes(StandardCharsets.ISO_8859_1));
List<Object> localStorageItems = new ArrayList<>();
List<Indeximos> dbStorageItems = new ArrayList<>(); List<Indeximos> dbStorageItems = new ArrayList<>();
BulkRequest bulkRequest = new BulkRequest(); BulkRequest bulkRequest = new BulkRequest();
bulkRequest.timeout("5s"); bulkRequest.timeout("5s");
@ -138,61 +128,37 @@ public class StorageService extends StreamService {
} }
} }
} }
String uuid = UUID.randomUUID().toString().replaceAll("-", ""); // 只导出目标站点的数据
String es_urlname = indeximos.getEs_urlname(); if (websiteWhiteList.contains(indeximos.getEs_sitename())) {
if (!es_urlname.isEmpty()) { logger.info("开始处理站点【" + indeximos.getEs_sitename() + "】的数据入库流程");
// 根据urlname生成固定的UUID，避免重复入库相同的文章 String uuid = UUID.randomUUID().toString().replaceAll("-", "");
UUID _uuid = UUID.nameUUIDFromBytes(es_urlname.getBytes()); String es_urlname = indeximos.getEs_urlname();
uuid = _uuid.toString().replaceAll("-", ""); if (!es_urlname.isEmpty()) {
} // 根据urlname生成固定的UUID，避免重复入库相同的文章
indeximos.setEs_sid(uuid); UUID _uuid = UUID.nameUUIDFromBytes(es_urlname.getBytes());
indeximos.setEs_links(indeximos.getEs_links()); uuid = _uuid.toString().replaceAll("-", "");
indeximos.setEs_loadtime(StringUtils.TimestampToStringDate(System.currentTimeMillis())); }
builder.setEsSid(uuid); indeximos.setEs_urltitle(indeximos.getEs_urltitle().trim());
for (Field f : indeximos.getClass().getDeclaredFields()) { indeximos.setEs_sid(uuid);
f.setAccessible(true); indeximos.setEs_links(indeximos.getEs_links());
//判断字段是否为空，并且对象属性中的基本都会转为对象类型来判断 indeximos.setEs_loadtime(StringUtils.TimestampToStringDate(System.currentTimeMillis()));
if (f.get(indeximos) == null) { builder.setEsSid(uuid);
String fieldType = DBUtils.getFieldType(Indeximos.class, f.getName()); for (Field f : indeximos.getClass().getDeclaredFields()) {
if (fieldType.contains("Float")) { f.setAccessible(true);
f.set(indeximos, 0.0f); //判断字段是否为空，并且对象属性中的基本都会转为对象类型来判断
} else { if (f.get(indeximos) == null) {
if (!dateFields.contains(f.getName())) { String fieldType = DBUtils.getFieldType(Indeximos.class, f.getName());
f.set(indeximos, ""); if (fieldType.contains("Float")) {
f.set(indeximos, 0.0f);
} else {
if (!dateFields.contains(f.getName())) {
f.set(indeximos, "");
}
} }
} }
} }
}
IndexRequest indexRequest = new IndexRequest(esIndex);
indexRequest.id(indeximos.getEs_sid());
indexRequest.source(objectMapper.writeValueAsString(indeximos), XContentType.JSON);
bulkRequest.add(indexRequest);
Es es_temp = builder.build();
esSetsBuilder.addEs(es_temp);
List<String> localizedOption = JSON.parseArray(indeximos.getEs_urltopic(), String.class);
if (indeximos.getEs_carriertype().equals("wechat")) {
dbStorageItems.add(indeximos); dbStorageItems.add(indeximos);
} }
if (localizedOption != null && localizedOption.size() > 0) {
//本地存储用
if (localizedOption.contains("json")) {
localStorageItems.add(indeximos);
}
//入库MySQL
if (localizedOption.contains("mysql")) {
dbStorageItems.add(indeximos);
}
}
}
EsUtils.EsSaveBulkRequest(esIp, esPort, esUsername, esPassword, bulkRequest);
if (localStorageItems.size() > 0) {
String entityItemsString = JSON.toJSONString(localStorageItems);
String entityFileFullPath = localFileStoragePath + esIndex + "_" + System.currentTimeMillis() + ".json";
if (FileUtils.saveStringToFile(entityItemsString, entityFileFullPath)) {
logger.info("Local file store to " + entityFileFullPath);
} else {
logger.error("Local file store error!");
}
} }
if (dbStorageItems.size() > 0) { if (dbStorageItems.size() > 0) {
if (DBUtils.insertIntoDB(dbDriver, dbUrl, dbUser, dbPassword, dbStorageItems)) { if (DBUtils.insertIntoDB(dbDriver, dbUrl, dbUser, dbPassword, dbStorageItems)) {