首次提交代码

src/main/java/com/gzzn/omms/msgexchangeapi/MsgexchangeApiApplication.java

@@ -0,0 +1,14 @@
+package com.gzzn.omms.msgexchangeapi;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.scheduling.annotation.EnableScheduling;
+
+@SpringBootApplication
+@EnableScheduling
+public class MsgexchangeApiApplication {
+
+	public static void main(String[] args) {
+		SpringApplication.run(MsgexchangeApiApplication.class, args);
+	}
+}

src/main/java/com/gzzn/omms/msgexchangeapi/config/KafkaProducerConfig.java

@@ -0,0 +1,52 @@
+package com.gzzn.omms.msgexchangeapi.config;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.annotation.EnableKafka;
+import org.springframework.kafka.core.DefaultKafkaProducerFactory;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.core.ProducerFactory;
+
+@Configuration
+@EnableKafka
+public class KafkaProducerConfig {
+
+    @Value("${kafka.producer.servers}")
+    private String servers;
+    @Value("${kafka.producer.retries}")
+    private int retries;
+    @Value("${kafka.producer.batch.size}")
+    private int batchSize;
+    @Value("${kafka.producer.linger}")
+    private int linger;
+    @Value("${kafka.producer.buffer.memory}")
+    private int bufferMemory;
+
+
+    public Map<String, Object> producerConfigs() {
+        Map<String, Object> props = new HashMap<>();
+        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, servers);
+        props.put(ProducerConfig.RETRIES_CONFIG, retries);
+        props.put(ProducerConfig.BATCH_SIZE_CONFIG, batchSize);
+        props.put(ProducerConfig.LINGER_MS_CONFIG, linger);
+        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, bufferMemory);
+        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+        return props;
+    }
+
+    public ProducerFactory<String, String> producerFactory() {
+        return new DefaultKafkaProducerFactory<>(producerConfigs());
+    }
+
+    @Bean
+    public KafkaTemplate<String, String> kafkaTemplate() {
+        return new KafkaTemplate<String, String>(producerFactory());
+    }
+}

src/main/java/com/gzzn/omms/msgexchangeapi/controller/KafkaController.java

@@ -0,0 +1,22 @@
+package com.gzzn.omms.msgexchangeapi.controller;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+import com.gzzn.omms.msgexchangeapi.dto.ResponseDto;
+import com.gzzn.omms.msgexchangeapi.service.IKafkaService;
+
+@RestController
+@RequestMapping("/kafka")
+public class KafkaController {
+	    @Autowired
+	    private IKafkaService kafkaservice;
+	
+	    @GetMapping(value = "/send")
+	    public ResponseDto sendKafka(String msg,String topic) {
+	    	    kafkaservice.msgSend(topic, msg);
+	            return ResponseDto.success();
+	    }
+}

src/main/java/com/gzzn/omms/msgexchangeapi/dao/CminmsgDao.java

@@ -0,0 +1,34 @@
+package com.gzzn.omms.msgexchangeapi.dao;
+
+import java.util.List;
+
+import org.springframework.data.domain.Pageable;
+import org.springframework.data.jpa.repository.Query;
+import org.springframework.data.repository.CrudRepository;
+
+import com.gzzn.omms.msgexchangeapi.entiy.Cminmsg;
+
+
+public interface CminmsgDao  extends CrudRepository<Cminmsg, Long> {
+
+	/**
+	 * 
+	 * @return
+	 */
+	@Query("select count(*) from Cminmsg c where cminmsgsDateProcessed is not null")
+	public Long getCminmsgsDateProcessedIsNotNullCount();
+	
+	
+   /**
+    * 根据处理时间字段查询列表,Null情况
+    * @param date
+    * @return
+    */
+   public List<Cminmsg> findByCminmsgsDateProcessedIsNull(Pageable pageable);
+   
+   /**
+    * 非空情况
+    * @return
+    */
+   public List<Cminmsg> findByCminmsgsDateProcessedIsNotNull(Pageable pageable);
+}

src/main/java/com/gzzn/omms/msgexchangeapi/dto/ResponseDto.java

@@ -0,0 +1,97 @@
+package com.gzzn.omms.msgexchangeapi.dto;
+
+import com.gzzn.omms.msgexchangeapi.enums.ResultCode;
+
+/**
+ * 
+ * @author xiunai
+ *
+ * @date 2018年5月16日
+ * 
+ * 
+ */
+public class ResponseDto<T> {
+	
+	private Boolean is_success; 
+	
+	private Integer err_code;
+	
+	private String err_msg;
+	
+	private T body;
+	
+	public ResponseDto() {}
+
+    public ResponseDto(Integer code, String msg) {
+    	this.setIs_success(false);
+        this.setErr_code(code);
+        this.setErr_msg(msg);
+    }
+
+    public static ResponseDto success() {
+    	ResponseDto result = new ResponseDto();
+    	result.setIs_success(true);
+        result.setResultCode(ResultCode.SUCCESS);
+        return result;
+    }
+
+    public static <T> ResponseDto<T> success(T data) {
+    	ResponseDto result = new ResponseDto<T>();
+    	result.setIs_success(true);
+        result.setResultCode(ResultCode.SUCCESS);
+        result.setBody(data);
+        return result;
+    }
+
+    public static ResponseDto failure(ResultCode resultCode) {
+    	ResponseDto result = new ResponseDto();
+    	result.setIs_success(false);
+        result.setResultCode(resultCode);
+        return result;
+    }
+
+    public static <T> ResponseDto<T> failure(ResultCode resultCode, T data) {
+    	ResponseDto result = new ResponseDto<T>();
+    	result.setIs_success(false);
+        result.setResultCode(resultCode);
+        result.setBody(data);
+        return result;
+    }
+
+    public void setResultCode(ResultCode code) {
+        this.setErr_code(code.code());
+        this.setErr_msg(code.message());
+    }
+
+	public T getBody() {
+		return body;
+	}
+
+	public void setBody(T body) {
+		this.body = body;
+	}
+
+	public Boolean getIs_success() {
+		return is_success;
+	}
+
+	public void setIs_success(Boolean is_success) {
+		this.is_success = is_success;
+	}
+
+	public Integer getErr_code() {
+		return err_code;
+	}
+
+	public void setErr_code(Integer err_code) {
+		this.err_code = err_code;
+	}
+
+	public String getErr_msg() {
+		return err_msg;
+	}
+
+	public void setErr_msg(String err_msg) {
+		this.err_msg = err_msg;
+	}
+}

src/main/java/com/gzzn/omms/msgexchangeapi/entiy/Cminmsg.java

@@ -0,0 +1,136 @@
+package com.gzzn.omms.msgexchangeapi.entiy;
+
+import java.io.Serializable;
+import javax.persistence.*;
+import java.util.Date;
+
+
+/**
+ * The persistent class for the CMINMSGS database table.
+ * 
+ */
+@Entity
+@Table(name="CMINMSGS")
+@NamedQuery(name="Cminmsg.findAll", query="SELECT c FROM Cminmsg c")
+public class Cminmsg implements Serializable {
+	private static final long serialVersionUID = 1L;
+
+	@Id
+	@Column(name="CMINMSGS_ID")
+	private long cminmsgsId;
+
+	@Lob
+	@Column(name="CMINMSGS_CLOB_MSG")
+	private String cminmsgsClobMsg;
+
+	@Temporal(TemporalType.DATE)
+	@Column(name="CMINMSGS_DATE_PROCESSED")
+	private Date cminmsgsDateProcessed;
+
+	@Temporal(TemporalType.DATE)
+	@Column(name="CMINMSGS_DATE_RECEIVED")
+	private Date cminmsgsDateReceived;
+
+	@Column(name="CMINMSGS_STATUS")
+	private String cminmsgsStatus;
+
+	@Temporal(TemporalType.DATE)
+	@Column(name="CMINMSGS_SUBSYSTEM_DATE_SENT")
+	private Date cminmsgsSubsystemDateSent;
+
+	@Column(name="CMINMSGS_SUBSYSTEM_NAME")
+	private String cminmsgsSubsystemName;
+
+	@Column(name="CMINMSGS_SUBSYSTEM_SEQUENCE")
+	private String cminmsgsSubsystemSequence;
+
+	@Column(name="CMINMSGS_SUBSYSTEM_SUBTYPE")
+	private String cminmsgsSubsystemSubtype;
+
+	@Column(name="CMINMSGS_SUBSYSTEM_TYPE")
+	private String cminmsgsSubsystemType;
+
+	public Cminmsg() {
+	}
+
+	public long getCminmsgsId() {
+		return this.cminmsgsId;
+	}
+
+	public void setCminmsgsId(long cminmsgsId) {
+		this.cminmsgsId = cminmsgsId;
+	}
+
+	public String getCminmsgsClobMsg() {
+		return this.cminmsgsClobMsg;
+	}
+
+	public void setCminmsgsClobMsg(String cminmsgsClobMsg) {
+		this.cminmsgsClobMsg = cminmsgsClobMsg;
+	}
+
+	public Date getCminmsgsDateProcessed() {
+		return this.cminmsgsDateProcessed;
+	}
+
+	public void setCminmsgsDateProcessed(Date cminmsgsDateProcessed) {
+		this.cminmsgsDateProcessed = cminmsgsDateProcessed;
+	}
+
+	public Date getCminmsgsDateReceived() {
+		return this.cminmsgsDateReceived;
+	}
+
+	public void setCminmsgsDateReceived(Date cminmsgsDateReceived) {
+		this.cminmsgsDateReceived = cminmsgsDateReceived;
+	}
+
+	public String getCminmsgsStatus() {
+		return this.cminmsgsStatus;
+	}
+
+	public void setCminmsgsStatus(String cminmsgsStatus) {
+		this.cminmsgsStatus = cminmsgsStatus;
+	}
+
+	public Date getCminmsgsSubsystemDateSent() {
+		return this.cminmsgsSubsystemDateSent;
+	}
+
+	public void setCminmsgsSubsystemDateSent(Date cminmsgsSubsystemDateSent) {
+		this.cminmsgsSubsystemDateSent = cminmsgsSubsystemDateSent;
+	}
+
+	public String getCminmsgsSubsystemName() {
+		return this.cminmsgsSubsystemName;
+	}
+
+	public void setCminmsgsSubsystemName(String cminmsgsSubsystemName) {
+		this.cminmsgsSubsystemName = cminmsgsSubsystemName;
+	}
+
+	public String getCminmsgsSubsystemSequence() {
+		return this.cminmsgsSubsystemSequence;
+	}
+
+	public void setCminmsgsSubsystemSequence(String cminmsgsSubsystemSequence) {
+		this.cminmsgsSubsystemSequence = cminmsgsSubsystemSequence;
+	}
+
+	public String getCminmsgsSubsystemSubtype() {
+		return this.cminmsgsSubsystemSubtype;
+	}
+
+	public void setCminmsgsSubsystemSubtype(String cminmsgsSubsystemSubtype) {
+		this.cminmsgsSubsystemSubtype = cminmsgsSubsystemSubtype;
+	}
+
+	public String getCminmsgsSubsystemType() {
+		return this.cminmsgsSubsystemType;
+	}
+
+	public void setCminmsgsSubsystemType(String cminmsgsSubsystemType) {
+		this.cminmsgsSubsystemType = cminmsgsSubsystemType;
+	}
+
+}

src/main/java/com/gzzn/omms/msgexchangeapi/enums/ResultCode.java

@@ -0,0 +1,83 @@
+package com.gzzn.omms.msgexchangeapi.enums;
+
+public enum ResultCode {
+	 /* 成功状态码 */
+    SUCCESS(1, "成功"),
+    Failure(0,"失败"),
+    
+    /* 参数错误：10001-19999 */
+    PARAM_IS_INVALID(10001, "参数无效"),
+    PARAM_IS_BLANK(10002, "参数为空"),
+    PARAM_TYPE_BIND_ERROR(10003, "参数类型错误"),
+    PARAM_NOT_COMPLETE(10004, "参数缺失"),
+
+    /* 用户错误：20001-29999*/
+    USER_NOT_LOGGED_IN(20001, "用户未登录"),
+    USER_LOGIN_ERROR(20002, "账号不存在或密码错误"),
+    USER_ACCOUNT_FORBIDDEN(20003, "账号已被禁用"),
+    USER_NOT_EXIST(20004, "用户不存在"),
+    USER_HAS_EXISTED(20005, "用户已存在"),
+
+    /* 业务错误：30001-39999 */
+    SPECIFIED_QUESTIONED_USER_NOT_EXIST(30001, "某业务出现问题"),
+
+    /* 系统错误：40001-49999 */
+    SYSTEM_INNER_ERROR(40001, "系统繁忙，请稍后重试"),
+
+    /* 数据错误：50001-599999 */
+    RESULE_DATA_NONE(50001, "数据未找到"),
+    DATA_IS_WRONG(50002, "数据有误"),
+    DATA_ALREADY_EXISTED(50003, "数据已存在"),
+
+    /* 接口错误：60001-69999 */
+    INTERFACE_INNER_INVOKE_ERROR(60001, "内部系统接口调用异常"),
+    INTERFACE_OUTTER_INVOKE_ERROR(60002, "外部系统接口调用异常"),
+    INTERFACE_FORBID_VISIT(60003, "该接口禁止访问"),
+    INTERFACE_ADDRESS_INVALID(60004, "接口地址无效"),
+    INTERFACE_REQUEST_TIMEOUT(60005, "接口请求超时"),
+    INTERFACE_EXCEED_LOAD(60006, "接口负载过高"),
+    INTERFACE_NOT_IMPLEMENT(60007, "接口暂未实现"),
+    
+    /* 权限错误：70001-79999 */
+    PERMISSION_NO_ACCESS(70001, "无访问权限");
+	
+
+   private Integer code;
+   private String message;
+
+    ResultCode(Integer code, String message) {
+        this.code = code;
+        this.message = message;
+    }
+
+    public Integer code() {
+        return this.code;
+    }
+
+    public String message() {
+        return this.message;
+    }
+
+    public static String getMessage(String name) {
+        for (ResultCode item : ResultCode.values()) {
+            if (item.name().equals(name)) {
+                return item.message;
+            }
+        }
+        return name;
+    }
+
+    public static Integer getCode(String name) {
+        for (ResultCode item : ResultCode.values()) {
+            if (item.name().equals(name)) {
+                return item.code;
+            }
+        }
+        return null;
+    }
+
+    @Override
+    public String toString() {
+        return this.name();
+    }
+}

src/main/java/com/gzzn/omms/msgexchangeapi/service/ExchangeServiceImpl.java

@@ -0,0 +1,28 @@
+package com.gzzn.omms.msgexchangeapi.service;
+
+import java.io.IOException;
+import java.util.Map;
+
+import org.springframework.stereotype.Service;
+
+import com.fasterxml.jackson.core.JsonParseException;
+import com.fasterxml.jackson.databind.JsonMappingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.dataformat.xml.XmlMapper;
+
+@Service
+public class ExchangeServiceImpl implements IExchangeService {
+
+	@Override
+	public String xmlToJson(String xml) throws JsonParseException, JsonMappingException, IOException {
+		
+		ObjectMapper xmlMapper = new XmlMapper();
+		Map map = xmlMapper.readValue(xml, Map.class);
+		
+		ObjectMapper jsonMapper = new ObjectMapper();
+		String json = jsonMapper.writeValueAsString(map);
+		
+		return json;
+	}
+
+}

src/main/java/com/gzzn/omms/msgexchangeapi/service/IExchangeService.java

@@ -0,0 +1,16 @@
+package com.gzzn.omms.msgexchangeapi.service;
+
+import java.io.IOException;
+
+import com.fasterxml.jackson.core.JsonParseException;
+import com.fasterxml.jackson.databind.JsonMappingException;
+
+/**
+ * 消息转换服务
+ * @author Administrator
+ *
+ */
+public interface IExchangeService {
+
+	public String xmlToJson(String xml)throws JsonParseException, JsonMappingException, IOException;
+}

src/main/java/com/gzzn/omms/msgexchangeapi/service/IKafkaService.java

@@ -0,0 +1,5 @@
+package com.gzzn.omms.msgexchangeapi.service;
+
+public interface IKafkaService {
+    public void msgSend(String topic,String msg) ;
+}

src/main/java/com/gzzn/omms/msgexchangeapi/service/KafkaServiceImpl.java

@@ -0,0 +1,18 @@
+package com.gzzn.omms.msgexchangeapi.service;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.stereotype.Service;
+
+@Service
+public class KafkaServiceImpl implements IKafkaService{
+	 
+    @Autowired
+    private KafkaTemplate kafkaTemplate;
+
+    
+	@Override
+	public void msgSend(String topic, String msg) {
+		kafkaTemplate.send(topic, msg);
+	}
+}

src/main/java/com/gzzn/omms/msgexchangeapi/task/ExchangeTask.java

@@ -0,0 +1,76 @@
+package com.gzzn.omms.msgexchangeapi.task;
+
+import java.io.IOException;
+import java.util.List;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.data.domain.PageRequest;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+import org.springframework.util.StringUtils;
+
+import com.fasterxml.jackson.core.JsonParseException;
+import com.fasterxml.jackson.databind.JsonMappingException;
+import com.gzzn.omms.msgexchangeapi.dao.CminmsgDao;
+import com.gzzn.omms.msgexchangeapi.entiy.Cminmsg;
+import com.gzzn.omms.msgexchangeapi.service.IExchangeService;
+import com.gzzn.omms.msgexchangeapi.service.IKafkaService;
+
+/**
+ * 转换任务，获取cim的消息，转换成json，写入到kafka队列中
+ * @author Administrator
+ *
+ */
+@Component
+public class ExchangeTask {
+	private static Logger logger = LoggerFactory.getLogger(ExchangeTask.class);
+	
+	 @Autowired
+	 private IKafkaService kafkaservice;
+	 
+	 @Autowired
+	 IExchangeService exchangeService;
+	 
+	 @Autowired
+	 private CminmsgDao cminmsgDao;
+	  
+	@Scheduled(cron="0 0/1 * * * ?")
+	public void corn()
+	{
+		logger.info("定时任务启动....");
+		
+		Long maxPerLong = 1000L; //每次最大发送多少条消息
+		Long lenTotal = cminmsgDao.getCminmsgsDateProcessedIsNotNullCount();
+		Long lenSend = 0L;
+		Integer pageIndex = 0;
+		Integer pageSize = 50;
+		while(lenSend < lenTotal && lenSend < maxPerLong)
+		{
+			List<Cminmsg>  ls = cminmsgDao.findByCminmsgsDateProcessedIsNotNull(new PageRequest(pageIndex,pageSize));
+			ls.forEach(x -> {   
+					String strJson = null;
+					try {
+						strJson = exchangeService.xmlToJson(x.getCminmsgsClobMsg());
+						if(!StringUtils.isEmpty(strJson))
+						{
+							kafkaservice.msgSend("topic2", strJson);  
+						}
+					} catch (JsonParseException e) {
+						e.printStackTrace();
+					} catch (JsonMappingException e) {
+						e.printStackTrace();
+					} catch (IOException e) {
+						e.printStackTrace();
+					}
+				});
+
+			
+			lenSend += ls.size();
+			pageIndex++;
+		}
+		//
+		logger.info("同步完成");
+	}
+}