一、业务背景

由于需要从A数据库提取大量数据同步到B系统&＃xff0c;采用了tomikos&＃43;jta进行分布式事务管理,先将系统数据源切换到数据提供方&＃xff0c;将需要同步的数据查询出来&＃xff0c;然后再将系统数据源切换到数据接收方&＃xff0c;进行批量的插入和更新操作&＃xff0c;

二、批量插入的具体实现

1.查询需要同步的数据&＃xff1a;

&＃64;Autowired

SysPersonPOMapper sysPersonPOMapper;public voiddataDs(){//根据具体情况可以创建查询条件或者采用自定义的Mapper进行查询

SysPersonPOExample sysPersonPOExample &＃61; newSysPersonPOExample();

sysPersonPOExample.createCriteria().andIsDeleteEqualTo(false);//查询需要同步的数据

List persons &＃61;sysPersonPOMapper.selectByExample(sysPersonPOExample);

}

2.将不能进行遍历的PO实体对象转为Map&＃xff0c;此处有一个前提条件&＃xff1a;MySQL中的表字段是按全大写加下划线命名&＃xff0c;实体类PO映射字段为对应的标准驼峰命名方式&＃xff0c;进行PO到Map

的转换时&＃xff0c;会遵循这一规则。

&＃64;Autowired

SysPersonPOMapper sysPersonPOMapper;public void dataDs() throwsException{//1.查询需要同步的数据//根据具体情况可以创建查询条件或者采用自定义的Mapper进行查询

SysPersonPOExample sysPersonPOExample &＃61; newSysPersonPOExample();

sysPersonPOExample.createCriteria().andIsDeleteEqualTo(false);//查询需要同步的数据

List persons &＃61;sysPersonPOMapper.selectByExample(sysPersonPOExample);//2.将不能进行遍历的PO实体对象转为Map//用于存放转换后的对象的List

List> insertItems &＃61;Lists.newArrayList();for(SysPersonPO sysPersonPO : persons) {

Map insertItem &＃61;BeanMapUtil.convertBean2MapWithUnderscoreName(sysPersonPO);

insertItems.add(insertItem);

}

}

BeanMapUtil类&＃xff0c;PO到Map的转换方法&＃xff0c;基于类反射技术&＃xff1a;

importcom.fms.common.utils.other.StringUtil;importjava.beans.BeanInfo;importjava.beans.Introspector;importjava.beans.MethodDescriptor;importjava.beans.PropertyDescriptor;importjava.lang.reflect.Method;importjava.sql.Timestamp;importjava.util.Date;importjava.util.HashMap;importjava.util.Map;importorg.apache.commons.beanutils.BeanUtils;public classBeanMapUtil {

&＃64;SuppressWarnings({"unchecked", "rawtypes"})public static Map convertBean2MapWithUnderscoreName(Object bean) throwsException {

Map returnMap&＃61; null;try{

Class type&＃61;bean.getClass();

returnMap&＃61; newHashMap();

BeanInfo beanInfo&＃61;Introspector.getBeanInfo(type);

PropertyDescriptor[] propertyDescriptors&＃61;beanInfo

.getPropertyDescriptors();for (int i &＃61; 0; i

PropertyDescriptor descriptor&＃61;propertyDescriptors[i];

String propertyName&＃61;descriptor.getName();if (!propertyName.equalsIgnoreCase("class")) {

Method readMethod&＃61;descriptor.getReadMethod();

Object result&＃61; readMethod.invoke(bean, new Object[0]);

returnMap.put( StringUtil.underscoreName(propertyName), result);

}

}

}catch(Exception e) {//解析错误时抛出服务器异常 记录日志

throw new Exception("从bean转换为map时异常!", e);

}returnreturnMap;

}

}

StringUtil类&＃xff0c;标准驼峰命名与数据库下划线命名之间转换的方法&＃xff1a;

public classStringUtil {/*** 将驼峰式命名的字符串转换为下划线大写方式。如果转换前的驼峰式命名的字符串为空&＃xff0c;则返回空字符串。

* 例如&＃xff1a;HelloWorld->HELLO_WORLD

*

*&＃64;paramname 转换前的驼峰式命名的字符串

*&＃64;return转换后下划线大写方式命名的字符串*/

public staticString underscoreName(String name) {

StringBuilder result&＃61; newStringBuilder();if (name !&＃61; null && name.length() > 0) {//将第一个字符处理成大写

result.append(name.substring(0, 1).toUpperCase());//循环处理其余字符

for (int i &＃61; 1; i

String s&＃61; name.substring(i, i &＃43; 1);//在大写字母前添加下划线

if (s.equals(s.toUpperCase()) && !Character.isDigit(s.charAt(0))) {

result.append("_");

}//其他字符直接转成大写

result.append(s.toUpperCase());

}

}returnresult.toString();

}}

3.编写Mybatis映射文件fmsDataDsMapper.xml

insert into ${table_name}${field}values

#{item}

4.调用sqlsession相关API的insert方法插入数据&＃xff1a;

&＃64;Autowired

SysPersonPOMapper sysPersonPOMapper;public void dataDs() throwsException{//1.查询需要同步的数据//根据具体情况可以创建查询条件或者采用自定义的Mapper进行查询

SysPersonPOExample sysPersonPOExample &＃61; newSysPersonPOExample();

sysPersonPOExample.createCriteria().andIsDeleteEqualTo(false);//查询需要同步的数据

List persons &＃61;sysPersonPOMapper.selectByExample(sysPersonPOExample);//2.将不能进行遍历的PO实体对象转为Map//用于存放转换后的对象的List

List> insertItems &＃61;Lists.newArrayList();for(SysPersonPO sysPersonPO : persons) {

Map insertItem &＃61;BeanMapUtil.convertBean2MapWithUnderscoreName(sysPersonPO);

insertItems.add(insertItem);

}//3.插入数据

insertDatas(insertItems,"sys_person");

}

&＃64;Autowired

SqlSessionTemplate sqlSessionTemplate;private String dataDsNameSpace &＃61; "com.fms.common.dao.fmsDataDsMapper";private void insertDatas(List>insertItems, String tableName){if (!insertItems.isEmpty()) {

Map params &＃61;Maps.newHashMap();

//这里把数据分成每1000条执行一次&＃xff0c;可根据实际情况进行调整int count &＃61; insertItems.size() / 1000;int yu &＃61; insertItems.size() % 1000;for (int i &＃61; 0; i

List> subList &＃61;Lists.newArrayList();if (i &＃61;&＃61;count) {if(yu !&＃61; 0){

subList&＃61; insertItems.subList(i * 1000, 1000 * i &＃43;yu);

}else{continue;

}

}else{

subList&＃61; insertItems.subList(i * 1000, 1000 * (i &＃43; 1));

}

params.put("table_name", tableName);

params.put("fields", subList.get(0));

params.put("list", subList);

sqlSessionTemplate.insert(dataDsNameSpace&＃43;".insertDatas", params);

}

}

}

三 、批量更新的具体实现

通常我们根据主键更新时的语句是 update  table_name set column1 &＃61; val1 , column2 &＃61; val2 [,......] where id &＃61; ?  或者使用Mybatis提供的updateByPrimaryKey接口&＃xff0c;

当我们希望一次性更新大量数据时&＃xff0c;每条SQL只能执行一条更新&＃xff0c;MySQL支持另外一种更新方式&＃xff0c;让我们可以实现通过一条SQL语句一次性更新多条 根据主键更新的记录&＃xff1a;

ON DUPLICATE KEY UPDATE

注意&＃xff1a;这不同于在满足where条件时将某个table中的某一字段全部更新为同一各值&＃xff0c;而是每条记录修改后的值都不一样。

1.构建需要更新的数据,需要注意的是如果PO中存在不想要更新的字段&＃xff0c;你有两种处理方式&＃xff0c;一种是将该字段的值在此处设为数据库中原来的值&＃xff0c;另一种方式就是在进行Map转换时&＃xff0c;

不将此字段构建到Map中&＃xff0c;这只需要对convertBean2MapWithUnderscoreName方法做一些小的修改&＃xff0c;你可以把不需要保留的字段作为参数传给它&＃xff0c;然后在转换的时候过滤掉(主键不能省略)。

//4.构建批量更新的数据//用于存放转换后的对象的List

List> updateItems &＃61;Lists.newArrayList();for(SysPersonPO sysPersonPO : persons) {

sysPersonPO.setCode(sysPersonPO.getCode()&＃43;"updateTest");

Map updatetItem &＃61;BeanMapUtil.convertBean2MapWithUnderscoreName(sysPersonPO);

updateItems.add(updatetItem);

}

2.编写Mybatis映射文件fmsDataDsMapper.xml

INSERT INTO ${table_name}${field}VALUES

#{item}

ON DUPLICATE KEY UPDATE${field}&＃61;VALUES(${field})

3.更新数据

//5.更新数据

updateDatas(updateItems,"sys_person");private void updateDatas(List>updateItems, String tableName){if (!updateItems.isEmpty()) {

Map params &＃61;Maps.newHashMap();

//这里将数据分成每1000条执行一次&＃xff0c;可根据实际情况调整int count &＃61; updateItems.size() / 1000;int yu &＃61; updateItems.size() % 1000;for (int i &＃61; 0; i

List> subList &＃61;Lists.newArrayList();if (i &＃61;&＃61;count) {if(yu !&＃61; 0){

subList&＃61; updateItems.subList(i * 1000, 1000 * i &＃43;yu);

}else{continue;

}

}else{

subList&＃61; updateItems.subList(i * 1000, 1000 * (i &＃43; 1));

}

params.put("table_name", tableName);

params.put("fields", subList.get(0));

params.put("list", subList);

sqlSessionTemplate.insert(dataDsNameSpace&＃43;".updateDatas", params);

}

}

}