ziin/TKDataSave
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

入库方法修改为多线程，大幅提高入库效率

Browse Source
This commit is contained in:
Ziin
2025-06-19 21:51:25 +08:00
parent 6268391baf
commit 167c1ec29e
5 changed files with 52 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/main/java/com/yupi/springbootinit/MainApplication.java
View File

@@ -5,6 +5,7 @@ import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.autoconfigure.data.redis.RedisAutoConfiguration;
import org.springframework.context.annotation.EnableAspectJAutoProxy;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.annotation.EnableScheduling;
/**
@@ -17,6 +18,7 @@ import org.springframework.scheduling.annotation.EnableScheduling;
@SpringBootApplication()
@MapperScan("com.yupi.springbootinit.mapper")
@EnableScheduling
@EnableAsync
@EnableAspectJAutoProxy(proxyTargetClass = true, exposeProxy = true)
public class MainApplication {

8
src/main/java/com/yupi/springbootinit/rabbitMQ/MQReceiver.java
View File

@@ -10,6 +10,7 @@ import com.yupi.springbootinit.service.HostInfoService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import javax.annotation.Resource;
@@ -39,14 +40,15 @@ public class MQReceiver {
// }
@RabbitListener(queues = "HOST_INFO_QUEUE")
@Async
public void receive(List<NewHosts> hosts, Channel channel, Message message) throws IOException {
long deliveryTag = message.getMessageProperties().getDeliveryTag();
try {
// 等待所有异步任务完成
hostInfoService.processHosts(hosts).join(); // 这里会抛出异常
hostInfoService.processHosts(hosts); // 这里会抛出异常
channel.basicAck(deliveryTag, false);
log.info("deliveryTag:{}", deliveryTag);
log.info("{} 消息消费内容大小-------> {}",DateTime.now(),hosts.size());
// log.info("deliveryTag:{}", deliveryTag);
// log.info("{} 消息消费内容大小-------> {}",DateTime.now(),hosts.size());
} catch (Exception e) {
channel.basicNack(deliveryTag, false, false);
log.error("消息消费失败------->消息内容大小{}", hosts.size(), e);

2
src/main/java/com/yupi/springbootinit/rabbitMQ/MQSender.java
View File

@@ -23,7 +23,7 @@ public class MQSender {
//方法:发送消息
public void send(List<NewHosts> list){
try {
log.info("{} 接收到的消息数量----------->{}", DateTime.now(),list.size());
// log.info("{} 接收到的消息数量----------->{}", DateTime.now(),list.size());
this.rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());
//指定你队列的名字
rabbitTemplate.convertAndSend("HOST_INFO_QUEUE",list);

34
src/main/java/com/yupi/springbootinit/service/impl/HostInfoServiceImpl.java
View File

@@ -56,7 +56,7 @@ public class HostInfoServiceImpl extends ServiceImpl<NewHostsMapper, NewHosts> i
saveBatch(newHosts);
stopWatch.stop();
long totalTimeMillis = stopWatch.getTotalTimeMillis();
log.info("存储花费: {}ms", totalTimeMillis);
log.info("当前存储数据量大小 {}, 存储花费: {}ms",newHosts.size(),totalTimeMillis);
return CompletableFuture.completedFuture(null);
} catch (Exception e) {
// 将异常包装到Future，使调用方能处理
@@ -82,21 +82,25 @@ public class HostInfoServiceImpl extends ServiceImpl<NewHostsMapper, NewHosts> i
// });
// }
public CompletableFuture<Void> processHosts(List<NewHosts> hosts) {
List<CompletableFuture<Void>> futures = new ArrayList<>();
this.saveHostInfo(hosts);
// log.info("当前存储数据量大小 {}",hosts.size());
// List<CompletableFuture<Void>> futures = new ArrayList<>();
// 分片提交（避免单批次过大）
Lists.partition(hosts, 1500).forEach(batch -> {
log.info("当前存储数据量大小 {}", batch.size());
CompletableFuture<Void> future = this.saveHostInfo(batch);
futures.add(future);
});
return CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]))
.whenComplete((result, ex) -> {
if (ex != null) {
log.error("部分批次处理失败", ex);
} else {
log.info("所有批次处理完成");
}
});
// Lists.partition(hosts, 1500).forEach(batch -> {
// log.info("当前存储数据量大小 {}", batch.size());
// CompletableFuture<Void> future = this.saveHostInfo(batch);
// futures.add(future);
// });
// return CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]))
// .whenComplete((result, ex) -> {
// if (ex != null) {
// log.error("部分批次处理失败", ex);
// } else {
// log.info("所有批次处理完成");
// }
// });
return CompletableFuture.completedFuture(null);
}
@Override