成都创新互联网站制作重庆分公司

如何将数据从Web服务处理到MongoDB中

本篇内容主要讲解“如何将数据从Web服务处理到MongoDB中”，感兴趣的朋友不妨来看看。本文介绍的方法操作简单快捷，实用性强。下面就让小编来带大家学习“如何将数据从Web服务处理到MongoDB中”吧!

创新互联建站长期为1000+客户提供的网站建设服务，团队从业经验10年，关注不同地域、不同群体，并针对不同对象提供差异化的产品和服务；打造开放共赢平台，与合作伙伴共同营造健康的互联网生态环境。为凤城企业提供专业的成都网站建设、做网站，凤城网站改版等技术服务。拥有10余年丰富建站经验和众多成功案例,为您定制开发。

概观

如何创建一个使用Web服务数据并将其插入MongoDB数据库的Spring Batch应用程序。

要求

阅读本文的开发人员必须熟悉Spring Batch（示例）和MongoDB。

环境

  • Mongo数据库部署在MLab中。请按照本快速入门中的步骤操作。

  • 批处理应用程序部署在Heroku PaaS中。详情  请看这里。

  • IDE STS或IntelliJ或Eclipse。

  • Java 8 JDK。

注意：批处理也可以在本地运行。

脚本

全局场景步骤是：

  1. 从Web服务读取数据，在这种情况下：https：//sunrise-sunset.org/api

获取城市列表的坐标，然后调用API以读取日出和日落日期时间。

2.处理数据并提取业务数据

收集数据的业务处理

3.在MongoDB中插入已处理的数据

将处理过的数据保存为mongo文档

编码

  • 输入：本地文件中JSON格式的城市数据列表，如下所示：

[
   {
      “名字”：“Danemark”，
      “城市”：[
         {
            “名字”：“Copenhague”，
            “lat”：55.676098，
            “lng”：12.568337，
            “timeZone”：“CET”
         }，
         {
            “名字”：“奥胡斯”，
            “lat”：56.162939，
            “lng”：10.203921，
            “timeZone”：“CET”
         }，
         {
            “名字”：“欧登塞”，
            “lat”：55.39594，
            “lng”：10.38831，
            “timeZone”：“CET”
         }，
         {
            “名字”：“奥尔堡”，
            “lat”：57.046707，
            “lng”：9.935932，
            “timeZone”：“CET”
         }
      ]
   }
]

我们的场景从本地json文件获取输入数据。映射bean如下：

国豆：

导入 java。io。可序列化 ;
导入 java。util。清单 ;

进口 com。fastxml。杰克逊。注释。JsonIgnoreProperties ;
@JsonIgnoreProperties（ignoreUnknown  =  true）
public  class  BCountry  实现 Serializable {
private  static  final  long  serialVersionUID  =  1L ;
私有 字符串 名称 ;
私人 名单< BCity >  城市 ;
public  BCountry（）{
super（）;
}
public  BCountry（String  name，List < BCity >  cities）{
super（）;
这个。name  =  name ;
这个。城市 =  城市 ;
}
public  BCountry（String  name）{
super（）;
这个。name  =  name ;
}
public  String  getName（）{
返回 名称 ;
}
public  void  setName（String  name）{
这个。name  =  name ;
}
public  List < BCity >  getCities（）{
返回 城市 ;
}
public  void  setCities（List < BCity >  cities）{
这个。城市 =  城市 ;
}
@覆盖
public  int  hashCode（）{
final  int  prime  =  31 ;
int  result  =  1 ;
结果 =  黄金 *  结果 +（（城市 ==  空）？ 0：城市。的hashCode（））;
结果 =  黄金 *  结果 +（（名称 ==  空）？ 0：名称。的hashCode（））;
返回 结果 ;
}
@覆盖
public  boolean  equals（Object  obj）{
if（this  ==  obj）
返回 true ;
if（obj  ==  null）
返回 虚假 ;
如果（的getClass（）！=  OBJ。的getClass（））
返回 虚假 ;
BCountry  other  =（BCountry）obj ;
if（cities  ==  null）{
如果（其他。城市 ！=  空）
返回 虚假 ;
} 否则 如果（！城市。平等（等。城市））
返回 虚假 ;
if（name  ==  null）{
如果（其他。名字 ！=  空）
返回 虚假 ;
} 否则 如果（！名字。平等（其它。名））
返回 虚假 ;
返回 true ;
}
@覆盖
public  String  toString（）{
返回 “BCountry [name =”  +  name  +  “，cities =”  +  cities  +  “]” ;
}
}

和城市豆：

导入 java。io。可序列化 ;
进口 com。fastxml。杰克逊。注释。JsonIgnoreProperties ;
@JsonIgnoreProperties（ignoreUnknown  =  true）
公共 类 BCity  实现 Serializable {
private  static  final  long  serialVersionUID  =  1L ;
private  String  name，timeZone ;
私人 双 拉特，lng ;
public  BCity（）{
super（）;
}
public  BCity（String  name，String  timeZone，double  lat，double  lng）{
super（）;
这个。name  =  name ;
这个。timeZone  =  timeZone ;
这个。lat  =  lat ;
这个。lng  =  lng ;
}
public  String  getName（）{
返回 名称 ;
}
public  void  setName（String  name）{
这个。name  =  name ;
}
public  String  getTimeZone（）{
返回时 区 ;
}
public  void  setTimeZone（String  timeZone）{
这个。timeZone  =  timeZone ;
}
public  double  getLat（）{
返回 纬度 ;
}
public  void  setLat（double  lat）{
这个。lat  =  lat ;
}
public  double  getLng（）{
返回 lng ;
}
public  void  setLng（double  lng）{
这个。lng  =  lng ;
}
@覆盖
public  String  toString（）{
返回 “BCity [name =”  +  name  +  “，timeZone =”  +  timeZone  +  “，lat =”  +  lat  +  “，lng =”  +  lng  +  “]” ;
}
@覆盖
public  int  hashCode（）{
final  int  prime  =  31 ;
int  result  =  1 ;
长 温度 ;
temp  =  Double。doubleToLongBits（lat）;
result  =  prime  *  result  +（int）（temp  ^（temp  >>>  32））;
temp  =  Double。doubleToLongBits（lng）;
result  =  prime  *  result  +（int）（temp  ^（temp  >>>  32））;
结果 =  黄金 *  结果 +（（名称 ==  空）？ 0：名称。的hashCode（））;
结果 =  素 *  结果 +（（的timeZone  ==  空）？ 0：的timeZone。的hashCode（））;
返回 结果 ;
}
@覆盖
public  boolean  equals（Object  obj）{
if（this  ==  obj）
返回 true ;
if（obj  ==  null）
返回 虚假 ;
如果（的getClass（）！=  OBJ。的getClass（））
返回 虚假 ;
BCity  other  =（BCity）obj ;
如果（双。doubleToLongBits的（LAT）！=  双。doubleToLongBits的（其他的。LAT））
返回 虚假 ;
如果（双。doubleToLongBits的（LNG）！=  双。doubleToLongBits的（其他的。LNG））
返回 虚假 ;
if（name  ==  null）{
如果（其他。名字 ！=  空）
返回 虚假 ;
} 否则 如果（！名字。平等（其它。名））
返回 虚假 ;
if（timeZone  ==  null）{
如果（其他。的timeZone  ！=  空）
返回 虚假 ;
} 否则 如果（！的timeZone。平等（其它。的timeZone））
返回 虚假 ;
返回 true ;
}
}

批量阅读器实现@LineMapper。您可以使读者适应我们的数据源（示例）：

导入 java。util。清单 ;
进口 组织。弹簧框架。批次。项目。档案。LineMapper ;
进口 com。ahajri。批次。豆子。BCountry ;
进口 com。fastxml。杰克逊。数据绑定。ObjectMapper ;
进口 com。fastxml。杰克逊。数据绑定。类型。CollectionType ;
公共 类 BCountryJsonLineMapper  实现了 LineMapper < List < BCountry >> {
private  final  ObjectMapper  mapper  =  new  ObjectMapper（）;
@覆盖
public  List < BCountry >  mapLine（String  line，int  lineNumber）throws  Exception {
CollectionType  collectionType  = mapper。getTypeFactory（）。constructCollectionType（列表。类，BCountry。类）;
返回 映射器。readValue（line，collectionType）;
}
}

处理数据批处理时，检查同一天某个城市的业务处理数据是否已保存在数据库中。在MongoDB的搜索数据的方式就是在这个详细的岗位。

ItemProcessor将@BCountry对象转换为MongoDB Document对象。该过程详述如下：

public  class  BCountryPrayTimeEventItemProcessor  实现 ItemProcessor < List < BCountry >，List < Document >> {
private  static  final  String  EVENTS_COLLECTION_NAME  =  “event” ;
private  static  final  Logger  LOG  =  LoggerFactory。getLogger（BCountryPrayTimeEventItemProcessor。类）;
@Autowired
private  PrayTimeService  prayTimeService ;
@Autowired
private  CloudMongoService  cloudMongoService ;
@覆盖
public  List < Document >  进程（List < BCountry >  items）抛出 Exception {
final  List < Document >  docs  =  new  ArrayList <>（）;
物品。stream（）。forEach（item  - > {
final  String  countryName  =  item。getName（）;
项目。getCities（）。stream（）。forEach（c  - > {
final  Document  prayTimeCityEventDoc  =  new  Document（）;
//循环城市并为今天提取祈祷时间
final  String  cityName  =  c。getName（）;
final  String  cityTimeZone  =  c。getTimeZone（）;
final  double  lat  =  c。getLat（）;
final  double  lng  =  c。getLng（）;
final  LocalDateTime  nowOfCity  =  LocalDateTime。现在（了zoneid。的（cityTimeZone））;
final  QueryParam [] queryParams  =  new  QueryParam [ 5 ];
queryParams [ 0 ] =  新 QueryParam（“CITY_NAME” ，OperatorEnum。EQ。名称（），的cityName）;
queryParams [ 1 ] =  新 QueryParam（“EVENT_TYPE” ，OperatorEnum。EQ。名称（），事件类型。PRAY_TIME。名称（））;
queryParams [ 2 ] =  新 QueryParam（“月”，OperatorEnum。EQ。名称（），nowOfCity。getMonthValue（））;
queryParams [ 3 ] =  新 QueryParam（“DAY_OF_MONTH” ，OperatorEnum。EQ。名称（），nowOfCity。getDayOfMonth（））;
queryParams [ 4 ] =  新 QueryParam（“COUNTRY_NAME” ，OperatorEnum。EQ。名称（），国家名称）;
List < Document >  foundEvents  =  null ;
尝试 {
foundEvents  =  cloudMongoService。搜索（EVENTS_COLLECTION_NAME，queryParams）;
} catch（BusinessException  e1）{
记录。错误（“====>未找到城市祈祷时间”  +  的cityName  +  “对”  +  nowOfCity。getDayOfMonth（）+  “/”
+  nowOfCity。getMonthValue（））;
}
尝试 {
如果（CollectionUtils。的isEmpty（foundEvents））{
//祈祷时间尚未创建
prayTimeCityEventDoc。put（“country_name”，countryName）;
prayTimeCityEventDoc。put（“city_name”，cityName）;
prayTimeCityEventDoc。把（“EVENT_TYPE” ，事件类型。PRAY_TIME。名称（））;
prayTimeCityEventDoc。把（“复发”，RecurringEnum。YEARLY。名称（））;
prayTimeCityEventDoc。把（“月”，nowOfCity。getMonthValue（））;
prayTimeCityEventDoc。把（“DAY_OF_MONTH” ，nowOfCity。getDayOfMonth（））;
prayTimeCityEventDoc。put（“lat”，lat）;
prayTimeCityEventDoc。put（“lng”，lng）;
prayTimeCityEventDoc。把（“CREATION_DATE” ，HCDateUtils。convertToDateViaSqlTimestamp（nowOfCity））;
final  Map < String，Object >  prayInfos  =  prayTimeService。getPrayTimeByLatLngDate（lat，lng，
日期。从（nowOfCity。atZone（了zoneid。的（cityTimeZone））。toInstant（）），cityTimeZone）;
prayTimeCityEventDoc。把（“pray_infos” ，文件。解析（新 GSON（）的toJSON（prayInfos）））;
docs。add（prayTimeCityEventDoc）;
} else {
记录。信息（字符串。格式（“====>祈祷的时间已经存在的城市：％S，月：％d，日：％d” ，
cityName，nowOfCity。getMonthValue（），nowOfCity。getDayOfMonth（）））;
}
} catch（BusinessException  e）{
记录。错误（“计算祈祷时间时出问题：”，e）;
抛出 新的 RuntimeException（e）;
}
}）;
}）;
返回 文档 ;
}
}

批量配置类：

@组态
@EnableBatchProcessing
@EnableScheduling
公共 类 BatchConfiguration {
private  static  final  String  SCANDINAVIAN_COUNTRIES_JSON_FILE  =  “scandinavian-countries.json” ;
private  static  final  String  EVENT_COLLECTION_NAME  =  “event_collection” ;
private  static  final  Logger  LOG  =  LoggerFactory。getLogger（BatchConfiguration。类）;
@Autowired
private  JobBuilderFactory  jobBuilderFactory ;
@Autowired
private  StepBuilderFactory  stepBuilderFactory ;
@Autowired
私有的 MLabMongoService  mlabMongoService ;
@豆
public  ResourcelessTransactionManager  transactionManager（）{
返回 新的 ResourcelessTransactionManager（）;
}
@豆
public  MapJobRepositoryFactoryBean  mapJobRepositoryFactory（ResourcelessTransactionManager  txManager）
抛出 异常 {
MapJobRepositoryFactoryBean  factory  =  new  MapJobRepositoryFactoryBean（txManager）;
工厂。afterPropertiesSet（）;
返回 工厂 ;
}
@豆
public  JobRepository  jobRepository（MapJobRepositoryFactoryBean  factory）抛出 异常 {
return（JobRepository）工厂。getObject（）;
}
private  SimpleJobLauncher  jobLauncher ;

@豆
public  SimpleJobLauncher  jobLauncher（JobRepository  jobRepository）{
jobLauncher。setJobRepository（jobRepository）;
return  jobLauncher ;
}
@PostConstruct
private  void  initJobLauncher（）{
jobLauncher  =  new  SimpleJobLauncher（）;
}
@豆
FlatFileItemReader < List < BCountry >>  reader（）{
FlatFileItemReader < List < BCountry >>  reader  =  new  FlatFileItemReader <>（）;
读者。setName（“scandinaviandCountriesReader”）;
读者。setResource（new  ClassPathResource（SCANDINAVIAN_COUNTRIES_JSON_FILE））;
读者。setLineMapper（new  BCountryJsonLineMapper（））;
回报 读者 ;
}
@豆
public  ItemWriter < List < Document >>  writer（）{
返回 新的 ItemWriter < List < Document >>（）{
@覆盖
public  void  write（List <？ extends  List < Document >>  items）抛出 Exception {
尝试 {
如果（！CollectionUtils。的isEmpty（项目）&&  项目。大小（）>  0）{
List < Document >  flatDocs  =  items。stream（）。flatMap（List :: stream）。收集（收藏家。toList（））;
mlabMongoService。insertMany（EVENT_COLLECTION_NAME，flatDocs）;
} else {
记录。警告（“没有事件可以救......”）;
}
} catch（BusinessException  e）{
抛出 新的 RuntimeException（e）;
}
}
};
}

@豆
public  BCountryTimeEventItemProcessor  processor（）{
返回 新的 BCountryTimeEventItemProcessor（）;
}

@豆
public  Job  scandvTimeJob（）{
返回 jobBuilderFactory。get（“scandvTimeJob”）。incrementmenter（new  RunIdIncrementer（））。流程（step1（））。结束（）
。build（）;
}

@豆
public  Step  step1（）{
返回 stepBuilderFactory。得到（“step1”）。< List < BCountry >，List < Document >> chunk（10）。读者（读者（））
。处理器（处理器（））。作家（作家（））。build（）;
}
// end :: jobstep []

  //每天午夜15分钟
@Scheduled（cron  =  “0 15 0 * * *”）
public  void  startScandvEventTimeJob（）throws  Exception {
记录。info（“====>工作开始时间：”  +  新 日期（））;
JobParameters  param  =  new  JobParametersBuilder（）。addString（“作业ID” ，字符串。的valueOf（系统。的currentTimeMillis（）））
。toJobParameters（）;
JobExecution  执行 =  jobLauncher。run（scandvPrayTimeJob（），param）;
记录。信息（“====>工作完成了状态：”  +  执行。的getStatus（））;
}
}

部署de Batch到Heroku：

git add。
git commit -m  “Deploy Batch”
git push heroku master

注意：要禁用默认批量启动，请将此添加到application.yml

s p r i n g：
  b a t c h：
    j o b：
      Ë Ñ 一个b 升é d：˚F 一升小号ë

到此，相信大家对“如何将数据从Web服务处理到MongoDB中”有了更深的了解，不妨来实际操作一番吧！这里是创新互联网站，更多相关内容可以进入相关频道进行查询，关注我们，继续学习！


本文标题：如何将数据从Web服务处理到MongoDB中
标题URL：http://cxhlcq.com/article/peicgh.html

其他资讯

全国免费咨询:

400-028-6601

业务咨询：028-86922220 / 13518219792

节假值班：18980820575 / 13518219792

联系地址：成都市太升南路288号锦天国际A幢1002号

Copyright © 成都创新互联科技有限公司重庆分公司 渝ICP备2021005571号
友情链接:： 成都网站建设 重庆网站建设 四川网站建设 重庆建设网站 移动服务器托管 成都服务器托管 云服务器 广告设计制作 重庆网页设计 重庆做网站 重庆网站制作 重庆网站建设 重庆网站公司 渝中网站制作 重庆网站设计

在线咨询

微信咨询

电话咨询

028-86922220（工作日）

18980820575（7×24）

提交需求

返回顶部