windy/msgexchange-api
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

1.新增pro生产环境配置。

Browse Source 
2.新增CMINMSGS和COUTMSGS 脚本
3.去除启动自动获取最新动态的功能。
4.完善测试以上功能。修改Sender为OSH5
This commit is contained in:
zhouxiunai
2018-12-29 17:30:43 +08:00
parent 440864110e
commit f799c64682
11 changed files with 443 additions and 264 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/com/gzzn/omms/msgexchangeapi/entity/Cminmsg.java
+1 -2
View File
@@ -11,12 +11,11 @@ import java.util.Date;
*/
@Entity
@Table(name="CMINMSGS")
@SequenceGenerator(name = "sequenceGenerator", sequenceName = "SEQ_CMINMSGS")
public class Cminmsg implements Serializable {
private static final long serialVersionUID = 1L;
@Id
@GeneratedValue(strategy=GenerationType.SEQUENCE,generator="sequenceGenerator")
@GeneratedValue(strategy=GenerationType.IDENTITY)
@Column(name="CMINMSGS_ID")
private long cminmsgsId;
src/main/java/com/gzzn/omms/msgexchangeapi/entity/Cmoutmsg.java
+119 -65
View File
@@ -1,136 +1,190 @@
package com.gzzn.omms.msgexchangeapi.entity;
import java.io.Serializable;
import javax.persistence.*;
import java.util.Date;
/**
* The persistent class for the CMINMSGS database table.
* The persistent class for the COUTMSGS database table.
*
*/
@Entity
@Table(name="CMOUTMSGS")
@Table(name="COUTMSGS")
public class Cmoutmsg implements Serializable {
private static final long serialVersionUID = 1L;
@Id
@GeneratedValue(strategy=GenerationType.SEQUENCE)
@Column(name="CMINMSGS_ID")
private long cminmsgsId;
@GeneratedValue(strategy=GenerationType.IDENTITY)
@Column(name="COUTMSGS_ID")
private int coutmsgsId;
@Temporal(TemporalType.TIMESTAMP)
@Column(name="COUTMSGS_ACK_DATE_RECV")
private Date coutmsgsAckDateRecv;
@Column(name="COUTMSGS_ACK_REQD")
private int coutmsgsAckReqd;
@Column(name="COUTMSGS_ACK_RESEND_TIMES")
private int coutmsgsAckResendTimes;
@Lob
@Column(name="CMINMSGS_CLOB_MSG")
private String cminmsgsClobMsg;
@Column(name="COUTMSGS_CLOB_MSG")
private String coutmsgsClobMsg;
@Temporal(TemporalType.TIMESTAMP)
@Column(name="CMINMSGS_DATE_PROCESSED")
private Date cminmsgsDateProcessed;
@Column(name="COUTMSGS_DATE_INSERTED")
private Date coutmsgsDateInserted;
@Temporal(TemporalType.TIMESTAMP)
@Column(name="CMINMSGS_DATE_RECEIVED")
private Date cminmsgsDateReceived;
@Column(name="COUTMSGS_DATE_SENT")
private Date coutmsgsDateSent;
@Column(name="CMINMSGS_STATUS")
private String cminmsgsStatus;
@Column(name="COUTMSGS_ENCRYPT")
private int coutmsgsEncrypt;
@Temporal(TemporalType.TIMESTAMP)
@Column(name="CMINMSGS_SUBSYSTEM_DATE_SENT")
private Date cminmsgsSubsystemDateSent;
@Column(name="COUTMSGS_ERROR")
private int coutmsgsError;
@Column(name="CMINMSGS_SUBSYSTEM_NAME")
private String cminmsgsSubsystemName;
@Column(name="COUTMSGS_FINAL_GROUP_IND")
private int coutmsgsFinalGroupInd;
@Column(name="CMINMSGS_SUBSYSTEM_SEQUENCE")
private String cminmsgsSubsystemSequence;
@Column(name="COUTMSGS_GROUP_ID")
private int coutmsgsGroupId;
@Column(name="CMINMSGS_SUBSYSTEM_SUBTYPE")
private String cminmsgsSubsystemSubtype;
@Column(name="COUTMSGS_GROUP_ORDER")
private int coutmsgsGroupOrder;
@Column(name="CMINMSGS_SUBSYSTEM_TYPE")
private String cminmsgsSubsystemType;
@Column(name="COUTMSGS_NO_MESSAGES")
private int coutmsgsNoMessages;
@Column(name="COUTMSGS_TRUEFALS_GROUP")
private int coutmsgsTruefalsGroup;
@Column(name="ROUTINGID")
private String routingid;
public Cmoutmsg() {
}
public long getCminmsgsId() {
return this.cminmsgsId;
public int getCoutmsgsId() {
return this.coutmsgsId;
}
public void setCminmsgsId(long cminmsgsId) {
this.cminmsgsId = cminmsgsId;
public void setCoutmsgsId(int coutmsgsId) {
this.coutmsgsId = coutmsgsId;
}
public String getCminmsgsClobMsg() {
return this.cminmsgsClobMsg;
public Date getCoutmsgsAckDateRecv() {
return this.coutmsgsAckDateRecv;
}
public void setCminmsgsClobMsg(String cminmsgsClobMsg) {
this.cminmsgsClobMsg = cminmsgsClobMsg;
public void setCoutmsgsAckDateRecv(Date coutmsgsAckDateRecv) {
this.coutmsgsAckDateRecv = coutmsgsAckDateRecv;
}
public Date getCminmsgsDateProcessed() {
return this.cminmsgsDateProcessed;
public int getCoutmsgsAckReqd() {
return this.coutmsgsAckReqd;
}
public void setCminmsgsDateProcessed(Date cminmsgsDateProcessed) {
this.cminmsgsDateProcessed = cminmsgsDateProcessed;
public void setCoutmsgsAckReqd(int coutmsgsAckReqd) {
this.coutmsgsAckReqd = coutmsgsAckReqd;
}
public Date getCminmsgsDateReceived() {
return this.cminmsgsDateReceived;
public int getCoutmsgsAckResendTimes() {
return this.coutmsgsAckResendTimes;
}
public void setCminmsgsDateReceived(Date cminmsgsDateReceived) {
this.cminmsgsDateReceived = cminmsgsDateReceived;
public void setCoutmsgsAckResendTimes(int coutmsgsAckResendTimes) {
this.coutmsgsAckResendTimes = coutmsgsAckResendTimes;
}
public String getCminmsgsStatus() {
return this.cminmsgsStatus;
public String getCoutmsgsClobMsg() {
return this.coutmsgsClobMsg;
}
public void setCminmsgsStatus(String cminmsgsStatus) {
this.cminmsgsStatus = cminmsgsStatus;
public void setCoutmsgsClobMsg(String coutmsgsClobMsg) {
this.coutmsgsClobMsg = coutmsgsClobMsg;
}
public Date getCminmsgsSubsystemDateSent() {
return this.cminmsgsSubsystemDateSent;
public Date getCoutmsgsDateInserted() {
return this.coutmsgsDateInserted;
}
public void setCminmsgsSubsystemDateSent(Date cminmsgsSubsystemDateSent) {
this.cminmsgsSubsystemDateSent = cminmsgsSubsystemDateSent;
public void setCoutmsgsDateInserted(Date coutmsgsDateInserted) {
this.coutmsgsDateInserted = coutmsgsDateInserted;
}
public String getCminmsgsSubsystemName() {
return this.cminmsgsSubsystemName;
public Date getCoutmsgsDateSent() {
return this.coutmsgsDateSent;
}
public void setCminmsgsSubsystemName(String cminmsgsSubsystemName) {
this.cminmsgsSubsystemName = cminmsgsSubsystemName;
public void setCoutmsgsDateSent(Date coutmsgsDateSent) {
this.coutmsgsDateSent = coutmsgsDateSent;
}
public String getCminmsgsSubsystemSequence() {
return this.cminmsgsSubsystemSequence;
public int getCoutmsgsEncrypt() {
return this.coutmsgsEncrypt;
}
public void setCminmsgsSubsystemSequence(String cminmsgsSubsystemSequence) {
this.cminmsgsSubsystemSequence = cminmsgsSubsystemSequence;
public void setCoutmsgsEncrypt(int coutmsgsEncrypt) {
this.coutmsgsEncrypt = coutmsgsEncrypt;
}
public String getCminmsgsSubsystemSubtype() {
return this.cminmsgsSubsystemSubtype;
public int getCoutmsgsError() {
return this.coutmsgsError;
}
public void setCminmsgsSubsystemSubtype(String cminmsgsSubsystemSubtype) {
this.cminmsgsSubsystemSubtype = cminmsgsSubsystemSubtype;
public void setCoutmsgsError(int coutmsgsError) {
this.coutmsgsError = coutmsgsError;
}
public String getCminmsgsSubsystemType() {
return this.cminmsgsSubsystemType;
public int getCoutmsgsFinalGroupInd() {
return this.coutmsgsFinalGroupInd;
}
public void setCminmsgsSubsystemType(String cminmsgsSubsystemType) {
this.cminmsgsSubsystemType = cminmsgsSubsystemType;
public void setCoutmsgsFinalGroupInd(int coutmsgsFinalGroupInd) {
this.coutmsgsFinalGroupInd = coutmsgsFinalGroupInd;
}
public int getCoutmsgsGroupId() {
return this.coutmsgsGroupId;
}
public void setCoutmsgsGroupId(int coutmsgsGroupId) {
this.coutmsgsGroupId = coutmsgsGroupId;
}
public int getCoutmsgsGroupOrder() {
return this.coutmsgsGroupOrder;
}
public void setCoutmsgsGroupOrder(int coutmsgsGroupOrder) {
this.coutmsgsGroupOrder = coutmsgsGroupOrder;
}
public int getCoutmsgsNoMessages() {
return this.coutmsgsNoMessages;
}
public void setCoutmsgsNoMessages(int coutmsgsNoMessages) {
this.coutmsgsNoMessages = coutmsgsNoMessages;
}
public int getCoutmsgsTruefalsGroup() {
return this.coutmsgsTruefalsGroup;
}
public void setCoutmsgsTruefalsGroup(int coutmsgsTruefalsGroup) {
this.coutmsgsTruefalsGroup = coutmsgsTruefalsGroup;
}
public String getRoutingid() {
return this.routingid;
}
public void setRoutingid(String routingid) {
this.routingid = routingid;
}
}
src/main/java/com/gzzn/omms/msgexchangeapi/entity/msg/SUBSYSTEMS.java
+1 -1
View File
@@ -64,7 +64,7 @@ import javax.xml.bind.annotation.XmlType;
public enum SUBSYSTEMS {
AODB,
OMMS,
OSH5,
BHS,
FIDS,
RMS,
src/main/java/com/gzzn/omms/msgexchangeapi/msghandler/flop/base/FlopBaseHandler.java
+1
View File
@@ -73,6 +73,7 @@ public class FlopBaseHandler implements IBaseHandler {
}
else {
//没有该动态航班信息
this.updateCminmsgToProcessed(cminmsg);
logger.warn("没有flid为:{}的消息",flop.getFLID());
return HandlerResult.failure();
}
src/main/java/com/gzzn/omms/msgexchangeapi/runner/AppRunner.java
+15 -17
View File
@@ -1,7 +1,8 @@
package com.gzzn.omms.msgexchangeapi.runner;
import java.util.Date;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -12,7 +13,6 @@ import org.springframework.stereotype.Component;
import com.gzzn.omms.msgexchangeapi.redis.RedisKeyConstant;
import com.gzzn.omms.msgexchangeapi.redis.RedisService;
import com.gzzn.omms.msgexchangeapi.service.ICmoutmsgService;
@Component
public class AppRunner implements CommandLineRunner {
@@ -22,28 +22,26 @@ public class AppRunner implements CommandLineRunner {
@Autowired
RedisService redisService;
@Autowired
private ICmoutmsgService cmoutmsgService;
@Value("${msgExchange.waitDateLeftOffset}")
private Long waitDateLeftOffset;
@Value("${msgExchange.intervalSeconds}")
private Long intervalSeconds;
@Override
public void run(String... args) throws Exception {
logger.info("发送获取动态航班日计划的消息");
Date nowDate = new Date();
Date waitDate = new Date(nowDate.getTime() - waitDateLeftOffset);//往后偏移指定时钟，避免因不同应用系统时钟相差过大导致的问题。
cmoutmsgService.sendSchdGetMsg(null,null);//发送获取当天日计划的申请
logger.info("app runner start");
//
redisService.set(RedisKeyConstant.KEY_ISWAITSCHD, true); //等待返回日计划状态
redisService.set(RedisKeyConstant.KEY_WAITSCHDDATE,waitDate); //初始化消息进度
logger.info("发送获取动态航班日计划的消息完成");
Executors.newFixedThreadPool(1).execute(new SchdWaitRunner());
redisService.set(RedisKeyConstant.KEY_LASTBEGINID,0L);
//启动消息采集线程
ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(1);
scheduledExecutorService.scheduleAtFixedRate(new MsgExchangeRunner(), 0, intervalSeconds , TimeUnit.SECONDS);
logger.info("app runner end");
}//end function run
}
src/main/java/com/gzzn/omms/msgexchangeapi/service/CmoutmsgServiceImpl.java
+7 -3
View File
@@ -1,5 +1,6 @@
package com.gzzn.omms.msgexchangeapi.service;
import java.math.BigInteger;
import java.util.Date;
import java.util.Locale;
@@ -28,13 +29,15 @@ public class CmoutmsgServiceImpl implements ICmoutmsgService {
public Boolean sendSchdGetMsg(Date startDate,Date endDate)
{
Long nowtime = System.currentTimeMillis();
//构造消息
MSG rqfdMsg = new MSGBuilder().
setMETA(
new METABuilder()
.setSNDR(SUBSYSTEMS.OMMS)
.setSNDR(SUBSYSTEMS.OSH5)
.setTYPE(MESSAGETYPES.RQFD)
.setSTYP(MESSAGESUBTYPES.NONE)
.setDTTM(new BigInteger(nowtime.toString()))
.build()
)
.build();
@@ -53,8 +56,9 @@ public class CmoutmsgServiceImpl implements ICmoutmsgService {
//发送消息到数据库
Cmoutmsg cmoutmsg = new Cmoutmsg();
cmoutmsg.setCminmsgsClobMsg(exchangeService.objectToXmlstr(rqfdMsg));
cmoutmsg.setCminmsgsStatus("VALID");
cmoutmsg.setCoutmsgsDateInserted(new Date());
cmoutmsg.setCoutmsgsClobMsg(exchangeService.objectToXmlstr(rqfdMsg));
cmoutmsg.setRoutingid(SUBSYSTEMS.OSH5.name());
Cmoutmsg rtCmoutmsg = cmoutmsgDao.save(cmoutmsg);
src/main/resources/application-pro.yml
+86
View File
@@ -0,0 +1,86 @@
spring:
jpa:
hibernate:
ddl-auto: none
naming:
physical-strategy: org.hibernate.boot.model.naming.PhysicalNamingStrategyStandardImpl
datasource:
username: root
password: gzzn09ap
url: jdbc:mysql://172.17.35.162:3306/cdairport?useSSL=false&useUnicode=true&characterEncoding=utf-8&allowMultiQueries=true&autoReconnect=true
tomcat:
max-active: 30
test-on-borrow: true
initial-size: 3
redis:
password:
database: 1
port: 6379
pool:
max-idle: 10
min-idle: 0
max-active: 200
max-wait: -1
host: 172.17.35.165
timeout: 1000
kafka:
producer:
retries: 0
servers: 172.17.35.161:32796,172.17.35.161:32797,172.17.35.161:32798
linger: 1
batch:
size: 4096
buffer:
memory: 10240000
max:
request:
size: 10240000
consumer:
auto:
offset:
reset: latest
commit:
interval: 100
servers: 172.17.35.161:32796,172.17.35.161:32797,172.17.35.161:32798
zookeeper:
connect: 172.17.35.161:2181
session:
timeout: 6000
enable:
auto:
commit: true
topic: test
concurrency: 10
group:
id: test
msgExchange:
intervalSeconds: 5
waitDateLeftOffset: 600000
logstash:
host: 172.17.35.160:5000
scheduled:
#每日凌晨3:30分点采集历史数据 0 30 3 * * ?
flightInfoCron: 0 30 3 * * ?
corePoolSize: 10
maxPoolSize: 50
queueCapacity: 10
#计入历史航班数据的条件
hstCondition:
#航班到达超过多久则成为历史航班数据（单位：秒）
ARRIVE_HST_TIME: 3600
#航班取消超过多久则成为历史航班数据（单位：秒）
CANCEL_HST_TIME: 3600
#备降超过多久则成为历史航班数据 FROM CTU 本应降落到成都备降到其他地方（单位：秒）
FDIV_HST_TIME: 59400
#计划时间超过多久则成为历史航班数据（单位：秒）
SODT_HST_TIME: 259200
# Elasticsearch
# 9200端口是用来让HTTP REST API来访问ElasticSearch，而9300端口是传输层监听的默认端口
elasticsearch:
ip: 172.17.35.160
port: 9300
pool: 5
#注意cluster.name需要与config/elasticsearch.yml中的cluster.name一致
cluster:
name: docker-cluster
src/main/resources/application.yml
+1 -1
View File
@@ -1,3 +1,3 @@
spring:
profiles:
active: dev
active: pro
src/main/resources/db/migration/CMINMSGS.sql
+13
View File
@@ -0,0 +1,13 @@
CREATE TABLE CMINMSGS
(
CMINMSGS_ID int(23) not null AUTO_INCREMENT PRIMARY KEY,
CMINMSGS_CLOB_MSG LONGTEXT not null,
CMINMSGS_DATE_RECEIVED datetime(0),
CMINMSGS_DATE_PROCESSED datetime(0),
CMINMSGS_SUBSYSTEM_NAME VARCHAR(4),
CMINMSGS_SUBSYSTEM_SEQUENCE int(6),
CMINMSGS_SUBSYSTEM_DATE_SENT datetime(0),
CMINMSGS_SUBSYSTEM_SUBTYPE VARCHAR(4),
CMINMSGS_SUBSYSTEM_TYPE VARCHAR(4),
CMINMSGS_STATUS VARCHAR(20)
);
src/main/resources/db/migration/COUTMSGS.sql
+18
View File
@@ -0,0 +1,18 @@
CREATE TABLE COUTMSGS
(
COUTMSGS_ID int(23) not null AUTO_INCREMENT PRIMARY KEY,
COUTMSGS_CLOB_MSG LONGTEXT,
COUTMSGS_DATE_INSERTED datetime(0) DEFAULT current_timestamp() not null,
COUTMSGS_DATE_SENT datetime(0),
ROUTINGID VARCHAR(23) not null,
COUTMSGS_TRUEFALS_GROUP int(1) default 0 not null,
COUTMSGS_NO_MESSAGES int(2),
COUTMSGS_GROUP_ORDER int(2),
COUTMSGS_GROUP_ID int(6),
COUTMSGS_FINAL_GROUP_IND int(1) default 0 not null,
COUTMSGS_ACK_REQD int(1) default 0 not null,
COUTMSGS_ACK_RESEND_TIMES int(4) default 0 not null,
COUTMSGS_ACK_DATE_RECV datetime(0),
COUTMSGS_ENCRYPT int(1) default 0 not null,
COUTMSGS_ERROR int(1) default 0 not null
);