Fedex / iClear-Connect-Pre-Clearance

Go to a project
Toggle navigation Toggle navigation pinning
tao.mo
c8f0b3ab 1 parent 9f0beaf7

All | 修改 | 配置文件调整、Imp001功能开发 

mt
2024年11月5日17:05:32
Expand all
Inline Side-by-side
Showing 53 changed files with 1435 additions and 200 deletions
......@@ -6,15 +6,16 @@ import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.autoconfigure.security.servlet.SecurityAutoConfiguration;
import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.boot.web.servlet.support.SpringBootServletInitializer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.transaction.annotation.EnableTransactionManagement;
import org.springframework.web.WebApplicationInitializer;
import org.springframework.web.client.RestTemplate;
import springfox.documentation.swagger2.annotations.EnableSwagger2;
@SpringBootApplication(exclude = SecurityAutoConfiguration.class)
@MapperScan({"com.fedex.connect.common.dao.*.**","com.fedex.connect.customer.repository.dao"})
@MapperScan({"com.fedex.connect.common.dao.*.**","com.fedex.connect.common.dependencies.repository.dao","com.fedex.connect.customer.repository.dao"})
@EnableTransactionManagement
@ComponentScan(basePackages = {"com.fedex.connect.common.dependencies.i18n", "com.fedex.connect.customer"})
@EnableSwagger2
public class CustomerApplication extends SpringBootServletInitializer implements WebApplicationInitializer {
public static void main(String[] args) {
......@@ -25,4 +26,9 @@ public class CustomerApplication extends SpringBootServletInitializer implements
protected SpringApplicationBuilder configure(SpringApplicationBuilder builder) {
return builder.sources(CustomerApplication.class);
}
@Bean
public RestTemplate restTemplate() {
return new RestTemplate();
}
}
......
spring:
datasource:
jndi-name: jdbc/icleartwDS
jndi-name: jdbc/iclearConnectDS
profiles:
#系统环境
env: prod
......
spring:
datasource:
jndi-name: jdbc/icleartwDS
jndi-name: jdbc/iclearConnectDS
profiles:
#系统环境
env: test
......
spring:
datasource:
jndi-name: jdbc/icleartwDS
jndi-name: jdbc/iclearConnectDS
profiles:
#系统环境
env: uat
......
package com.fedex.connect.common.dependencies.autoconfiguration;
import org.springframework.context.annotation.ComponentScan;
@ComponentScan(value = {"com.fedex.connect.common.dependencies"})
public class AutoConfiguration {
}
......@@ -14,19 +14,20 @@ import java.util.Locale;
*/
public class MessageLocaleResolver implements LocaleResolver {
@Value("${export.language}")
private String exportTwLanguage;
// @Value("${export.language}")
// private String exportTwLanguage;
@Override
public Locale resolveLocale(HttpServletRequest request) {
Locale locale;
if (StringUtils.isNotEmpty(exportTwLanguage)) {
//获取知道配置文件中的语言为： exportTwLanguage;
locale = new Locale(exportTwLanguage);
} else {
//读取不到指定语言文件，采用默认中文。即默认的message.properties
locale = Locale.getDefault();
}
// Locale locale;
// if (StringUtils.isNotEmpty(exportTwLanguage)) {
// //获取知道配置文件中的语言为： exportTwLanguage;
// locale = new Locale(exportTwLanguage);
// } else {
// //读取不到指定语言文件，采用默认中文。即默认的message.properties
// locale = Locale.getDefault();
// }
Locale locale = request.getLocale();
return locale;
}
......
package com.fedex.connect.common.dependencies.util;
import com.alibaba.fastjson.JSON;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.ParameterizedTypeReference;
import org.springframework.http.*;
import org.springframework.stereotype.Component;
import org.springframework.web.client.RestTemplate;
/**
* rpc调用工具类
* mt
* 2023年8月31日15:26:40
*/
@Component
public class RPCUtils {
@Autowired
private RestTemplate restTemplate;
/**
* 没有返回值的RPC调用
*
* @param url 请求地址
* @param body 请求体
* @return
*/
public Integer rpcInvoke(String url, String body) {
try {
HttpHeaders headers = new HttpHeaders();
headers.setContentType(MediaType.APPLICATION_JSON);
HttpEntity<String> httpEntity = new HttpEntity<>(body, headers);
ResponseEntity resp = restTemplate.exchange(url, HttpMethod.POST, httpEntity,
new ParameterizedTypeReference<String>() {
});
HttpStatus httpStatus = resp.getStatusCode();
return httpStatus.value();
} catch (Exception e) {
throw e;
}
}
/**
* 有实体类有返回值的RPC调用
*
* @param url 请求地址
* @param body 请求体
* @param r 返回值类型
* @param <R> 返回值类
* @return
*/
public <R> R rpcResponseInvoke(String url, String body, Class<R> r) {
try {
if (body != null) {
HttpHeaders headers = new HttpHeaders();
headers.setContentType(MediaType.APPLICATION_JSON);
HttpEntity<String> httpEntity = new HttpEntity<>(body, headers);
ResponseEntity<String> resp = restTemplate.exchange(url, HttpMethod.POST, httpEntity,
new ParameterizedTypeReference<String>() {
});
// String bodyJson = JSON.toJSONString(resp.getBody());
String bodyJson = resp.getBody();
R res = JSON.parseObject(bodyJson, r);
return res;
} else {
ResponseEntity<String> resp = restTemplate.exchange(url, HttpMethod.POST, null,
new ParameterizedTypeReference<String>() {
});
// String bodyJson = JSON.toJSONString(resp.getBody());
String bodyJson = resp.getBody();
R res = JSON.parseObject(bodyJson, r);
return res;
}
} catch (Exception e) {
throw e;
}
}
}
\ No newline at end of file
org.springframework.boot.autoconfigure.EnableAutoConfiguration=com.fedex.connect.common.dependencies.autoconfiguration.AutoConfiguration
\ No newline at end of file
......@@ -6,12 +6,14 @@ import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.autoconfigure.security.servlet.SecurityAutoConfiguration;
import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.boot.web.servlet.support.SpringBootServletInitializer;
import org.springframework.context.annotation.Bean;
import org.springframework.transaction.annotation.EnableTransactionManagement;
import org.springframework.web.WebApplicationInitializer;
import org.springframework.web.client.RestTemplate;
import springfox.documentation.swagger2.annotations.EnableSwagger2;
@SpringBootApplication(exclude = SecurityAutoConfiguration.class)
@MapperScan({"com.fedex.connect.common.dao.*.**","com.fedex.connect.kafka.repository.dao"})
@MapperScan({"com.fedex.connect.common.dao.*.**","com.fedex.connect.common.dependencies.repository.dao","com.fedex.connect.kafka.repository.dao"})
@EnableTransactionManagement
@EnableSwagger2
public class KafkaApplication extends SpringBootServletInitializer implements WebApplicationInitializer {
......@@ -23,4 +25,9 @@ public class KafkaApplication extends SpringBootServletInitializer implements We
protected SpringApplicationBuilder configure(SpringApplicationBuilder builder) {
return builder.sources(KafkaApplication.class);
}
@Bean
public RestTemplate restTemplate() {
return new RestTemplate();
}
}
......
......@@ -203,6 +203,11 @@ public class Dm501Util {
AssignmentFieldUtils.assignmentTableBaseField(resultConsignment);
rsConsignment = resultConsignment;
}else{
//原产国全称
// String originCountry = consignment.getOriginCountry();
//发货人国家二字码
// String shipperCountry = consignment.getShipperCountry();
BeanUtils.copyProperties(ceInfo,consignment);
rsConsignment = consignment;
}
......
......@@ -6,12 +6,14 @@ import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.autoconfigure.security.servlet.SecurityAutoConfiguration;
import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.boot.web.servlet.support.SpringBootServletInitializer;
import org.springframework.context.annotation.Bean;
import org.springframework.transaction.annotation.EnableTransactionManagement;
import org.springframework.web.WebApplicationInitializer;
import org.springframework.web.client.RestTemplate;
import springfox.documentation.swagger2.annotations.EnableSwagger2;
@SpringBootApplication(exclude = SecurityAutoConfiguration.class)
@MapperScan({"com.fedex.connect.common.dao.*.**","com.fedex.connect.manager.repository.dao"})
@MapperScan({"com.fedex.connect.common.dao.*.**","com.fedex.connect.common.dependencies.repository.dao","com.fedex.connect.manager.repository.dao"})
@EnableTransactionManagement
@EnableSwagger2
public class ManagerApplication extends SpringBootServletInitializer implements WebApplicationInitializer {
......@@ -23,4 +25,9 @@ public class ManagerApplication extends SpringBootServletInitializer implements
protected SpringApplicationBuilder configure(SpringApplicationBuilder builder) {
return builder.sources(ManagerApplication.class);
}
@Bean
public RestTemplate restTemplate() {
return new RestTemplate();
}
}
......
spring:
datasource:
url: jdbc:oracle:thin:@192.168.1.250:1521:orcl
username: icleartw
password: icleartw
username: iclearimp
password: iclearimp
driver-class-name: oracle.jdbc.OracleDriver
profiles:
#系统环境
......
spring:
datasource:
jndi-name: jdbc/icleartwDS
jndi-name: jdbc/iclearConnectDS
profiles:
#系统环境
env: prod
......
spring:
datasource:
jndi-name: jdbc/icleartwDS
jndi-name: jdbc/iclearConnectDS
profiles:
#系统环境
env: test
......
spring:
datasource:
jndi-name: jdbc/icleartwDS
jndi-name: jdbc/iclearConnectDS
profiles:
#系统环境
env: uat
......
......@@ -6,14 +6,17 @@ import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.autoconfigure.security.servlet.SecurityAutoConfiguration;
import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.boot.web.servlet.support.SpringBootServletInitializer;
import org.springframework.context.annotation.Bean;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.transaction.annotation.EnableTransactionManagement;
import org.springframework.web.WebApplicationInitializer;
import org.springframework.web.client.RestTemplate;
import springfox.documentation.swagger2.annotations.EnableSwagger2;
@SpringBootApplication(exclude = SecurityAutoConfiguration.class)
@MapperScan({"com.fedex.connect.common.dao.*.**","com.fedex.connect.task.repository.dao"})
@MapperScan({"com.fedex.connect.common.dao.*.**","com.fedex.connect.common.dependencies.repository.dao","com.fedex.connect.task.repository.dao"})
@EnableTransactionManagement
@EnableSwagger2
@EnableScheduling
public class TaskApplication extends SpringBootServletInitializer implements WebApplicationInitializer {
public static void main(String[] args) {
SpringApplication.run(TaskApplication.class, args);
......@@ -23,4 +26,9 @@ public class TaskApplication extends SpringBootServletInitializer implements Web
protected SpringApplicationBuilder configure(SpringApplicationBuilder builder) {
return builder.sources(TaskApplication.class);
}
@Bean
public RestTemplate restTemplate() {
return new RestTemplate();
}
}
......
package com.fedex.connect.task.annotation;
import java.lang.annotation.*;
/**
* @Author mt
* @Description 输出执行时间日志
* @Date 2024/5/23
*/
@Target({ ElementType.PARAMETER, ElementType.METHOD })
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface ProcessingTime {
}
\ No newline at end of file
package com.fedex.connect.task.aspect;
import lombok.extern.slf4j.Slf4j;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.Signature;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.aspectj.lang.annotation.Pointcut;
import org.aspectj.lang.reflect.MethodSignature;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
import java.lang.reflect.Method;
/**
* @Author mt
* @Description 当前记录各定时任务执行时间
* @Date 2024/5/23
*/
@Aspect
@Slf4j
@Component
public class ExecuteAspect {
private Logger log = LoggerFactory.getLogger(ExecuteAspect.class);
@Pointcut("@annotation(com.fedex.connect.task.annotation.ProcessingTime)")
public void executePointCut()
{
}
/**
* aop输出方法执行时间
* @param pjp
* @return
*/
@Around(value = "executePointCut()")
public Object executeAround(ProceedingJoinPoint pjp){
Object obj = null;
try {
Signature signature = pjp.getSignature();
MethodSignature methodSignature = (MethodSignature) signature;
Method method = methodSignature.getMethod();
long begin = System.currentTimeMillis();
log.info("{} Job Task Start......", method.getName());
obj = pjp.proceed();
long end = System.currentTimeMillis();
log.info("{} Job Task end......execution time--{}ms", method.getName(),(end-begin));
}catch(Throwable throwable){
log.error("ExecuteAspect.executeAround error : " + throwable.getMessage() , throwable);
}
return obj;
}
}
\ No newline at end of file
package com.fedex.connect.task.config;
import lombok.Data;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Configuration;
/**
* @Author mt
* @Description 类说明 所有properties
* @Date 2024年4月18日20:59:34
*/
@Data
@Configuration
public class PropertiesConfig {
@Value("${spring.profiles.env}")
private String env;
@Value("${rpc.url.rpcKafkaReceive1}")
private String rpcKafkaReceive145;
@Value("${rpc.url.rpcKafkaReceive2}")
private String rpcKafkaReceive146;
}
\ No newline at end of file
package com.fedex.connect.task.config;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.SchedulingConfigurer;
import org.springframework.scheduling.config.ScheduledTaskRegistrar;
import java.util.concurrent.Executors;
/**
* 动态定时任务
* @author Administrator
*/
@Configuration
public class SchedulingConfigurerConfig implements SchedulingConfigurer {
@Override
public void configureTasks(ScheduledTaskRegistrar scheduledTaskRegistrar) {
//设定一个长度10的定时任务线程池
scheduledTaskRegistrar.setScheduler(Executors.newScheduledThreadPool(10));
}
}
\ No newline at end of file
package com.fedex.connect.task.data.bo;
import lombok.Data;
import java.util.Date;
/**
* @Author mt
* @Description 类说明 文件信息bo
* @Date 2024/5/30
*/
@Data
public class Imp001FileInfoBo {
//文件类型id
Long fileTypeId;
//文件类型名称
String fileTypeName;
//文件名称
String fileName;
//生成时间
Date pushTime;
}
package com.fedex.connect.task.data.bo;
import com.fedex.connect.common.model.bi.DictionaryEntries;
import com.fedex.connect.task.data.dto.imp001.Imp001RootJsonDto;
import lombok.Data;
import java.util.Dictionary;
import java.util.List;
/**
* @Author mt
* @Description 类说明 tw001生成bo
* @Date 2024/5/24
*/
@Data
public class Imp001GenerateBo {
//tw001报文对象
Imp001RootJsonDto imp001RootJsonDto;
//tw001临时文件夹目录
String workFilePath;
//需要打包压缩包文件信息名称
List<Imp001FileInfoBo> fileInfoList;
//json Dir对象
DictionaryEntries jsonDictionary;
//zip文件名称
String zipFileName;
//zip推送目标目录
String zipFileTargetPath;
}
\ No newline at end of file
package com.fedex.connect.task.data.bo;
import lombok.Data;
import java.util.List;
/**
* @Author mt
* @Description tw001推送bo
* @Date 2024/5/30
*/
@Data
public class Imp001PushBo {
//压缩包中文件列表
List<String> filePaths;
//zip文件完整名称
String zipFilePath;
}
package com.fedex.connect.task.data.dto.imp001;
import com.fedex.connect.common.dependencies.arithmetic.AESUtil;
import lombok.Data;
import org.apache.commons.lang3.StringUtils;
import java.io.Serializable;
/**
* @Author mt
* @Description 申报信息
* @Date 2024/5/23
*/
@Data
public class Imp001Consignment implements Serializable {
/**
* @Author mt
* @Description 提单号码
* 12位数字，提单号码，可以重复
* @Date 2024/5/23
*/
private String deliveryNo;
/**
* @Author mt
* @Description 联络人
* @Date 2024/5/23
*/
private String liaisons;
/**
* @Author mt
* @Description 货物输出统一编号
* 8-12位，数字加英文大写
* @Date 2024/5/23
*/
private String consigneeNo;
/**
* @Author mt
* @Description 联络人email
* 标准邮件格式：xxx@xx.com
* @Date 2024/5/23
*/
private String email;
/**
* @Author mt
* @Description 联络电话
* 包含：数字、空格-()（）#:：,，;；+.
* @Date 2024/5/23
*/
private String tel;
/**
* @Author mt
* @Description 分机
* 包含：数字、空格-()（）#:：,，;；+.
* @Date 2024/5/23
*/
private String extensionTel;
/**
* @Author mt
* @Description 行动电话
* 包含：数字、空格-()（）#:：,，;；+.
* @Date 2024/5/23
*/
private String phone;
/**
* @Author mt
* @Description 特殊交代事项
* @Date 2024/5/23
*/
private String customsClearanceInstruction;
/**
* @Author mt
* @Description 对部分字段进行解密
* @Date 2024/6/13
* @param
* @return void
*/
public void decode() {
if (StringUtils.isNotBlank(this.getLiaisons())){
this.setLiaisons(AESUtil.decode_default(this.getLiaisons()));
}
if (StringUtils.isNotBlank(this.getTel())){
this.setTel(AESUtil.decode_default(this.getTel()));
}
if (StringUtils.isNotBlank(this.getPhone())){
this.setPhone(AESUtil.decode_default(this.getPhone()));
}
}
}
\ No newline at end of file
package com.fedex.connect.task.data.dto.imp001;
import lombok.Data;
import java.io.Serializable;
/**
* @Author mt
* @Description 申报信息集合
* @Date 2024/5/23
*/
@Data
public class Imp001Consignments implements Serializable {
/**
* @Author mt
* @Description 申报信息
* @Date 2024/5/23
*/
private Imp001Consignment consignment;
/**
* @Author mt
* @Description 文件列表信息
* @Date 2024/5/23
*/
private Imp001Files[] files;
}
\ No newline at end of file
package com.fedex.connect.task.data.dto.imp001;
import lombok.Data;
import java.io.Serializable;
/**
* @Author mt
* @Description 数据
* @Date 2024/5/23
*/
@Data
public class Imp001Data implements Serializable {
/**
* @Author mt
* @Description 运单集合
* @Date 2024/5/23
*/
private Imp001Consignments[] consignments;
}
\ No newline at end of file
package com.fedex.connect.task.data.dto.imp001;
import lombok.Data;
import java.io.Serializable;
/**
* @Author mt
* @Description 文件列表信息
* @Date 2024/5/23
*/
@Data
public class Imp001Files implements Serializable {
/**
* @Author mt
* @Description 文件名称
* 使用iclearConnect中存储的文件名称，文件名没有固定格式(因为客户上传的文件名是不固定的)
* @Date 2024/5/23
*/
private String name;
/**
* @Author mt
* @Description 文件来源
* "文件类型
* (1:系统生成的的“聯絡方式與特殊交待事項”。2：系统生成的“出口報關檢核表”。3：用户上传的文件)"
* @Date 2024/5/23
*/
private Integer source;
/**
* @Author mt
* @Description 文件名后缀
* 小写
* @Date 2024/5/23
*/
private String suffix;
/**
* @Author mt
* @Description 文件大小
* 字节数
* @Date 2024/5/23
*/
private Integer fileSizeByte;
}
\ No newline at end of file
package com.fedex.connect.task.data.dto.imp001;
import com.alibaba.fastjson.annotation.JSONField;
import lombok.Data;
import java.io.Serializable;
/**
* @Author mt
* @Description tw001消息主体
* @Date 2024/5/23
*/
@Data
public class Imp001RootJsonDto implements Serializable {
//由发送方定义的序列号，唯一 UUID JSON自动生成
@JSONField(name="messageId",ordinal = 1)
private String messageId;
//消息代码/类型
@JSONField(name="messageCode",ordinal = 2)
private String messageCode;
//发送程序ID
@JSONField(name="senderId",ordinal = 3)
private String senderId;
//接受程序ID
@JSONField(name="receiverId",ordinal = 4)
private String receiverId;
//发送时间,时间戳
@JSONField(name="sendTime",ordinal = 5)
private String sendTime;
@JSONField(name="data",ordinal = 6)
private Imp001Data data;
}
\ No newline at end of file
package com.fedex.connect.task.job;
import com.fedex.connect.task.annotation.ProcessingTime;
import com.fedex.connect.task.service.biz.IPushObService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
/**
* @Author mt
* @Description IMP001申报数据推送任务
* @Date 2024/11/5
*/
@Component
public class Imp001Job {
@Autowired
protected IPushObService sendObService;
/**
* 推送Imp001数据给到进口组
*/
@ProcessingTime
@Scheduled(cron = "${export.task.allocation.sendOb}")
public void sendObTask() {
sendObService.sendOb();
}
/**
* 推送OB数据错误重试
*/
@ProcessingTime
@Scheduled(cron = "${export.task.allocation.sendObRetry}")
public void sendObRetryTask() {
sendObService.sendObRetry();
}
}
package com.fedex.connect.task.job;
import com.fedex.connect.task.annotation.ProcessingTime;
import com.fedex.connect.task.service.sys.IKafkaDmService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
/**
* @Author mt
* @Description kafka定时任务，处理kafka相关数据
* @Date 2024/6/17
*/
@Component
public class KafkaJob {
@Autowired
private IKafkaDmService kafkaDmService;
/**
* kafka补偿定时任务
*/
@ProcessingTime
@Scheduled(cron = "${export.task.allocation.kafkaRetry}")
public void kafkaRetryTask() {
kafkaDmService.scanKafkaTemporaryStorage();
}
}
package com.fedex.connect.task.repository.base;
import com.fedex.connect.common.dao.sys.KafkaStorageHistoryMapper;
import com.fedex.connect.common.dao.sys.KafkaTemporaryStorageMapper;
import com.fedex.connect.task.repository.dao.KafkaTemporaryStorageMapperExt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
public class AbstractDaoRepository {
protected Logger log = LoggerFactory.getLogger(this.getClass());
@Autowired
protected KafkaTemporaryStorageMapperExt kafkaTemporaryStorageMapperExt;
@Autowired
protected KafkaTemporaryStorageMapper kafkaTemporaryStorageMapper;
@Autowired
protected KafkaStorageHistoryMapper kafkaStorageHistoryMapper;
}
\ No newline at end of file
package com.fedex.connect.task.repository.dao;
import com.fedex.connect.common.model.sys.KafkaTemporaryStorage;
import org.apache.ibatis.annotations.Mapper;
import org.apache.ibatis.annotations.Param;
import java.util.Date;
import java.util.List;
/**
* kafka临时表dao
* mt
* 2022年11月3日15:30:53
*/
@Mapper
public interface KafkaTemporaryStorageMapperExt {
/**
* 查找临时表中大于一个小时待处理以及处理失败，并且处理次数小于3次
* @return
*/
List<KafkaTemporaryStorage> findKafkaTemporaryStoragePending(@Param("curDate") Date curDate);
/**
* 根据集合批量删除kafka临时表
* @param list
* @return
*/
int deleteInBatch(@Param("list") List<KafkaTemporaryStorage> list);
}
\ No newline at end of file
<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd" >
<!--
mapper为映射的根节点，用来管理DAO接口
namespace指定DAO接口的完整类名，表示mapper配置文件管理哪个DAO接口(包.接口名)
mybatis会依据这个接口动态创建一个实现类去实现这个接口，而这个实现类是一个Mapper对象
-->
<mapper namespace="com.fedex.connect.task.repository.dao.KafkaTemporaryStorageMapperExt">
<resultMap id="BaseResultMap" type="com.fedex.connect.common.model.sys.KafkaTemporaryStorage">
<id column="ID" jdbcType="NUMERIC" property="id" />
<result column="MESSAGE_ID" jdbcType="VARCHAR" property="messageId" />
<result column="MESSAGE_CODE" jdbcType="VARCHAR" property="messageCode" />
<result column="SENDER_ID" jdbcType="VARCHAR" property="senderId" />
<result column="RECEIVER_ID" jdbcType="VARCHAR" property="receiverId" />
<result column="SEND_TIME" jdbcType="TIMESTAMP" property="sendTime" />
<result column="CONSIGNMENT_CODE" jdbcType="VARCHAR" property="consignmentCode" />
<result column="FREQUENCY" jdbcType="NUMERIC" property="frequency" />
<result column="STATUS" jdbcType="NUMERIC" property="status" />
<result column="REMARK" jdbcType="VARCHAR" property="remark" />
<result column="CREATE_TIME" jdbcType="TIMESTAMP" property="createTime" />
</resultMap>
<sql id="Base_Column_List">
ID, MESSAGE_ID, MESSAGE_CODE, SENDER_ID, RECEIVER_ID, SEND_TIME, CONSIGNMENT_CODE,
FREQUENCY, STATUS, REMARK, CREATE_TIME
</sql>
<select id="findKafkaTemporaryStoragePending" resultMap="BaseResultMap">
SELECT * FROM T_SYS_KAFKA_TEMPORARY_STORAGE
WHERE (cast(#{curDate,jdbcType=TIMESTAMP} as date) - CREATE_TIME) * 24 * 60 &gt;= 10
AND (STATUS = 0 OR STATUS = 2) AND ROWNUM &lt;= 1000
</select>
<delete id="deleteInBatch">
DELETE T_SYS_KAFKA_TEMPORARY_STORAGE WHERE ID IN
<foreach item="item" index="index" collection="list" open="(" separator="," close=")">
#{item.id}
</foreach>
</delete>
</mapper>
\ No newline at end of file
package com.fedex.connect.task.repository.repo;
import com.fedex.connect.common.model.sys.KafkaStorageHistory;
import java.util.List;
/**
* kafka历史表dao
* mt
* 2022年11月3日15:30:53
*/
public interface IKafkaStorageHistoryRepository {
KafkaStorageHistory save(KafkaStorageHistory entity);
void saveAll(List<KafkaStorageHistory> list);
}
package com.fedex.connect.task.repository.repo;
import com.fedex.connect.common.model.sys.KafkaTemporaryStorage;
import java.util.List;
/**
* kafka临时表dao
* mt
* 2022年11月3日15:30:53
*/
public interface IKafkaTemporaryStorageRepository {
/**
* 查找临时表中大于一个小时待处理以及处理失败，并且处理次数小于3次
* @return
*/
List<KafkaTemporaryStorage> findKafkaTemporaryStoragePending();
/**
* 批量删除kafka临时表
* @param list
* @return
*/
int deleteInBatch(List<KafkaTemporaryStorage> list);
/**
* 批量保存
* @param list
* @return
*/
List<KafkaTemporaryStorage> saveAll(List<KafkaTemporaryStorage> list);
}
\ No newline at end of file
package com.fedex.connect.task.repository.repo.impl;
import com.fedex.connect.common.model.sys.KafkaStorageHistory;
import com.fedex.connect.task.repository.base.AbstractDaoRepository;
import com.fedex.connect.task.repository.repo.IKafkaStorageHistoryRepository;
import org.springframework.stereotype.Repository;
import java.util.List;
/**
* kafka历史表dao
* mt
* 2022年11月3日15:30:53
*/
@Repository
public class KafkaStorageHistoryRepositoryImpl extends AbstractDaoRepository implements IKafkaStorageHistoryRepository {
@Override
public KafkaStorageHistory save(KafkaStorageHistory entity){
if(entity != null) {
if (entity.getId() == null || entity.getId().longValue() <= 0) {
kafkaStorageHistoryMapper.insertSelective(entity);
} else if (entity.getId() != null && entity.getId().longValue() > 0){
kafkaStorageHistoryMapper.updateByPrimaryKeySelective(entity);
}
}
return entity;
}
@Override
public void saveAll(List<KafkaStorageHistory> list){
list.stream().forEach(p->{
if(p.getId() == null || p.getId().longValue() <=0){
kafkaStorageHistoryMapper.insertSelective(p);
}else if(p.getId() != null && p.getId().longValue() >0){
kafkaStorageHistoryMapper.updateByPrimaryKeySelective(p);
}
});
}
}
package com.fedex.connect.task.repository.repo.impl;
import com.fedex.connect.common.model.sys.KafkaTemporaryStorage;
import com.fedex.connect.task.repository.base.AbstractDaoRepository;
import com.fedex.connect.task.repository.repo.IKafkaTemporaryStorageRepository;
import org.springframework.stereotype.Repository;
import java.util.Date;
import java.util.List;
/**
* kafka临时表dao
* mt
* 2022年11月3日15:30:53
*/
@Repository
public class KafkaTemporaryStorageRepositoryImpl extends AbstractDaoRepository implements IKafkaTemporaryStorageRepository {
@Override
public List<KafkaTemporaryStorage> findKafkaTemporaryStoragePending() {
Date curDate = new Date();
return kafkaTemporaryStorageMapperExt.findKafkaTemporaryStoragePending(curDate);
}
@Override
public int deleteInBatch(List<KafkaTemporaryStorage> list) {
return kafkaTemporaryStorageMapperExt.deleteInBatch(list);
}
@Override
public List<KafkaTemporaryStorage> saveAll(List<KafkaTemporaryStorage> list) {
list.stream().forEach(p->{
if(p.getId() == null || p.getId().longValue() <=0){
kafkaTemporaryStorageMapper.insertSelective(p);
}else if(p.getId() != null && p.getId().longValue() >0){
kafkaTemporaryStorageMapper.updateByPrimaryKeySelective(p);
}
});
return list;
}
}
\ No newline at end of file
package com.fedex.connect.task.service.base;
import com.fedex.connect.task.repository.repo.IKafkaStorageHistoryRepository;
import com.fedex.connect.task.repository.repo.IKafkaTemporaryStorageRepository;
import org.springframework.beans.factory.annotation.Autowired;
/**
* @Author mt
* @Description 用于管理公共Service
* @Date 2024/4/18
*/
public class BaseService {
@Autowired
protected IKafkaTemporaryStorageRepository kafKaTemporaryStorageRepository;
@Autowired
protected IKafkaStorageHistoryRepository kafkaStorageHistoryRepository;
}
\ No newline at end of file
package com.fedex.connect.task.service.biz;
import com.fedex.connect.common.model.biz.Consignment;
public interface IPushObService {
/**
* @Author mt
* @Description 保存pushLog表记录
* @Date 2024/8/2
* @param consignment
* @return void
*/
void savePushLog(Consignment consignment);
/**
* @Author mt
* @Description 推送Imp001报文数据zip包文件
* @Date 2024/8/2
* @param
* @return void
*/
void sendOb();
/**
* @Author mt
* @Description 推送Imp001文件错误重试
* @Date 2024/8/2
* @param
* @return void
*/
void sendObRetry();
}
\ No newline at end of file
package com.fedex.connect.task.service.biz.impl;
import com.fedex.connect.common.model.biz.Consignment;
import com.fedex.connect.common.model.biz.PushOb;
import com.fedex.connect.task.service.base.BaseService;
import com.fedex.connect.task.service.biz.IPushObService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import java.util.List;
/**
* @Author mt
* @Description 发送Imp001到进口组
* @Date 2024/5/24
*/
@Slf4j
@Service
public class PushObServiceImpl extends BaseService implements IPushObService {
public void savePushLog(Consignment consignment){
}
/**
* 推送OB报文数据zip包文件
*/
@Override
public void sendOb() {
}
/**
* 推送OB文件错误重试
*/
@Override
public void sendObRetry() {
}
/**
* @Author mt
* @Description 推送tw001数据
* @Date 2024/5/29
* @param pushObList
* @return void
*/
private void pushOb(List<PushOb> pushObList){
}
}
\ No newline at end of file
package com.fedex.connect.task.service.sys;
public interface IKafkaDmService {
/**
* 查询临时表中创建时间为一小时之前，处理次数小于3次数据
* @return
*/
void scanKafkaTemporaryStorage();
}
\ No newline at end of file
package com.fedex.connect.task.service.sys.impl;
import com.fedex.connect.common.model.sys.KafkaTemporaryStorage;
import com.fedex.connect.task.service.base.BaseService;
import com.fedex.connect.task.service.sys.IKafkaDmService;
import com.fedex.connect.task.utils.sys.KafkaDmUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import java.util.List;
/**
* @Author mt
* @Description kafka操作service
* @Date 2024/4/18
*/
@Service
public class KafkaDmServiceImpl extends BaseService implements IKafkaDmService {
@Autowired
KafkaDmUtil kafkaDmUtil;
/**
* @Author mt
* @Description 查询临时表中创建时间为一小时之前，处理次数小于3次数据
* @Date 2024/5/14
* @param
* @return void
*/
public void scanKafkaTemporaryStorage(){
List<KafkaTemporaryStorage> kafKaTemporaryStorageList = kafKaTemporaryStorageRepository.findKafkaTemporaryStoragePending();
//kafka数据下发至kafka模块
kafkaDmUtil.kafKaTemporaryStorageRetry(kafKaTemporaryStorageList);
}
}
\ No newline at end of file
//package com.fedex.connect.task.utils.biz.imp001;
//
//import com.fedex.export.common.util.*;
//import com.fedex.export.config.PropertiesConfig;
//import com.fedex.export.constants.Constant;
//import com.fedex.export.data.bo.Imp001FileInfoBo;
//import com.fedex.export.data.bo.Imp001GenerateBo;
//import com.fedex.export.repository.entity.DeclareInfo;
//import com.fedex.export.repository.entity.SysDictionary;
//import org.slf4j.Logger;
//import org.slf4j.LoggerFactory;
//import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.stereotype.Component;
//
//import java.io.File;
//import java.io.FileOutputStream;
//import java.util.ArrayList;
//import java.util.Date;
//import java.util.List;
//import java.util.Objects;
//
///**
// * @Author mt
// * @Description tw001报文推送，util
// * @Date 2024/5/29
// */
//@Component
//public class Tw001PushUtil {
// private static Logger log = LoggerFactory.getLogger(Tw001PushUtil.class);
// @Autowired
// PropertiesConfig propertiesConfig;
// @Autowired
// ZipUtils zipUtils;
// @Autowired
// NioFileUtils nioFileUtils;
//
// /**
// * @Author mt
// * @Description 推送tw001
// * @Date 2024/5/29
// * @param tw001GenerateBo
// * @return com.fedex.export.data.dto.tw001.Imp001RootJsonDto
// */
// public void pushTw001(Imp001GenerateBo tw001GenerateBo) throws Exception{
// long start = System.currentTimeMillis();
// //生成zip临时文件到指定临时目录
// String zipFileName = this.toZips(tw001GenerateBo);
// //zip原始临时目录，zip为临时文件
// String zipSourceTempFilePath = tw001GenerateBo.getWorkFilePath() + zipFileName + Constant.FILE_SUFFIX_KEYS.TEMP;
// //推送zip文件最终目录
// String zipTargetFilePath = propertiesConfig.getTw001PathFinal() + zipFileName;
// //记录zip文件名称
// tw001GenerateBo.setZipFileName(zipFileName);
// //记录zip文件目标路径
// tw001GenerateBo.setZipFileTargetPath(zipTargetFilePath);
// //如果目录不存在则创建完整目录
// nioFileUtils.mkdirs(propertiesConfig.getTw001PathFinal());
// //推送zip临时文件到指定目标目录，复制文件到指定目录，采用
// String zipTargetTempFilePath = zipTargetFilePath + Constant.FILE_SUFFIX_KEYS.TEMP;
// //复制文件至目标目录，为临时文件
// nioFileUtils.copyFile(zipSourceTempFilePath,zipTargetTempFilePath);
// log.info("复制到目标目录完成 sourceTempFilePath ： {} , targetTempFilePath : {} ",zipSourceTempFilePath,zipTargetTempFilePath);
// //重命名为正式文件
// nioFileUtils.moveFile(zipTargetTempFilePath,zipTargetFilePath);
// log.info("重命名为正式文件成功 zipTargetTempFilePath ： {} , zipTargetFilePath : {} ",zipTargetTempFilePath,zipTargetFilePath);
// long end = System.currentTimeMillis();
// log.info("运单号TW001：{} 推送完成，总计耗时：{} ms",tw001GenerateBo.getDeclareInfo().getDeliveryNo() ,(end - start));
// }
//
// /**
// * @Author mt
// * @Description 功能说明 生成zip临时文件到指定临时目录
// * @Date 2024/5/30
// * @param tw001GenerateBo
// * @return java.lang.String
// */
// private String toZips(Imp001GenerateBo tw001GenerateBo) throws Exception{
// /******************初始化需要压缩文件列表****************/
// List<File> filesList = new ArrayList<File>();
// //生成json文件名称
// String jsonFileName = this.generateJsonFileName(tw001GenerateBo,"", Constant.FILE_SUFFIX_KEYS.JSON);
// //生成json文件
// JsonToJava.createJsonFile(JsonToJava.toJson(tw001GenerateBo.getTw001RootJsonDto()),
// tw001GenerateBo.getWorkFilePath(), jsonFileName);
// //获取json完整路径
// String jsonFilePath = tw001GenerateBo.getWorkFilePath() + jsonFileName;
// filesList.add(new File(jsonFilePath));
// //文件名称
// Utils.listOf(tw001GenerateBo.getFileInfoList()).stream().filter(Objects::nonNull).forEach(fileInfo ->{
// //文件如果存在则添加到文件集合中
// File file = new File(tw001GenerateBo.getWorkFilePath() + fileInfo.getFileName());
// if(file.exists()){
// filesList.add(file);
// }
// });
// //初始化json文件对象，并添加到文件集合中
// this.initJsonFileInfo(tw001GenerateBo,jsonFileName);
// //生成zip文件名称 .zip
// String zipFileName = this.generateJsonFileName(tw001GenerateBo, Constant.SEND_TW001_KEYS.TW001_HEAD_KEYS.MESSAGE_CODE, Constant.FILE_SUFFIX_KEYS.ZIP);
// //生成zip临时文件名称 .temp
// String zipTempFileName = zipFileName + Constant.FILE_SUFFIX_KEYS.TEMP;
// FileOutputStream fos = null;
// try{
// //完整zip临时文件名称
// File zipTemp = new File(tw001GenerateBo.getWorkFilePath() + zipTempFileName);
// fos = new FileOutputStream(zipTemp);
// zipUtils.toZip(filesList, fos);
// }catch(Exception ex){
// log.error("生成压缩包错误：{}",ex.getMessage(),ex);
// throw ex;
// }finally{
// try {
// if (fos != null) {
// fos.close();
// fos = null;
// }
// } catch (Exception e) {
// log.error(e.getMessage(),e);
// throw e;
// }
// }
// return zipFileName;
// }
//
// /**
// * @Author mt
// * @Description 功能说明 初始化json文件对象
// * @Date 2024/5/30
// * @param tw001GenerateBo
// * @param jsonFileName
// * @return void
// */
// private void initJsonFileInfo(Imp001GenerateBo tw001GenerateBo,String jsonFileName){
// SysDictionary jsonDic = tw001GenerateBo.getJsonDictionary();
// Imp001FileInfoBo fileInfoBo = new Imp001FileInfoBo();
// fileInfoBo.setFileName(jsonFileName);
// fileInfoBo.setFileTypeId(jsonDic.getId());
// fileInfoBo.setFileTypeName(jsonDic.getChineseName());
// fileInfoBo.setPushTime(new Date());
// tw001GenerateBo.getFileInfoList().add(fileInfoBo);
// }
//
// /**
// * @Author mt
// * @Description 功能说明 生成文件名称
// * @Date 2024/5/30
// * @param tw001GenerateBo
// * @param prefix 前缀
// * @param suffix 后缀
// * @return java.lang.String
// */
// private String generateJsonFileName(Imp001GenerateBo tw001GenerateBo,String prefix,String suffix){
// DeclareInfo declareInfo = tw001GenerateBo.getDeclareInfo();
// String fileName = prefix + declareInfo.getDeliveryNo() + Constant.COMMON_KEYS.UNDER_LINE
// + StringExtUtil.idLeftPadStr(declareInfo.getDeclareId()) + Constant.COMMON_KEYS.UNDER_LINE +
// DateUtil.yyyyMMddHH24mmssSSS(new Date()) + suffix;
// return fileName;
// }
//}
\ No newline at end of file
//package com.fedex.connect.task.utils.biz.imp001;
//
//import com.fedex.export.common.enums.PushLogEnum;
//import com.fedex.export.common.util.DateUtil;
//import com.fedex.export.common.util.NioFileUtils;
//import com.fedex.export.common.util.Utils;
//import com.fedex.export.config.PropertiesConfig;
//import com.fedex.export.constants.Constant;
//import com.fedex.export.data.bo.Imp001GenerateBo;
//import com.fedex.export.repository.entity.DeclareInfo;
//import com.fedex.export.repository.entity.log.PushObDetail;
//import com.fedex.export.repository.entity.log.PushObLog;
//import com.fedex.export.repository.repo.business.IPushObLogRepository;
//import com.fedex.export.service.business.IPushObDetailService;
//import org.slf4j.Logger;
//import org.slf4j.LoggerFactory;
//import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.stereotype.Component;
//
//import java.util.ArrayList;
//import java.util.Date;
//import java.util.List;
//import java.util.Objects;
//
///**
// * @Author mt
// * @Description 类说明 推送tw001完成，记录日志，并且备份数据
// * @Date 2024/5/30
// */
//@Component
//public class Tw001RecordLogUtil {
// private static Logger log = LoggerFactory.getLogger(Tw001RecordLogUtil.class);
// @Autowired
// PropertiesConfig propertiesConfig;
// @Autowired
// IPushObLogRepository pushObLogRepository;
// @Autowired
// IPushObDetailService pushObDetailService;
// @Autowired
// NioFileUtils nioFileUtils;
//
// /**
// * @Author mt
// * @Description 功能说明 推送成功记录日志
// * @Date 2024/5/30
// * @param tw001GenerateBo
// * @return void
// */
// public void pushSuccessRecordLog(Imp001GenerateBo tw001GenerateBo, PushObLog obLog){
// try {
// if(Objects.nonNull(tw001GenerateBo)){
// //备份文件，并且记录日志
// String targetPath = this.recordLog(tw001GenerateBo,obLog , propertiesConfig.getTw001PathBak(),PushLogEnum.PUSHED_SUCCESS,null);
// obLog.setFileName(tw001GenerateBo.getZipFileName());
// obLog.setFilePath(tw001GenerateBo.getZipFileTargetPath());
// obLog.setFileBackPath(targetPath);
// }
// obLog.setPushNum(obLog.getPushNum() + 1);
// obLog.setPushStatus(PushLogEnum.PUSHED_SUCCESS.getKey());
// obLog.setPushStatusDescribe(PushLogEnum.PUSHED_SUCCESS.getValue());
// obLog.setPushTime(new Date());
// obLog.setRemark("");
// //更新推送日志主表
// pushObLogRepository.updateById(obLog);
// }catch(Exception ex){
// log.error("pushSuccessRecordLog : {} " ,ex.getMessage(),ex);
// }
// }
//
// /**
// * @Author mt
// * @Description 功能说明 推送失败记录日志
// * @Date 2024/5/30
// * @param tw001GenerateBo
// * @param errorMsg
// * @return void
// */
// public void pushFailRecordLog(Imp001GenerateBo tw001GenerateBo, PushObLog obLog, String errorMsg){
// try {
// if(Objects.nonNull(tw001GenerateBo)){
// //备份文件，并且记录失败日志
// String targetPath = this.recordLog(tw001GenerateBo,obLog,propertiesConfig.getTw001PathError(),PushLogEnum.PUSH_FAILED,errorMsg);
// obLog.setFileName(tw001GenerateBo.getZipFileName());
// obLog.setFilePath(tw001GenerateBo.getZipFileTargetPath());
// obLog.setFileBackPath(targetPath);
// }
// obLog.setPushNum(obLog.getPushNum() + 1);
// obLog.setPushStatus(PushLogEnum.PUSH_FAILED.getKey());
// obLog.setPushStatusDescribe(PushLogEnum.PUSH_FAILED.getValue());
// obLog.setPushTime(new Date());
// obLog.setRemark(errorMsg);
// //更新推送日志主表
// pushObLogRepository.updateById(obLog);
// }catch(Exception ex){
// log.error("pushFailRecordLog : {} " ,ex.getMessage(),ex);
// }
// }
//
// /**
// * @Author mt
// * @Description 功能说明 备份文件
// * @Date 2024/5/30
// * @param tw001GenerateBo
// * @param tarPath
// * @return 返回目标备份路径
// */
// private String recordLog(Imp001GenerateBo tw001GenerateBo,
// PushObLog obLog,
// String tarPath,
// PushLogEnum pushLogEnum,
// String remark){
// String returnTargetPath = "";
// try{
// DeclareInfo declareInfo = tw001GenerateBo.getDeclareInfo();
// //文件原始路径
// String sourcePath = tw001GenerateBo.getWorkFilePath();
// //生成文件目标路径
// String targetPath = this.generateFilePath(declareInfo.getDeliveryNo(),declareInfo.getDeclareId(),tarPath);
// returnTargetPath = targetPath;
// //推送文件明细
// List<PushObDetail> pushObDetailList = new ArrayList<>();
// Utils.listOf(tw001GenerateBo.getFileInfoList()).stream().filter(Objects::nonNull).forEach(o->{
// try {
// //源文件完整路径
// String sPath = sourcePath + o.getFileName();
// //目标文件完整路径
// String tPath = targetPath + o.getFileName();
// //备份文件
// nioFileUtils.moveFile(sPath, tPath);
// //记录日志明细表
// PushObDetail pushObDetail = new PushObDetail();
// pushObDetail.setCreateTime(new Date());
// pushObDetail.setDeclareId(declareInfo.getDeclareId());
// pushObDetail.setDeliveryNo(declareInfo.getDeliveryNo());
// pushObDetail.setPushLogId(obLog.getId());
// pushObDetail.setPushStatus(pushLogEnum.getKey());
// pushObDetail.setPushStatusDescribe(pushLogEnum.getValue());
// pushObDetail.setFileTypeId(o.getFileTypeId());
// pushObDetail.setFileType(o.getFileTypeName());
// pushObDetail.setFileName(o.getFileName());
// pushObDetail.setPushTime(new Date());
// pushObDetail.setRemark(remark);
// pushObDetailList.add(pushObDetail);
// }catch(Exception ex){
// log.error("move file error : {} ",ex.getMessage(),ex);
// }
// });
// //批量插入推送日志明细
// pushObDetailService.batchInsert(pushObDetailList);
// log.info("batchInsert size : {} ",pushObDetailList.size());
// //zip包源文件完整路径
// String sPath = sourcePath + tw001GenerateBo.getZipFileName() + Constant.FILE_SUFFIX_KEYS.TEMP;
// //zip包目标文件完整路径
// String tPath = targetPath + tw001GenerateBo.getZipFileName();
// //备份zip包文件
// nioFileUtils.moveFile(sPath, tPath);
// //删除临时目录文件夹
// nioFileUtils.deleteDir(log,sourcePath,true);
// log.info("删除临时目录文件夹 sourcePath : {} ",sourcePath);
// }catch(Exception ex){
// log.error("recordLog error : {} ",ex.getMessage(),ex);
// }
// return returnTargetPath;
// }
//
// /**
// * @Author mt
// * @Description 功能说明 生成目标文件夹路径
// * @Date 2024/5/30
// * @param filePath
// * @return java.lang.String
// */
// private String generateFilePath(String deliveryNo,Long declareId,String filePath){
// String yyyyMMdd = DateUtil.format(new Date());
// String fileDir = filePath + yyyyMMdd + Constant.COMMON_KEYS.BACK_SLASH +
// deliveryNo + Constant.COMMON_KEYS.UNDER_LINE + declareId + Constant.COMMON_KEYS.BACK_SLASH;
// //如果目录不存在则创建完整目录
// nioFileUtils.mkdirs(fileDir);
// return fileDir;
// }
//}
package com.fedex.connect.task.utils.sys;
import com.alibaba.fastjson.JSONObject;
import com.fedex.connect.common.dependencies.util.JsonToJava;
import com.fedex.connect.common.dependencies.util.RPCUtils;
import com.fedex.connect.common.model.sys.KafkaStorageHistory;
import com.fedex.connect.common.model.sys.KafkaTemporaryStorage;
import com.fedex.connect.task.config.PropertiesConfig;
import com.fedex.connect.task.repository.repo.IKafkaStorageHistoryRepository;
import com.fedex.connect.task.repository.repo.IKafkaTemporaryStorageRepository;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import java.util.*;
/**
* @Author mt
* @Description 定时任务处理kafka异常数据补偿
* @Date 2024/5/14
*/
@Component
public class KafkaDmUtil {
@Autowired
protected IKafkaTemporaryStorageRepository kafKaTemporaryStorageRepository;
@Autowired
protected IKafkaStorageHistoryRepository kafkaStorageHistoryRepository;
@Autowired
RPCUtils rpcUtils;
@Autowired
PropertiesConfig propertiesConfig;
public void kafKaTemporaryStorageRetry(List<KafkaTemporaryStorage> kafKaTemporaryStorageList){
//待更新状态并下发消息集合
List<KafkaTemporaryStorage> issuedList = new ArrayList<KafkaTemporaryStorage>();
//待删除临时表，并备份历史表消息集合
List<KafkaTemporaryStorage> backHisList = new ArrayList<KafkaTemporaryStorage>();
kafKaTemporaryStorageList.forEach(kts ->{
//判断是否已经处理3次
if(kts.getFrequency().intValue() >= 3){
//标注状态为2失败
kts.setStatus(2L);
//备注重新处理原因
kts.setRemark((kts.getRemark() == null ? "" : kts.getRemark()) + "--->处理次数达到3次进行备份。");
backHisList.add(kts);
}else{
//处理次数加一
kts.setFrequency(kts.getFrequency() + 1L);
//备注重新处理原因
kts.setRemark((kts.getRemark() == null ? "" : kts.getRemark()) + "--->定时扫描任务重新下发");
issuedList.add(kts);
}
});
/***已处理3次数据不再进行下发，直接删除临时表记录，添加历史表记录***/
if(backHisList.size() > 0){
kafKaTemporaryStorageRepository.deleteInBatch(backHisList);
//kafka临时表转历史表，并且将临时表id置空
List<KafkaStorageHistory> kafKaStorageHistoryList = this.convertTempToHis(backHisList);
//保存至历史表
kafkaStorageHistoryRepository.saveAll(kafKaStorageHistoryList);
}
/***更新临时表状态，并实体数据转换为json字符串进行数据下发***/
if(issuedList.size() > 0) {
kafKaTemporaryStorageRepository.saveAll(issuedList);
Map<String,List<String>> messageMap = new HashMap<String,List<String>>();
issuedList.forEach(kts -> {
String message = JSONObject.toJSONString(kts);
//数据封装至map
this.pushMessageMap(kts.getMessageCode(),message,messageMap);
});
String json = JsonToJava.toJson(messageMap);
Random random = new Random();
String url = "";
//kafka模块未做LB，通过随机数控制调用随机其中一台
if(random.nextBoolean()){
url = this.propertiesConfig.getRpcKafkaReceive145();
}else{
url = this.propertiesConfig.getRpcKafkaReceive146();
}
//调用kafka rpc接口获取发送kafka结果
rpcUtils.rpcResponseInvoke(url, json, String.class);
}
}
/**
* 将数据封装至map
* @param messageCode
* @param message
* @param messageMap
*/
private void pushMessageMap(String messageCode,String message,Map<String,List<String>> messageMap){
//将数据进行分类
if(messageMap.containsKey(messageCode)){
List<String> messageList = messageMap.get(messageCode);
messageList.add(message);
messageMap.put(messageCode,messageList);
}else{
List<String> messageList = new ArrayList<String>();
messageList.add(message);
messageMap.put(messageCode,messageList);
}
}
/**
* 移出kafka临时表id值，并将临时表对象转换为kafka历史表对象
* @param kafKaTemporaryStorageList
*/
private static List<KafkaStorageHistory> convertTempToHis(List<KafkaTemporaryStorage> kafKaTemporaryStorageList){
List<KafkaStorageHistory> kafKaStorageHistoryList = new ArrayList<KafkaStorageHistory>();
kafKaTemporaryStorageList.forEach(kts ->{
String json = JSONObject.toJSONString(kts);
KafkaStorageHistory his = JSONObject.parseObject(json, KafkaStorageHistory.class);
his.setId(null);
kafKaStorageHistoryList.add(his);
});
return kafKaStorageHistoryList;
}
}
\ No newline at end of file
......@@ -2,65 +2,27 @@ spring:
#JDBC连接信息
datasource:
url: jdbc:oracle:thin:@192.168.1.250:1521:orcl
username: icleartw
password: icleartw
username: iclearimp
password: iclearimp
driver-class-name: oracle.jdbc.OracleDriver
profiles:
#系统环境
env: dev
allocation:
nodeId: 1
export:
mail:
smtp:
host: smtp.qiye.aliyun.com
port: 25
needauth: true
proxy:
startFlag: false
host: sin-proxy.apac.fedex.com
port: 3128
mailAccountFlag: true
# propertyPath:
# redis: /opt/fedex/exporttw/redis/export-redis.properties
language: zh_TW
task:
allocation:
#定时重发送申报信息邮件
sendDeclareInfoMail: 0 0/5 * * * ?
#定时获取邮件发送结果回执（读取html界面）
readHtmlReceipt: 0 0/1 * * * ?
#定时任务处理邮件回执超时
receiptTimout: 0 0/3 * * * ?
#同步HTML结果到Declare
syncHtmlToDeclare: 0 0/1 * * * ?
#CE晚来数据补偿发送邮件
ceCompensate: 0 0/1 * * * ?
#EDD邮件或者清关组邮件，每一小时轮巡监控超过10票失败，发送预警邮件
warningEmail: 0 0/1 * * * ?
#进口预清关补偿发送邮件（发送失败的邮件补偿发送）
ceLogCompensate: 0 0/1 * * * ?
#kafka补偿定时任务
kafkaRetry: 0 0/1 * * * ?
#推送ob zip包文件任务
sendOb: 0 0/1 * * * ?
#推送ob文件异常重试任务
sendObRetry: 0 0/1 * * * ?
#定时清理文件夹
clearFolders: 0 0/1 * * * ?
url:
proxy_hostname: sin-proxy.apac.fedex.com
mailHtmlReceipt: http://47.103.140.98:8088/IcTw/Index_of_report_silkConvListRpt.html
baseUrl: http://192.168.1.251/IcTw/
fedex: fedex.png
line: line.png
rpc:
url:
rpcKafkaReceive145: http://localhost:8088/icleartwkafka/rpcKafka/rpcKafkaReceive
rpcKafkaReceive146: http://localhost:8088/icleartwkafka/rpcKafka/rpcKafkaReceive
rpcKafkaReceive1: http://localhost:8082/kafka-cndc-server/rpcKafka/rpcKafkaReceive
rpcKafkaReceive2: http://localhost:8082/kafka-cndc-server/rpcKafka/rpcKafkaReceive
tw001:
path:
......
spring:
datasource:
jndi-name: jdbc/icleartwDS
jndi-name: jdbc/iclearConnectDS
profiles:
#系统环境
env: prod
allocation:
nodeId: 1
export:
mail:
smtp:
host: mapper.gslb.fedex.com
port: 25
needauth: true
proxy:
startFlag: true
host: sg2-proxy.apac.fedex.com
port: 3128
mailAccountFlag: false
# propertyPath:
# redis: /opt/fedex/exporttw/redis/export-redis.properties
language: zh_TW
task:
allocation:
#定时重发送申报信息邮件
sendDeclareInfoMail: 0 0/5 * * * ?
#定时获取邮件发送结果回执（读取html界面）
readHtmlReceipt: 0 0/3 * * * ?
#定时任务处理邮件回执超时
receiptTimout: 0 0/3 * * * ?
#同步HTML结果到Declare
syncHtmlToDeclare: 0 0/1 * * * ?
#CE晚来数据补偿发送邮件
ceCompensate: 0 0/5 * * * ?
#EDD邮件或者清关组邮件，每一小时轮巡监控超过10票失败，发送预警邮件
warningEmail: 0 0 0/1 * * ?
#进口预清关补偿发送邮件（发送失败的邮件补偿发送）
ceLogCompensate: 0 0/10 * * * ?
#kafka补偿定时任务
kafkaRetry: 0 0/1 * * * ?
#推送ob zip包文件任务
sendOb: 0 0/1 * * * ?
#推送ob zip包文件异常重试任务
sendObRetry: 0 0/3 * * * ?
#定时清理文件夹
clearFolders: 0 0 01 * * ?
url:
proxy_hostname: sin-proxy.apac.fedex.com
mailHtmlReceipt: http://155.161.252.18/report/silkConvListRpt/
baseUrl: https://exportdeclaration-tw.apac.fedex.com/IcTw/
fedex: fedex.png
line: line.png
rpc:
url:
rpcKafkaReceive145: https://pjea0179.prod.apac.fedex.com:9002/icleartwkafka/rpcKafka/rpcKafkaReceive
rpcKafkaReceive146: https://pjea0180.prod.apac.fedex.com:9002/icleartwkafka/rpcKafka/rpcKafkaReceive
rpcKafkaReceive1: https://pjea0179.prod.apac.fedex.com:9002/icleartwkafka/rpcKafka/rpcKafkaReceive
rpcKafkaReceive2: https://pjea0180.prod.apac.fedex.com:9002/icleartwkafka/rpcKafka/rpcKafkaReceive
tw001:
path:
......
spring:
datasource:
jndi-name: jdbc/icleartwDS
jndi-name: jdbc/iclearConnectDS
profiles:
#系统环境
env: test
allocation:
nodeId: 1
export:
mail:
smtp:
host: smtp.qiye.aliyun.com
port: 25
needauth: true
proxy:
startFlag: false
host: sin-proxy.apac.fedex.com
port: 3128
mailAccountFlag: true
# propertyPath:
# redis: /opt/fedex/exporttw/redis/export-redis.properties
language: zh_TW
task:
allocation:
#定时重发送申报信息邮件
sendDeclareInfoMail: 0 0/5 * * * ?
#定时获取邮件发送结果回执（读取html界面）
readHtmlReceipt: 0 0/3 * * * ?
#定时任务处理邮件回执超时
receiptTimout: 0 0/3 * * * ?
#同步HTML结果到Declare
syncHtmlToDeclare: 0 0/1 * * * ?
#CE晚来数据补偿发送邮件
ceCompensate: 0 0/1 * * * ?
#EDD邮件或者清关组邮件，每一小时轮巡监控超过10票失败，发送预警邮件
warningEmail: 0 0/1 * * * ?
#进口预清关补偿发送邮件（发送失败的邮件补偿发送）
ceLogCompensate: 0 0/1 * * * ?
#kafka补偿定时任务
kafkaRetry: 0 0/1 * * * ?
#推送ob zip包文件任务
sendOb: 0 0/1 * * * ?
#推送ob zip包文件异常重试任务
sendObRetry: 0 0/1 * * * ?
#定时清理文件夹
clearFolders: 0 0/1 * * * ?
url:
proxy_hostname: sin-proxy.apac.fedex.com
mailHtmlReceipt: http://47.103.140.98:8088/IcTw/Index_of_report_silkConvListRpt.html
baseUrl: http://192.168.1.251/IcTw/
fedex: fedex.png
line: line.png
rpc:
url:
rpcKafkaReceive145: http://47.103.140.98:7010/icleartwkafka/rpcKafka/rpcKafkaReceive
rpcKafkaReceive146: http://47.103.140.98:7010/icleartwkafka/rpcKafka/rpcKafkaReceive
rpcKafkaReceive1: http://47.103.140.98:7010/icleartwkafka/rpcKafka/rpcKafkaReceive
rpcKafkaReceive2: http://47.103.140.98:7010/icleartwkafka/rpcKafka/rpcKafkaReceive
tw001:
path:
......
spring:
datasource:
jndi-name: jdbc/icleartwDS
jndi-name: jdbc/iclearConnectDS
profiles:
#系统环境
env: uat
allocation:
nodeId: 1
export:
mail:
smtp:
host: mapper.gslb.fedex.com
port: 25
needauth: true
proxy:
startFlag: true
host: sg2-proxy.apac.fedex.com
port: 3128
mailAccountFlag: false
# propertyPath:
# redis: /opt/fedex/exporttw/redis/export-redis.properties
language: zh_TW
task:
allocation:
#定时重发送申报信息邮件
sendDeclareInfoMail: 0 0/5 * * * ?
#定时获取邮件发送结果回执（读取html界面）
readHtmlReceipt: 0 0/3 * * * ?
#定时任务处理邮件回执超时
receiptTimout: 0 0/3 * * * ?
#同步HTML结果到Declare
syncHtmlToDeclare: 0 0/1 * * * ?
#CE晚来数据补偿发送邮件
ceCompensate: 0 0/5 * * * ?
#EDD邮件或者清关组邮件，每一小时轮巡监控超过10票失败，发送预警邮件
warningEmail: 0 0 0/1 * * ?
#进口预清关补偿发送邮件（发送失败的邮件补偿发送）
ceLogCompensate: 0 0/10 * * * ?
#kafka补偿定时任务
kafkaRetry: 0 0/1 * * * ?
#推送ob zip包文件任务
sendOb: 0 0/1 * * * ?
#推送ob zip包文件异常重试任务
sendObRetry: 0 0/3 * * * ?
#定时清理文件夹
clearFolders: 0 0 01 * * ?
url:
proxy_hostname: sin-proxy.apac.fedex.com
mailHtmlReceipt: http://155.161.252.18/report/silkConvListRpt/
baseUrl: https://exportdeclarationuat-tw.apac.fedex.com/IcTw/
fedex: fedex.png
line: line.png
rpc:
url:
rpcKafkaReceive145: http://ujea0145.nonprod.apac.fedex.com:9001/icleartwkafka/rpcKafka/rpcKafkaReceive
rpcKafkaReceive146: http://ujea0146.nonprod.apac.fedex.com:9001/icleartwkafka/rpcKafka/rpcKafkaReceive
rpcKafkaReceive1: http://ujea0145.nonprod.apac.fedex.com:9001/icleartwkafka/rpcKafka/rpcKafkaReceive
rpcKafkaReceive2: http://ujea0146.nonprod.apac.fedex.com:9001/icleartwkafka/rpcKafka/rpcKafkaReceive
tw001:
path:
......
spring.redis.host=223.166.86.220
spring.redis.port=6333
spring.redis.password=!QAZ2wsx
spring.redis.timeout=3000
spring.redis.jedis.pool.max-active=50
spring.redis.jedis.pool.max-wait=-1
spring.redis.jedis.pool.max-idle=4
spring.redis.jedis.pool.min-idle=0
\ No newline at end of file