首页
学习
活动
专区
圈层
工具
发布
首页
学习
活动
专区
圈层
工具
MCP广场
社区首页 >问答首页 >带有https请求多线程弹簧的jpa

带有https请求多线程弹簧的jpa
EN

Stack Overflow用户
提问于 2019-10-27 18:31:42
回答 3查看 377关注 0票数 0

我正在处理spring JPAHTTP post请求,逐行获取数据,然后将数据放到HTTP中到API中,它对我来说很好,但是这里我处理大量的数据,所以我不得不使用多线程,但是使用java和HTTP post来实现新的线程,如何实现与10个线程一起工作,并且每个线程每次并行读取1k?

我读了一些关于多线程的内容,每个线程每次读取1k行,我的数据库中有大约1000万条记录。

AccessingDataJpaApplication类:

代码语言:javascript
运行
复制
@SpringBootApplication
public class AccessingDataJpaApplication implements CommandLineRunner {

    private static final Logger logger = LoggerFactory.getLogger(AccessingDataJpaApplication.class);
    @Autowired

    private Bulk_repositoryRepository bulk_repositoryRepository;


    public static void main(String[] args) {
        SpringApplication.run(AccessingDataJpaApplication.class);
    }
    Date currentDate = new Date();

    @Override
    public void run(String... args) throws Exception {
        RestTemplate restTemplate = new RestTemplate();
        HttpHeaders headers = new HttpHeaders();
        headers.setAccept(Arrays.asList(MediaType.APPLICATION_JSON));
        headers.setBasicAuth("user", "pass");

        while(true) {
            Date currentDate = new Date();
            logger.info("Just Started"); 
            for (Bulk_repository churnss : bulk_repositoryRepository.findAllByStatusAndCampTypeAndCampStartDateLessThanEqualAndCampEndDateGreaterThanEqual(0,2,currentDate,currentDate)) {
                System.out.print(churnss);
                logger.info(churnss.toString());
                AddOfferRequest AddOffer = new AddOfferRequest("113", churnss.getMsisdn(),churnss.getParam1());

                logger.info(AddOffer.toString());
                HttpEntity<AddOfferRequest> entity = new HttpEntity<AddOfferRequest>(AddOffer,headers);

                ResponseEntity<String> responseEntity = restTemplate.exchange(
                        "api link", HttpMethod.POST, entity, String.class);

                if(responseEntity.getStatusCode() == HttpStatus.OK){
                    String response = responseEntity.getBody();
                    churnss.setStatus(1);
                    churnss.setProcessDate(new Date());
                    churnss.setFulfilment_status(response);
                    logger.info(churnss.toString() + ", Response: " + response);
                    bulk_repositoryRepository.save(churnss);
                }else {
                    logger.warn("Record Id: " + churnss.getId() + ", Http Failed Response: " + responseEntity.getStatusCode());
                }
            }
            Thread.sleep(1000);
        }
    }

}

Bulk_repository类:

代码语言:javascript
运行
复制
@Entity
@Table(name = "BULK_REPOSITORY")
public class Bulk_repository {

   @Id
   @GeneratedValue(strategy=GenerationType.AUTO)
   @Column(name = "id")
   private long id;

   @Column(name = "msisdn")
   private String msisdn;

   @Column(name = "camp_start_date")   
   private Date campStartDate;

   @Column(name = "camp_end_date")
   private Date campEndDate;

   @Column(name = "camp_type")
   private int campType;

   @Column(name = "camp_cd")
   private String camp_cd;

   @Column(name = "status")
   private int status;

   @Column(name = "process_date")
   private Date processDate;

   @Column(name = "entry_date")
   private Date entryDate;

   @Column(name = "entry_user")
   private String entry_user;

   @Column(name = "param1")
   private String param1;

   @Column(name = "param2")
   private String param2;

   @Column(name = "param3")
   private String param3;

   @Column(name = "param4")
   private String param4;

   @Column(name = "param5")
   private String param5;

   @Column(name = "error_desc")
   private String error_desc;

   @Column(name = "fulfilment_status")
   private int fulfilment_status;
   ##then getter and setters and tostring

Bulk_repositoryRepository类:

代码语言:javascript
运行
复制
public interface Bulk_repositoryRepository extends CrudRepository<Bulk_repository, Long> {

      Date today = new Date();
      List<Bulk_repository>findAllByStatusAndCampTypeAndCampStartDateLessThanEqualAndCampEndDateGreaterThanEqual(int status, int campType,Date today0, Date today1);
      Bulk_repository findById(long id);
}

AddOfferRequest类:

代码语言:javascript
运行
复制
public class AddOfferRequest {

    private String ChannelID="113";
    private String MSISDN;
    private String ServiceID;

    public AddOfferRequest() {
    }
    public AddOfferRequest(String channelID,String mSISDN,String serviceID ) {
        this.MSISDN = mSISDN;
        this.ServiceID = serviceID;

    }
    ## then getter and setter and tostring

我创建了AsyncConfiguration类:

代码语言:javascript
运行
复制
package com.example.accessingdatajpa;


import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import java.util.concurrent.Executor;


@Configuration
@EnableAsync
public class AsyncConfiguration {
    private static final Logger LOGGER = LoggerFactory.getLogger(AsyncConfiguration.class);
    @Bean (name = "taskExecutor")
    public Executor taskExecutor() {
        LOGGER.debug("Creating Async Task Executor");
        final ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(2);
        executor.setMaxPoolSize(2);
        executor.setQueueCapacity(1000);
        executor.setThreadNamePrefix("CarThread-");
        executor.initialize();
        return executor;
    }
}

但到目前为止,我还不知道如何才能将findby和http post与多线程结合起来

EN

回答 3

Stack Overflow用户

回答已采纳

发布于 2019-10-29 08:39:53

重写你的代码。而不是List<Bulk_repository>,而是返回一个Stream<Bulk_repository>。这将懒洋洋地从数据库加载记录,而不是尝试一次完成所有的工作。

然后使用TaskExecutor执行每个线程的不同请求,只需给它一个任务,当有一个空闲线程时,它将被执行。

代码语言:javascript
运行
复制
@SpringBootApplication
public class AccessingDataJpaApplication implements CommandLineRunner {

    private static final Logger logger = LoggerFactory.getLogger(AccessingDataJpaApplication.class);

    @Autowired
    private Bulk_repositoryRepository bulk_repositoryRepository;

    @Autowired
    private AsyncTaskExecutor executor;

    @Autowired
    private RestTemplate rest;

    public static void main(String[] args) {
        SpringApplication.run(AccessingDataJpaApplication.class);
    }

    @Override
    public void run(String... args) throws Exception {
        Date currentDate = new Date();

        Stream< Bulk_repository> results = Bulk_repository churnss : bulk_repositoryRepository.findAllByStatusAndCampTypeAndCampStartDateLessThanEqualAndCampEndDateGreaterThanEqual(0,2,currentDate,currentDate);

        results.forEach(it -> executor.submit(this.process(it)));
        Thread.sleep(1000);
    }

    private void process(RestTemplate rest, Bulk_repository churnss) {
      AddOfferRequest AddOffer = new AddOfferRequest("113", churnss.getMsisdn(),churnss.getParam1());

      HttpEntity<AddOfferRequest> entity = new HttpEntity<AddOfferRequest>(AddOffer,headers);

      try {
        ResponseEntity<String> responseEntity = restTemplate.exchange(
                        "api link", HttpMethod.POST, entity, String.class);
         if(responseEntity.getStatusCode() == HttpStatus.OK){
           String response = responseEntity.getBody();
           churnss.setStatus(1);
           churnss.setProcessDate(new Date());
           churnss.setFulfilment_status(response);
           bulk_repositoryRepository.save(churnss);
         }else {
           logger.warn("Record Id: {}, Http Failed Response: {}",churnss.getId(), responseEntity.getStatusCode());
                }
      } catch (RestClientException rce) {
          logger.warn("Record Id: {} Http Failed. ", churnss.getId(), rce);
      }               
    }

}

注意:这是从我的头顶输入的,没有经过测试。但是,应该提供一些指导。

票数 2
EN

Stack Overflow用户

发布于 2019-10-28 02:00:56

票数 0
EN

Stack Overflow用户

发布于 2019-10-28 10:21:37

尝试使用批处理插入/使用Hibernate/JPA更新。这是一个不错的教程

spring.jpa.properties.hibernate.jdbc.batch_size=500

票数 0
EN
页面原文内容由Stack Overflow提供。腾讯云小微IT领域专用引擎提供翻译支持
原文链接:

https://stackoverflow.com/questions/58582177

复制
相关文章

相似问题

领券
问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档