建设银行 网站无法打开,优酷网站模板下载,云主机可以做多少网站空间,性能优化工具文章目录 [toc] 1.问题2.解决办法2.1#xff09;设置合理的线程池参数2.2#xff09;设置url连接参数2.3) 优化msql的系统参数2.4#xff09;使用CountDownLatch减法计数器和数据插入的公共方法新开一个事务2.5#xff09;sql批量注入器执行成功后#xff0c;当前线程slee… 文章目录 [toc] 1.问题2.解决办法2.1设置合理的线程池参数2.2设置url连接参数2.3) 优化msql的系统参数2.4使用CountDownLatch减法计数器和数据插入的公共方法新开一个事务2.5sql批量注入器执行成功后当前线程sleep(1000)睡1s 3.业务代码套路如下4.总结
1.问题 使用上一篇文章的思路来实现数据库表全量数据同步遇到了一个奇葩的问题在本地跑代码数据条数对上了但是生产上线的时候跑数据居然条数对不上于是乎我进行了思考最大的问题有以下几个原因 1.1数据量太大线程池的参数设置不合理开的线程太多会导致数据库最大连接数不够而报一个最大连接数不够的异常从而多出来的处理不了的连接超时就会被数据库丢弃了。 1.2数据库的参数性能不一样导致发送过去插入的数据处理不过来发送给数据库插入的批次数据越多由于数据库性能不高处理不过来导致数据库端阻塞,参数设置最大连接数据缓冲区大小等等参数不合适在大批量插入场景需要优化优化数据库系统参数需要重启所以该方案不适合生产环境。 1.3线程提交到线程池执行是异步都在一个主线程中都是默认springBoot的事务隔离级别就会导致主线程执行完毕但是子线程没有执行完毕会导致所有的子线的程事务在同一个事务主线程事务里面这个主线程的事务提交的数据量就会很大主线程执行完后子线程还在执行不断的往mysql数据库发送数据导致mysql端阻塞处理不过来而丢了一些批次的数据。
2.解决办法
2.1设置合理的线程池参数 核心线程数设置为4个最大线程数设置为8个足够了设置太多的话已经超过了mysql数据库机器性能了一般机器就8核cpu所以设置4-8个线程处理数据足够了设置了太多的线程一个是消耗资源不说二一个是增大mysql的压力使用过多的线程给mysql端发送大量的数据。
package xxxxx;import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;import java.util.concurrent.Callable;
import java.util.concurrent.Future;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;
Slf4j
public class EventListenerExecutor {public static ThreadPoolTaskExecutor executor new ThreadPoolTaskExecutor();static {executor.setCorePoolSize(4);// 配置最大线程数executor.setMaxPoolSize(8);// 配置缓存队列大小executor.setQueueCapacity(10000);// 空闲线程存活时间executor.setKeepAliveSeconds(15);executor.setThreadNamePrefix(event-listener);// 线程池拒绝任务的处理策略这里采用了CallerRunsPolicy策略当线程池没有处理能力的时候该策略会直接在execute方法的调用线程中运行被拒绝的任务如果执行程序已关闭则会丢弃该任务
// executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());//自定义数据策略executor.setRejectedExecutionHandler((r, executor) - {try {executor.getQueue().put(r);} catch (InterruptedException e) {log.error(线程处理拒绝策略失败:{},e.getMessage());e.printStackTrace();}});// 等待所有任务结束后再关闭线程池executor.setWaitForTasksToCompleteOnShutdown(true);// 设置线程池中任务的等待时间如果超过这个时候还没有销毁就强制销毁以确保应用最后能够被关闭而不是被没有完成的任务阻塞executor.setAwaitTerminationSeconds(60);executor.initialize();}/*** 直接执行 不给返回值* param task*/public static void execute(Runnable task) {executor.execute(task);}/*** 执行一哈 给返回值* param task 定时处理* return* param T*/public static T FutureT submit(CallableT task){return executor.submit(task);}}2.2设置url连接参数
mysql的驱动连接的url需要加rewriteBatchedStatementstrue参数
关于rewriteBatchedStatements这个参数介绍
MySQL的JDBC连接的url中要加rewriteBatchedStatements参数并保证5.1.13以上版本的驱动才能实现高性能的批量插入。 MySQL JDBC驱动在默认情况下会无视executeBatch()语句把我们期望批量执行的一组sql语句拆散一条一条地发给MySQL数据库批量插入实际上是单条插入直接造成较低的性能。 只有把rewriteBatchedStatements参数置为true, 驱动才会帮你批量执行SQL 另外这个选项对INSERT/UPDATE/DELETE都有效
多数据源配置参看上一篇文章
spring:datasource:p6spy: truedynamic:datasource:master:username: xxxpassword: xxxxxdriver-class-name: com.mysql.cj.jdbc.Driverurl: jdbc:mysql://xxxx:3306/xxxxxx?useUnicodetruecharacterEncodingutf-8useSSLfalseserverTimezoneAsia/ShanghaizeroDateTimeBehaviorconvertToNullrewriteBatchedStatementstrueold_db:url: jdbc:mysql://xxxxxx:3306/xxxxx?characterEncodingutf8zeroDateTimeBehaviorconvertToNulluseSSLfalseallowMultiQueriestrueallowPublicKeyRetrievaltrueserverTimezoneAsia/Shanghaiusername: xxxxxxxpassword: xxxxdriver-class-name: com.mysql.cj.jdbc.Driver2.3) 优化msql的系统参数
set global max_allowed_packet1024 *1024 * 512; # 单个packet可以允许的最大值
set global max_connections 60000; # 并发连接请求量比较大建议调高此值以增加并行连接数量
set global innodb_lock_wait_timeout16 * 1024; # 事务锁超时时间默认50s超过就报错
set global bulk_insert_buffer_size512 * 1024 * 1024; # 加快insert插入效率
set global wsrep_max_ws_size1024*1024*1024*4; # 避免事务大小超过限制最大4G以上参数如果是在Navicat里面本次会话设置的使用Navicat执行批量插入sql脚本只对本次会话有效mysql重启后或会话关闭后失效永久配置需要在my.cnf配置文件中修改然后重启数据库即可永久配置可以参考网上教程。
2.4使用CountDownLatch减法计数器和数据插入的公共方法新开一个事务 使用CountDownLatch减法计数器是为了让主线程等待每个子线程都执行完在往下执行数据插入的公共方法上新开一个事务一批数据开启一个新的事务就相当于一个线程来执行数据提交隔离性好不会相互影响特别是id不唯一的情况下会在一个事务中会相互影响的小批次提高了效率。 Transactional(propagation Propagation.REQUIRES_NEW)public void dealData(CopyOnWriteArrayListCreditsRecordVO creditsRecordVOS, CountDownLatch countDownLatch) {if (CollectionUtil.isNotEmpty(creditsRecordVOS)) {EventListenerExecutor.execute(() - {doWork(creditsRecordVOS);countDownLatch.countDown();});}}2.5sql批量注入器执行成功后当前线程sleep(1000)睡1s 执行批量插入成功后睡1s是为了给数据库一个缓冲的时间让已经提交到mysql的数据执行完后在发送下一批数据过去不至于发送大量数据导致mysq那边处理不过来而交通拥挤被挤出了赛道。
Integer cr creditRecordMapper.insertBatchSomeColumn(result);
if (cr 0) {log.info(插入积分记录ok);Thread.sleep(1000);
}3.业务代码套路如下
McMembersServiceImpl类如下
package xxxxx;import cn.hutool.core.collection.CollectionUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.ibatis.cursor.Cursor;
import org.apache.ibatis.session.SqlSession;
import org.apache.ibatis.session.SqlSessionFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;Slf4j
DS(old_db)
Service
public class McMembersServiceImpl extends ServiceImplMcMembersMapper, McMembers implements McMembersService {Autowiredprivate SqlSessionFactory sqlSessionFactory;Autowiredprivate CreditRecordService creditRecordService;Overridepublic void dealData() {int pageSize 20000;AtomicInteger index new AtomicInteger(1);AtomicInteger totalIndex new AtomicInteger(0);Long total2 this.baseMapper.getTotal2();log.info(total2:{},total2);Long pageCount2 (total2 pageSize - 1) / pageSize; //推荐写法log.info(pageCount2:{}, pageCount2);CountDownLatch countDownLatch new CountDownLatch(Math.toIntExact(pageCount2));try (SqlSession sqlSession sqlSessionFactory.openSession();CursorCreditsRecordVO pageCursor sqlSession.getMapper(McMembersMapper.class).getCursorData()) {ListCreditsRecordVO creditsRecordVOS new ArrayList();for (CreditsRecordVO creditsRecordVO : pageCursor) {creditsRecordVOS.add(creditsRecordVO);totalIndex.getAndIncrement();log.info(total:{}, totalIndex.get());if (index.getAndIncrement() pageSize) {CopyOnWriteArrayListCreditsRecordVO creditsRecordVO2 new CopyOnWriteArrayList(creditsRecordVOS);log.info(creditsRecordVO2.size:{}, creditsRecordVO2.size());creditRecordService.dealData(creditsRecordVO2, countDownLatch);creditsRecordVOS.clear();log.info(清空list:{}, creditsRecordVOS.size());index.set(0);}}if (CollectionUtil.isNotEmpty(creditsRecordVOS)) {CopyOnWriteArrayListCreditsRecordVO creditsRecordVO3 new CopyOnWriteArrayList(creditsRecordVOS);log.info(creditsRecordVO3.size:{}, creditsRecordVO3.size());creditRecordService.dealData(creditsRecordVO3, countDownLatch);creditsRecordVOS.clear();log.info(清空list2:{}, creditsRecordVOS.size());index.set(0);}countDownLatch.await();} catch (Exception e) {log.error(游标查询异常:{}, e.getMessage());}log.info(total:{}, totalIndex.get());}Overridepublic Long getTotal2() {return baseMapper.getTotal2();}}CreditRecordServiceImpl类如下
package xxxxxxx;import cn.hutool.core.collection.CollectionUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.nacos.common.utils.MD5Utils;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.StringUtils;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.BeanUtils;
import org.springframework.jdbc.datasource.DataSourceTransactionManager;
import org.springframework.stereotype.Service;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;import java.math.BigDecimal;
import java.time.Instant;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.time.format.DateTimeFormatter;
import java.time.temporal.TemporalAdjusters;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;Slf4j
Service
RequiredArgsConstructor
public class CreditRecordServiceImpl extends ServiceImplCreditRecordMapper, CreditRecord implements CreditRecordService {private final TransactionDefinition transactionDefinition;private final DataSourceTransactionManager transactionManager;private final CreditRecordMapper creditRecordMapper;private final MemberMapper memberMapper;OverrideTransactional(propagation Propagation.REQUIRES_NEW)public void dealData(CopyOnWriteArrayListCreditsRecordVO creditsRecordVOS, CountDownLatch countDownLatch) {if (CollectionUtil.isNotEmpty(creditsRecordVOS)) {EventListenerExecutor.execute(() - {doWork(creditsRecordVOS);countDownLatch.countDown();});}}private void doWork(CopyOnWriteArrayListCreditsRecordVO creditLogVOS) {try {ListCreditRecord result new ArrayList();for (CreditsRecordVO creditLogVO : creditLogVOS) {CreditRecord creditRecord new CreditRecord();creditRecord.setId(Long.valueOf(creditLogVO.getId()));creditRecord.setContent(creditLogVO.getRemark() ! null ? creditLogVO.getRemark() : );if (Objects.nonNull(creditLogVO.getNum())) {creditRecord.setNum(BigDecimal.valueOf(creditLogVO.getNum()));if (creditLogVO.getNum() 0) {creditRecord.setFlowType(1);} else if (creditLogVO.getNum() 0) {creditRecord.setFlowType(1);}} else {continue;}creditRecord.setCreditType(creditLogVO.getCredittype());creditRecord.setStatus(1);if (Objects.nonNull(creditLogVO.getUid())) {creditRecord.setMemberId(Long.valueOf(creditLogVO.getUid()));} else {continue;}Integer createtime creditLogVO.getCreatetime();if (Objects.nonNull(createtime)) {LocalDateTime localDateTime LocalDateTime.ofInstant(Instant.ofEpochSecond(createtime), ZoneOffset.ofHours(8));creditRecord.setCreateTime(localDateTime);}creditRecord.setBalanceSnapshot(BigDecimal.valueOf(-1));creditRecord.setDeleted(0);creditRecord.setIsExp(0);creditRecord.setFromId(0L);creditRecord.setOpType(0);creditRecord.setOrderNo();creditRecord.setUpdateTime(LocalDateTime.now());result.add(creditRecord);}log.info(插入数据条数:{}, result.size());Integer cr creditRecordMapper.insertBatchSomeColumn(result);if (cr 0) {log.info(插入积分记录ok);Thread.sleep(1000);}} catch (Exception e) {e.printStackTrace();log.error(插入积分记录异常:{}, ExceptionUtils.getMessage(e));}}}4.总结 这个是一个经过实战的套路最近重构了一个会员模块涉及到会员表数据4百多万、粉丝表数据3百多万、关联映射表数据4百多万还有积分日志记录数据1千多万需要清洗同步到新的表中前三个表是有关联的白表总的数据量有1千多万后面的1千多万总的2千多万数据同步时间只需要20分钟就可以搞定的性能和效率还是杠杠的执行时间如下图所示 如果这个问题没有处理好生产少的的会员数据就需要重新去反查使用sql的join查出少了的数据重新补上去由于用户数据同步少了老用户登录系统就会变成一个新用户就会带来一系列的问题之前用老账号下的单子登录后是一个新用户关联不到订单了所以需要去修复各种业务数据数据补完后需要把新用户相关删除然后清除redis的用户相关的数据什么token/access_token/用户信息了数据量大那生产redis的用户相关的缓存数据也大的这些由于数据同步的问题带来的后续这种蛋疼的问题也是让人头疼所以写了一篇避坑提效的文章希望对你有帮助请一键三连么么哒如何批量删除redis中的用户信息请看下一篇文章
