public void updateByQuery ( ESSyncConfig config , Map < String , Object > paramsTmp , Map < String , Object > esFieldData ) { if ( paramsTmp . isEmpty ( ) ) { return ; } ESMapping mapping = config . getEsMapping ( ) ; BoolQueryBuilder queryBuilder = QueryBuilders . boolQuery ( ) ; paramsTmp . forEach ( ( fieldName , value ) -> queryBuilder . must ( QueryBuilders . termsQuery ( fieldName , value ) ) ) ; // 查询sql批量更新 DataSource ds = DatasourceConfig . DATA_SOURCES . get ( config . getDataSourceKey ( ) ) ; StringBuilder sql = new StringBuilder ( "SELECT * FROM (" + mapping . getSql ( ) + ") _v WHERE " ) ; List < Object > values = new ArrayList <> ( ) ; paramsTmp . forEach ( ( fieldName , value ) -> { sql . append ( "_v." ) . append ( fieldName ) . append ( "=? AND " ) ; values . add ( value ) ; } ) ; //TODO 直接外部包裹sql会导致全表扫描性能低, 待优化拼接内部where条件 int len = sql . length ( ) ; sql . delete ( len - 4 , len ) ; Integer syncCount = ( Integer ) Util . sqlRS ( ds , sql . toString ( ) , values , rs -> { int count = 0 ; try { while ( rs . next ( ) ) { Object idVal = getIdValFromRS ( mapping , rs ) ; append4Update ( mapping , idVal , esFieldData ) ; commitBulk ( ) ; count ++ ; } } catch ( Exception e ) { throw new RuntimeException ( e ) ; } return count ; } ) ; if ( logger . isTraceEnabled ( ) ) { logger . trace ( "Update ES by query affected {} records" , syncCount ) ; } }
update by query