Нужна конфигурация Java Spring Batch Mongo DB для чтения данных из БД с помощью Spring Boot

Question

Это моя Spring Mongo конфигурация для обычного Spring проекта.Я хочу знать, как отредактировать его для Spring-Batch.

import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.security.KeyManagementException;
import java.security.KeyStore;
import java.security.KeyStoreException;
import java.security.NoSuchAlgorithmException;
import java.security.UnrecoverableKeyException;
import java.security.cert.CertificateException;
import java.util.ArrayList;
import java.util.List;

import javax.net.ssl.KeyManagerFactory;
import javax.net.ssl.SSLContext;
import javax.net.ssl.SSLSocketFactory;
import javax.net.ssl.TrustManagerFactory;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Profile;
import org.springframework.core.io.ResourceLoader;
import org.springframework.data.mongodb.MongoDbFactory;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.SimpleMongoDbFactory;
import org.springframework.data.mongodb.core.convert.DbRefResolver;
import org.springframework.data.mongodb.core.convert.DefaultDbRefResolver;
import org.springframework.data.mongodb.core.convert.DefaultMongoTypeMapper;
import org.springframework.data.mongodb.core.convert.MappingMongoConverter;
import org.springframework.data.mongodb.core.mapping.MongoMappingContext;

import com.mongodb.MongoClient;
import com.mongodb.MongoClientOptions;
import com.mongodb.MongoCredential;
import com.mongodb.ServerAddress;



@Configuration
public class MongoConfiguration
{

    @Autowired
    ResourceLoader       resourceLoader;

    @Value("${mongodb.keyStoreFilePath}")
    private String       keyStoreFilePath;

    @Value("${mongodb.keyStoreType}")
    private String       keyStoreType;

    @Value("${mongodb.keyStorePassword}")
    private String       keyStorePassword;

    @Value("${mongodb.databasePort}")
    private String       databasePort;

    @Value("${mongodb.databaseName}")
    private String       databaseName;

    @Value("${mongodb.dbUserName}")
    private String       dbUserName;

    @Value("${mongodb.dbPassword}")
    private String       dbPassword;

    @Value("#{'${mongodb.seeds}'.split(',')}")
    private List<String> seeds;


    @Bean
    @Profile("default")
    public MongoTemplate mongoTemplate() throws UnrecoverableKeyException, KeyManagementException, KeyStoreException,
            NoSuchAlgorithmException, CertificateException, IOException
    {
        DbRefResolver dbRefResolver = new DefaultDbRefResolver(mongoDbFactory());
        MappingMongoConverter converter = new MappingMongoConverter(dbRefResolver, new MongoMappingContext());
        converter.setTypeMapper(new DefaultMongoTypeMapper(null));
        return new MongoTemplate(mongoDbFactory(), converter);
    }



    @Bean
    @Profile("default")
    public MongoDbFactory mongoDbFactory() throws UnrecoverableKeyException, KeyManagementException, KeyStoreException,
            NoSuchAlgorithmException, CertificateException, IOException
    {
        return new SimpleMongoDbFactory(mongoClient(), databaseName);

    }


    @Bean
    @Profile("default")
    public MongoClient mongoClient() throws IOException, KeyStoreException, NoSuchAlgorithmException,
            CertificateException, UnrecoverableKeyException, KeyManagementException
    {

        try (InputStream readStream = new FileInputStream(keyStoreFilePath))
        {
            KeyStore ksClient = KeyStore.getInstance(keyStoreType);
            ksClient.load(readStream, keyStorePassword.toCharArray());
            KeyManagerFactory kmf = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
            kmf.init(ksClient, keyStorePassword.toCharArray());

            KeyStore ksCACert = KeyStore.getInstance(KeyStore.getDefaultType());
            ksCACert.load(new FileInputStream(keyStoreFilePath), keyStorePassword.toCharArray());

            TrustManagerFactory tmf = TrustManagerFactory.getInstance("X509");
            tmf.init(ksCACert);

            SSLContext context = SSLContext.getInstance("TLS"); // We now
                                                                // provide
                                                                // our alternate
                                                                // KeyManager
            context.init(kmf.getKeyManagers(), tmf.getTrustManagers(), null);
            SSLSocketFactory sslSocketFactory = context.getSocketFactory();

            MongoClientOptions mongoClientOptions =
                    MongoClientOptions.builder().sslEnabled(true).socketFactory(sslSocketFactory).build();

            List<ServerAddress> seedList = new ArrayList<>();
            for (String ip : seeds)
            {
                seedList.add(new ServerAddress(ip, Integer.parseInt(databasePort)));
            }
            return new MongoClient(seedList,
                    MongoCredential.createCredential(dbUserName, databaseName, dbPassword.toCharArray()),
                    mongoClientOptions);
        }

    }

}

Теперь я хочу добавить вышеупомянутую Mongo конфигурацию в мою Spring-Batch конфигурацию.Как я должен это делать?(В основном это миграция с Oracle на MongoDB). Ниже приведен код моей предыдущей конфигурации с использованием Oracle DB.Нужно изменить это для MongoDB.Я имею в виду, что я запутался, что и все, что я должен добавить и удалить, как - JobRepository как изменить и все.

    import java.sql.SQLException;

    import javax.sql.DataSource;

    import org.apache.commons.httpclient.HostConfiguration;
    import org.apache.commons.httpclient.HttpClient;
    import org.apache.commons.httpclient.MultiThreadedHttpConnectionManager;
    import org.apache.commons.httpclient.params.HttpConnectionManagerParams;
    import org.springframework.batch.core.configuration.annotation.DefaultBatchConfigurer;
    import org.springframework.batch.core.repository.JobRepository;
    import org.springframework.batch.core.repository.support.JobRepositoryFactoryBean;
    import org.springframework.beans.factory.annotation.Qualifier;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.boot.autoconfigure.jdbc.DataSourceBuilder;
    import org.springframework.boot.context.properties.ConfigurationProperties;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.context.annotation.Primary;

    @Configuration
    public class ApplicationConfiguration extends DefaultBatchConfigurer {


        private DataSource dataSource;
        public HttpClient httpClient;

        @Value("#{'${connectionTimeOut}'}")
        int connectionTimeOut;
        @Value("#{'${maxTotalConnections}'}")
        int maxTotalConnections;
        @Value("#{'${defaultMaxConnections}'}")
        int defaultMaxConnections;
        @Value("#{'${hostInstance}'}")
        String hostInstance;
        @Value("#{'${wagProxy}'}")
        String wagProxy;
        @Value("#{'${wagProxyPort}'}")
        int wagProxyPort;
        @Value("#{'${connectionManagerTimeOut}'}")
        int connectionManagerTimeOut;


            @Override
            protected JobRepository createJobRepository() throws Exception {
                JobRepositoryFactoryBean factory = new JobRepositoryFactoryBean();
                factory.setDataSource(dataSource);
                factory.setTransactionManager(this.getTransactionManager());
                factory.afterPropertiesSet();
            return factory.getObject();
        }
            @Bean
            @Primary
            @ConfigurationProperties(prefix = "batch.datasource")
            @Qualifier("profileDataSource")
            public DataSource batchDataSource() throws SQLException {
                dataSource = DataSourceBuilder.create().build();
                super.setDataSource(dataSource);
                return dataSource;
            }

            @Bean
            @Primary
            @Qualifier("httpClient")
            public HttpClient httpClientConnection() {
                HttpConnectionManagerParams conManagerparams = null;
                MultiThreadedHttpConnectionManager connectionManager = null;
                conManagerparams = new HttpConnectionManagerParams();
                conManagerparams.setTcpNoDelay(true);
                conManagerparams.setStaleCheckingEnabled(true);
                conManagerparams.setConnectionTimeout(connectionTimeOut);
                conManagerparams.setDefaultMaxConnectionsPerHost(defaultMaxConnections);
                conManagerparams.setMaxTotalConnections(maxTotalConnections);
                connectionManager = new MultiThreadedHttpConnectionManager();
                connectionManager.setParams(conManagerparams);
                httpClient = new HttpClient(connectionManager);
                HostConfiguration hostCfg = new HostConfiguration();
                hostCfg.setHost(hostInstance);
                httpClient.setHostConfiguration(hostCfg);
                httpClient.getHostConfiguration().setProxy(wagProxy,wagProxyPort);
                httpClient.getParams().setConnectionManagerTimeout(connectionManagerTimeOut);
                return httpClient;
            }

    }

Answer 1

В основном это миграция с Oracle на MongoDB

Для чтения из Oracle вы можете использовать, например, JdbcCursorItemReader (или другие устройства чтения БД, предоставляемые Spring Batch):

@Bean
public JdbcCursorItemReader<Foo> itemReader() {
    return new JdbcCursorItemReaderBuilder<Foo>()
        .dataSource(batchDataSource()) // change accordingly if you use another data source
        .name("fooReader")
        .sql("SELECT * FROM FOO")
        .beanRowMapper(Foo.class)
        .build();
}

Для записи в MongoDB вы уже определили mongoTemplate, чтобы вы могли использовать его для создания MongoItemWriter:

@Bean
public MongoItemWriter<String> itemWriter() {
    return new MongoItemWriterBuilder<String>()
        .template(mongoTemplate())
        .build();
}

Затем вы определяете шаг, ориентированный на чанк, длячитать из Oracle и писать в MongoDB.

Надеюсь, это поможет.

РЕДАКТИРОВАТЬ: Добавление образца чтения монго:

@Bean
public MongoItemReader<Foo> mongoItemReader() { 
    Query query; // specify your mongo query
    return new MongoItemReaderBuilder<Foo>()
        .template(mongoTemplate())
        .targetType(Foo.class)
        .query(query)
        .build();
}