在现代软件开发中，性能优化是一个永恒的话题，尤其是在处理大规模数据时，如何提升数据库操作的效率成为了一个关键问题。在数据库操作中，批量插入空间矢量数据是一个常见的需求，尤其是在地理信息系统（GIS）和空间数据分析领域。调试日志是软件开发中不可或缺的工具，它帮助开发者追踪程序的运行状态，定位问题和异常。然而，日志记录本身是一个资源密集型的操作，尤其是在生产环境中，过多的日志记录可能会对性能产生负面影响。对于空间矢量数据的批量插入操作，这种影响尤为明显，因为这类操作通常涉及大量的I/O操作和数据库交互。

尽管调试日志对于开发和问题排查至关重要，但在生产环境中，它们可能会成为性能瓶颈。每次日志记录都会涉及到I/O操作，这会占用CPU时间和磁盘I/O资源。在批量插入空间矢量数据时，如果日志记录过于频繁，可能会导致以下问题：

降低I/O效率：日志记录会占用磁盘I/O资源，这可能会与数据库操作竞争资源，导致整体性能下降。
增加延迟：日志记录可能会引入额外的延迟，尤其是在高并发情况下，这会直接影响到批量插入操作的响应时间。
资源竞争：日志记录和数据库操作可能会竞争有限的系统资源，如CPU和内存，这可能会导致性能瓶颈。

调试日志是一把双刃剑，它在帮助开发者解决问题的同时，也可能对生产环境的性能产生影响。在处理空间矢量数据的批量插入时，合理控制和优化日志记录是提升性能的关键。本文通过在MybatisPlus中调整插入SQL的输出对比前后的耗时与内存的占用，最大限度地减少对性能的负面影响。本文将深入探讨这些策略的具体实现和最佳实践，以期为Java开发者提供实用的指导和建议。

一、一些缘由

其实在日常工作当中，空间矢量数据的数据量都是非常大。不仅是范围大，属性数据也尤其多，不仅属性列多，而且数据行数也可能非常多。那么我们在使用ORM框架在操作这些数据的时候，在进行空间数据入库的时候尤其需要注意性能的影响。有一些程序需要追求高性能，尤其是一些需要快速计算的场景，用户需要尽快的将数据入库，好开展后续的业务。

之前有一个朋友给发了私信，说他们在处理线上的生产数据时，数据的规模大约是几十W的规模。在使用GeoTools读取Shapefile后，然后调用Mybatis-Plus来进行数据入库。它的整体性能不高，耗时比较久，然后就找到博主聊了一下。原始聊天截图就不放出来了。分享其中遇到的一些问题：

1、这位朋友在进行批量数据入库的时候，使用循环来进行调用，没有使用批量操作。

2、系统的日志级别开的比较低，为了方便监控程序，系统的日志级别在生产环境上也是Debug。

3、服务器在空间数据库入库时，内存占用较高。

1、性能分析

在了解了一些程序的执行细节之后，我也做了一个对照实验。实验的主要目的是对比应用程序中调试日志的输出对性能影响，主要方法就是在程序执行时打开和关闭系统日志，通过观察打开前后的应用程序执行消耗时间和使用Java VisualVM监控的CPU和内存消耗情况来对比。

二、插入方式调整

为了首先将应用程序的插入调整到一个比较好的执行状态，我们先把原来的循环插入的方式进行了修改，改成批量插入的形式。因此这里有必要对批量插入的具体实现进行一个简单的介绍。

1、批量插入的实现

在我们的代码中，使用的ORM框架是Mybatis-Plus，熟悉这个框架的小伙伴们一定知道。在MP中除了有单个插入的方法，还提供了一个批量插入的实现。因此，如果您是使用了MP这种的增强框架，那么改造起来还是比较快的，否则就需要大家自己去实现批量插入的方法。在MP中需要调用service提供的saveBatch（List,Size）即可。在在我的示例代码中，实现批量插入的关键代码如下所示：

Long s3 = System.currentTimeMillis();
if(dataList.size() >0) {
	placeService.saveBatch(dataList, 600);
}
Long e3 = System.currentTimeMillis();
System.out.println("空间入库耗时:："+ (e3 - s3) + "毫秒");

2、MP的批量插入实现

上一节中对Mp的批量插入的方法进行了调用，这里我们依然对saveBatch方法进行简单的介绍，好让大家对saveBatch有一个直观的印象。我们可以打开ServiceImpl的实现类中的以下代码：

/**
 * 批量插入
 * @param entityList ignore
 * @param batchSize  ignore
 * @return ignore
 */
 @Transactional(rollbackFor = Exception.class)
 @Override
 public boolean saveBatch(Collection<T> entityList, int batchSize) {
     String sqlStatement = getSqlStatement(SqlMethod.INSERT_ONE);
     return executeBatch(entityList, batchSize, (sqlSession, entity) -> sqlSession.insert(sqlStatement, entity));
 }

这里的方法表示首先获取sql的Statement对象，然后调用批量执行的方法。被调用的方法如下：

/**
 * 执行批量操作
 *
 * @param entityClass 实体类
 * @param log         日志对象
 * @param list        数据集合
 * @param batchSize   批次大小
 * @param consumer    consumer
 * @param <E>         T
 * @return 操作结果
 * @since 3.4.0
 */
public static <E> boolean executeBatch(Class<?> entityClass, Log log, Collection<E> list, int batchSize, BiConsumer<SqlSession, E> consumer) {
        Assert.isFalse(batchSize < 1, "batchSize must not be less than one");
        return !CollectionUtils.isEmpty(list) && executeBatch(entityClass, log, sqlSession -> {
            int size = list.size();
            int idxLimit = Math.min(batchSize, size);
            int i = 1;
            for (E element : list) {
                consumer.accept(sqlSession, element);
                if (i == idxLimit) {
                    sqlSession.flushStatements();
                    idxLimit = Math.min(idxLimit + batchSize, size);
                }
                i++;
            }
     });
}

来看一下insert方法的处理逻辑，最终的执行update的方法如下：

@Override
public int doUpdate(MappedStatement ms, Object parameterObject) throws SQLException {
    final Configuration configuration = ms.getConfiguration();
    final StatementHandler handler = configuration.newStatementHandler(this, ms, parameterObject, RowBounds.DEFAULT, null, null);
    final BoundSql boundSql = handler.getBoundSql();
    final String sql = boundSql.getSql();
    final Statement stmt;
    if (sql.equals(currentSql) && ms.equals(currentStatement)) {
      int last = statementList.size() - 1;
      stmt = statementList.get(last);
      applyTransactionTimeout(stmt);
      handler.parameterize(stmt);// fix Issues 322
      BatchResult batchResult = batchResultList.get(last);
      batchResult.addParameterObject(parameterObject);
    } else {
      Connection connection = getConnection(ms.getStatementLog());
      stmt = handler.prepare(connection, transaction.getTimeout());
      handler.parameterize(stmt);    // fix Issues 322
      currentSql = sql;
      currentStatement = ms;
      statementList.add(stmt);
      batchResultList.add(new BatchResult(ms, sql, parameterObject));
    }
    handler.batch(stmt);
    return BATCH_UPDATE_RETURN_VALUE;
 }

当然大家在使用这个类的时候还是非常方便的，只要调用相应的方法即可实现分批导入。

3、日志的配置

在实例的应用开发过程中，日志的输出与管理，我们使用Logback组件。在最开始的时候，在对比实验中，首先我们采用默认的方式，即对应的ORM处理组件中的日志级别使用默认方法。具体如何在Logback中进行日志的设置，请大家结合互联网相关资料进行查询，这些都是比较成熟的。

三、默认处理方式

对比实验的第一种实现方法就是采用默认的方法，即使用默认的日志级别。但是在最开始时，我们还是给出测试的代码的全部。如果您也感兴趣，可以替换相应的文件来进行验证这个过程。测试结果可能随着数据量的不同，数据属性字段的不同而有所不同。

1、基础程序代码

为了还原网友提出的问题，也能尽快的找到原因。我们这里就以之前的全球主要城市为例，重点讲解如何进行数据的处理和融合，以及最终如何进入到数据库中。实例代码如下：

@Test
/**
 * 
 * @throws Exception
 */
public void shp2PostGIS() throws Exception {
	Long startTime = System.currentTimeMillis();
	File file = new File(SHP_FILE);
	if (!file.exists()) {
		System.out.println("文件不存在");
	}
	ShapefileDataStore store = new ShapefileDataStore(file.toURI().toURL());
	store.setCharset(Charset.defaultCharset());// 设置中文字符编码
	// 获取特征类型
	SimpleFeatureType featureType = store.getSchema(store.getTypeNames()[0]);
	CoordinateReferenceSystem crs = featureType.getGeometryDescriptor().getCoordinateReferenceSystem();
	Integer epsgCode = CRS.lookupEpsgCode(crs, true);
	List<HashMap<String, Object>> mapList = new ArrayList<HashMap<String,Object>>();
	ModelMapper modelMapper = new ModelMapper();
	//设置忽略字段
	PropertyMap<HashMap<String,Object>, Ne10mPopulatedPlaces> propertyMap = new PropertyMap<HashMap<String,Object>, Ne10mPopulatedPlaces>() {
		 protected void configure() {
		     skip(destination.getPkId());
		}
	};
	modelMapper.addMappings(propertyMap);
	//忽略大小写
	modelMapper.getConfiguration().setMatchingStrategy(MatchingStrategies.STRICT);
	// 设置命名约定，将下划线转换为驼峰
	modelMapper.getConfiguration().setSourceNameTokenizer(NameTokenizers.UNDERSCORE)
	.setDestinationNameTokenizer(NameTokenizers.CAMEL_CASE);
	//设置忽略模式
	modelMapper.getConfiguration().setSkipNullEnabled(true);
	Long s1 = System.currentTimeMillis();
	List<Ne10mPopulatedPlaces> dataList = new ArrayList<Ne10mPopulatedPlaces>();
	SimpleFeatureSource featureSource = store.getFeatureSource();
	// 执行查询
	SimpleFeatureCollection simpleFeatureCollection = featureSource.getFeatures();
	SimpleFeatureIterator itertor = simpleFeatureCollection.features();
	// 遍历featurecollection
	while (itertor.hasNext()) {
		HashMap<String, Object> map = new HashMap<String, Object>();
		SimpleFeature feature = itertor.next();
		Collection<Property> p = feature.getProperties();
		Iterator<Property> it = p.iterator();
		// 遍历feature的properties
		while (it.hasNext()) {
			Property pro = it.next();
			if (null != pro && null != pro.getValue()) {
				String field = pro.getName().toString();
				String value = pro.getValue().toString();
				map.put(field, value);
			}
		}
		// 获取空间字段
        org.locationtech.jts.geom.Geometry geometry = (org.locationtech.jts.geom.Geometry) feature.getDefaultGeometry();
        // 创建WKTWriter对象
        WKTWriter wktWriter = new WKTWriter();
        // 将Geometry对象转换为WKT格式的字符串
        String wkt = wktWriter.write(geometry);
        String geom = "SRID=" + epsgCode +";" + wkt;//拼接srid，实现动态写入
        map.put("geom", geom);
        mapList.add(map);
	}
	Long e1 = System.currentTimeMillis();
	System.out.println("解析shp："+ (e1 - s1) + "毫秒");
	Long s2 = System.currentTimeMillis();
	for(HashMap<String, Object> map : mapList) {
		Ne10mPopulatedPlaces places = modelMapper.map(map, Ne10mPopulatedPlaces.class);
		dataList.add(places);
	}
	Long e2 = System.currentTimeMillis();
	System.out.println("转化shp："+ (e2 - s2) + "毫秒");
	store.dispose();
	System.out.println(dataList.size());
	Long endTime = System.currentTimeMillis();
	Long time = endTime - startTime;
	System.out.println("程序运行耗时："+ time + "毫秒");
	Long s3 = System.currentTimeMillis();
	if(dataList.size() >0) {
		placeService.saveBatch(dataList, 600);
	}
	Long e3 = System.currentTimeMillis();
	System.out.println("空间入库耗时:："+ (e3 - s3) + "毫秒");
}

在不关闭执行SQL日志的情况，我们来看一下它的相关性能指标。

2、执行情况

在默认情况下，这段程序在执行过程中会输出大量的调试日志，如下图所示：

在我们的控制台中有许多的插入日志，同时可以看到，整个空间数据入库的时间为29512毫秒，即将近30秒。除了时间的消耗，我们再来看一内存方面的消耗。

可以很直观的看到，在进行大量的日志输出时，内存的使用量还是比较大，同时根据不同的批次呈现一个比较有规律的上升和下降，而最大的内存使用接近1000MB左右。接下来，我们再来看一下关闭日志输出后的效果。

四、提升调试日志等级

为了实现在运行时将这个插入SQL的日式调试等级，我们在Logback中进行相应的配置。以此来验证在提升SQL调试日志的等级后，这个批量插入的方法是不是有一个性能的提升。

1、在logback中进行设置

在系统中，我们采用logback来进行日志的配置，因此我们首先需要在logback中进行相应的设置。将日志的级别从debug提升到error，只有在发生错误的时候才进行输出。设置的关键代码如下所示：

<!--  Ne10mPopulatedPlacesMapper 关闭调试日志 add by 夜郎king in 2024-11-26  -->
<logger name="com.yelang.project.extend.earthquake.mapper.Ne10mPopulatedPlacesMapper" level="error"/>

请注意，这里的com.yelang.project.extend.earthquake.mapper.Ne10mPopulatedPlacesMapper标识我们需要关闭的ORM类的全名。我们将他的日志级别提升到了error。

2、提升后的效果

在将输出日志关闭之后，在控制台中首先就没有了sql的调试日志，说明配置成功。

可以看到控制台很干净，调试的SQL日志已经被清理掉。同时注意耗时情况，变成了7046，也就是7秒的时间就完成了处理。在来后台看一下是不是真的处理成功。在数据库进行相应的数据查询。

可以看到，数据的总条数也是7342条。因此可以判断，关闭sql调试日志后，对时间的消耗降低了很多，从30秒优化到了7秒，大概提升76%；再来看一下内存的占用情况。

相对于默认的处理情况而言，提升了日志等级的处理方式，其内存占用更加平稳，波动小。同时最大的内存占用在700MB左右，更多是500MB以下。从侧面也说明了优化的效果。

五、总结

以上就是本文的主要内容，本文通过在MybatisPlus中调整插入SQL的输出对比前后的耗时与内存的占用，最大限度地减少对性能的负面影响。文章通过对照实验，对比了开启调试日志和关闭调试日志后的数据插入性能，从对比实验结果可以看到。关闭调试日志后，我们的应用程序耗时更短，同时内存的占用也更低。如果在生产环境中进行使用，尤其是新手同志，为了观察参数就留下了很多调试信息，这样反而加大了系统的负担。所以要请大家一定综合理性的评估，关闭不必要的调试日志，让应用程序的性能最大。行文仓库，定有许多不足之处，如有不足，在此恳请各位专家在评论区批评指出，不胜感激。

点赞
收藏
关注作者

0/1000

抱歉，系统识别当前为高风险访问，暂不支持该操作

全部回复

上滑加载中

设置昵称

在此一键设置昵称，即可参与社区互动！

*长度不超过10个汉字或20个英文字符，设置后3个月内不可修改。

确认取消

加入云驻计划，成为创作者

华为云周边好礼
免费体验产品
特殊身份标识
线下官方门票
内部专家零距离
与10000+优质创作者共同成长

立即加入

生产慎用之调试日志对空间矢量数据批量插入的性能影响-以MybatisPlus为例

前言

一、一些缘由

1、性能分析

二、插入方式调整

1、批量插入的实现

2、MP的批量插入实现

3、日志的配置

三、默认处理方式

1、基础程序代码

2、执行情况

四、提升调试日志等级

1、在logback中进行设置

2、提升后的效果

五、总结

全部回复

设置昵称

关于作者

目录

加入云驻计划，成为创作者

生产慎用之调试日志对空间矢量数据批量插入的性能影响-以MybatisPlus为例

前言

一、一些缘由

1、性能分析

二、插入方式调整

1、批量插入的实现

2、MP的批量插入实现

3、日志的配置

三、默认处理方式

1、基础程序代码

2、执行情况

四、提升调试日志等级

1、在logback中进行设置

2、提升后的效果

五、总结

全部回复

设置昵称

关于作者

目录

加入云驻计划，成为创作者

推荐阅读

相关产品