jiangjiazhi / center

Download as
Browse Code »

Commit d5dba590cf457d5ad28c7a4c7753599ee36c6c4b

Authored by maliang
1 parent e71af13800
Exists in master

添加中心数据同步方法 

<p>该方法存在一个弊端

Showing 19 changed files with 879 additions and 25 deletions

center.manager/src/main/java/com/lyms/cm/controller/sys/SysRemoteController.java View file @ d5dba59
  1 +package com.lyms.cm.controller.sys;
  2 +
  3 +import java.io.IOException;
  4 +
  5 +import javax.servlet.http.HttpServletRequest;
  6 +
  7 +import org.springframework.stereotype.Controller;
  8 +import org.springframework.web.bind.annotation.RequestMapping;
  9 +import org.springframework.web.bind.annotation.ResponseBody;
  10 +
  11 +import com.lyms.cm.entity.sys.SysUsers;
  12 +import com.lyms.synch.entity.SyncnModel;
  13 +import com.lyms.web.controller.BaseController;
  14 +
  15 +@Controller
  16 +@RequestMapping("/remote")
  17 +public class SysRemoteController extends BaseController {
  18 +
  19 + @RequestMapping(value = "/push")
  20 + @ResponseBody
  21 + public String push(HttpServletRequest request) throws ClassNotFoundException, IOException {
  22 + SyncnModel model = conver(request);
  23 + SysUsers user = (SysUsers) model.getData();
  24 + System.out.println(user.getName() + " " + model.getMethod());
  25 + return "OK";
  26 + }
  27 +}
center.manager/src/main/resources/xml/app-shiro.xml View file @ d5dba59
... ... @@ -46,6 +46,7 @@
46 46 <property name="filterChainDefinitions">
47 47 <value>
48 48 /login = anon
  49 + /remote/** =anon
49 50 /logout = logout
50 51 /static/** = anon
51 52 /authenticated = authc
center.manager/src/main/resources/xml/app-syncn.xml View file @ d5dba59
  1 +<?xml version="1.0" encoding="UTF-8"?>
  2 +<beans xmlns="http://www.springframework.org/schema/beans"
  3 + xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  4 + xmlns:context="http://www.springframework.org/schema/context"
  5 + xmlns:tx="http://www.springframework.org/schema/tx"
  6 + xmlns:aop="http://www.springframework.org/schema/aop"
  7 + xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-4.2.xsd
  8 + http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-4.2.xsd
  9 + http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-4.2.xsd
  10 + http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-4.2.xsd">
  11 + <!-- 数据同步中心配置 -->
  12 + <bean id="center" class="com.lyms.synch.SyncnCenter" factory-method="install" >
  13 + <property name="queue" ref="queue" />
  14 + </bean>
  15 +
  16 + <!-- 使用的队列信息 -->
  17 + <bean id="queue" class="com.lyms.synch.queue.SyncnQueue">
  18 + <!-- 依赖redis template -->
  19 + <constructor-arg index="0" ref="redisTemplate"/>
  20 + </bean>
  21 +</beans>
center.manager/src/test/java/center/manager/test/user/WorkTest.java View file @ d5dba59
  1 +package center.manager.test.user;
  2 +
  3 +import org.junit.Test;
  4 +import org.springframework.beans.factory.annotation.Autowired;
  5 +
  6 +import com.lyms.cm.entity.sys.SysUsers;
  7 +import com.lyms.synch.CenterCallback;
  8 +import com.lyms.synch.SyncnCenter;
  9 +import com.lyms.synch.SyncnCenter.Work;
  10 +import com.lyms.synch.entity.ModelType;
  11 +import com.lyms.synch.entity.SyncnModel;
  12 +import com.lyms.util.HttpUtils;
  13 +import com.lyms.util.StrUtils;
  14 +
  15 +/**
  16 + * <li>@ClassName: WorkTest
  17 + * <li>@Description: 基于spring 注入方式创建 center
  18 + * <li>@author maliang
  19 + * <li>@date 2017年3月13日
  20 + * <li>
  21 + */
  22 +public class WorkTest extends BaseTest {
  23 +
  24 + @Autowired
  25 + public SyncnCenter center;
  26 +
  27 + @Test
  28 + public void workTest() {
  29 + SyncnModel model = new SyncnModel();
  30 + model.setId(StrUtils.uuid());
  31 + model.setModel("user");
  32 + model.setMethod("post");
  33 + model.setType(ModelType.ADD.name());
  34 + // 推送数据远程地址
  35 + model.setRemote("http://127.0.0.1:9090/center.manager/remote/push");
  36 + // 对于远程获取数据，回调地址
  37 + model.setCallBack("http://www.baidu.com");
  38 + model.setTs(System.currentTimeMillis());
  39 +
  40 + SysUsers user = new SysUsers();
  41 + user.setName("乱码测试");
  42 + model.setData(user);
  43 +
  44 + // 添加任务信息,临时任务信息
  45 + center.buildWork().push(model, new CenterCallback() {
  46 + @Override
  47 + public void callBack(Object object) {
  48 + System.out.println("push callBack : " + object);
  49 + }
  50 + });
  51 +
  52 + // 获取临时任务信息
  53 + // 包含ack 回调方法
  54 + Work work = center.buildWork();
  55 + try {
  56 + work.pullTmp(new CenterCallback() {
  57 + @Override
  58 + public void callBack(Object object) {
  59 + // 处理实际业务
  60 + SyncnModel model = (SyncnModel) object;
  61 + String result = HttpUtils.REMOTE.post(model);
  62 + System.out.println(result);
  63 + }
  64 + });
  65 + } catch (Exception e) {
  66 + e.printStackTrace();
  67 + // 业务逻辑错误，反压数据，将会输重写写入队列
  68 + work.backPressure();
  69 + }
  70 + // .ackGet(null);
  71 + // System.out.println("callBack: " + get);
  72 +
  73 + }
  74 +
  75 + public static void main(String[] args) {
  76 + SyncnModel model = new SyncnModel();
  77 + model.setId(StrUtils.uuid());
  78 + model.setModel("user");
  79 + model.setMethod("post");
  80 + model.setType(ModelType.ADD.name());
  81 + // 推送数据远程地址
  82 + model.setRemote("http://127.0.0.1:9090/center.manager/remote/push");
  83 + // 对于远程获取数据，回调地址
  84 + model.setCallBack("http://www.baidu.com");
  85 + model.setTs(System.currentTimeMillis());
  86 +
  87 + SysUsers user = new SysUsers();
  88 + user.setName("你好");
  89 + model.setData(user);
  90 + String result = HttpUtils.REMOTE.post(model);
  91 + System.out.println(result);
  92 + }
  93 +}
core.sdk/src/main/java/com/lyms/spring/redis/operation/RedisLock.java View file @ d5dba59
... ... @@ -41,6 +41,7 @@
41 41 Long nano = System.nanoTime();
42 42 while (System.nanoTime() - nano < TimeUnit.MILLISECONDS.toNanos(waitmill)) {
43 43 boolean tag = acquireLock(lockName);
  44 + // System.out.println("获取锁.....");
44 45 // 获取到锁资源
45 46 if (tag)
46 47 return true;
core.sdk/src/main/java/com/lyms/synch/CenterCallback.java View file @ d5dba59
  1 +package com.lyms.synch;
  2 +
  3 +public interface CenterCallback {
  4 + public void callBack(Object object);
  5 +}
core.sdk/src/main/java/com/lyms/synch/LinkedWork.java View file @ d5dba59
  1 +package com.lyms.synch;
  2 +
  3 +/**
  4 + * <li>@ClassName: LinkedWork
  5 + * <li>@Description: 任务链表
  6 + * <p>
  7 + * 主要解决多任务循环执行的过程
  8 + * <li>@author maliang
  9 + * <li>@date 2017年3月13日
  10 + * <li>
  11 + */
  12 +public class LinkedWork {
  13 +
  14 +}
core.sdk/src/main/java/com/lyms/synch/SyncnCenter.java View file @ d5dba59
  1 +package com.lyms.synch;
  2 +
  3 +import java.util.Map;
  4 +import java.util.concurrent.Executors;
  5 +import java.util.concurrent.ScheduledExecutorService;
  6 +
  7 +import org.apache.commons.lang3.StringUtils;
  8 +
  9 +import com.lyms.synch.entity.ModelType;
  10 +import com.lyms.synch.entity.SyncnModel;
  11 +import com.lyms.synch.queue.SyncnQueue;
  12 +import com.lyms.util.HttpUtils;
  13 +
  14 +/**
  15 + * <li>@ClassName: SyncnCenter
  16 + * <li>@Description: 数据同步中心,使用单例模式创建一个中心,中心依赖redis队列
  17 + * <li>@author maliang
  18 + * <li>@date 2017年3月13日
  19 + * <li>
  20 + */
  21 +public class SyncnCenter {
  22 +
  23 + public static ScheduledExecutorService scheduledPool = Executors.newScheduledThreadPool(5);
  24 +
  25 + public SyncnCenter() {
  26 + }
  27 +
  28 + private static class inner {
  29 + private static final SyncnCenter center = new SyncnCenter();
  30 + }
  31 +
  32 + /**
  33 + * 数据缓冲队列
  34 + */
  35 + private SyncnQueue queue;
  36 +
  37 + /**
  38 + * <li>@Description:创建一个中心实力
  39 + * <li>@return
  40 + * <li>创建人：maliang
  41 + * <li>创建时间：2017年3月13日
  42 + * <li>修改人：
  43 + * <li>修改时间：
  44 + */
  45 + public static final SyncnCenter install() {
  46 + return SyncnCenter.inner.center;
  47 + }
  48 +
  49 + /**
  50 + * <li>@Description:构建一个工作器
  51 + * <li>@return
  52 + * <li>创建人：maliang
  53 + * <li>创建时间：2017年3月13日
  54 + * <li>修改人：
  55 + * <li>修改时间：
  56 + */
  57 + public Work buildWork() {
  58 + return new Work();
  59 + }
  60 +
  61 + public class Work {
  62 +
  63 + public SyncnModel model;
  64 +
  65 + // public CenterCallback callBack;
  66 +
  67 + public Work push(SyncnModel model, CenterCallback callBack) {
  68 + this.model = model;
  69 + Boolean tag = queue.push(model);
  70 + callBack.callBack(tag);
  71 + return this;
  72 + }
  73 +
  74 + public Work pullFix(CenterCallback callBack) {
  75 + SyncnModel model = queue.pull(ModelType.GET);
  76 + this.model = model;
  77 + callBack.callBack(model);
  78 + return this;
  79 + }
  80 +
  81 + public Work pullTmp(CenterCallback callBack) {
  82 + SyncnModel model = queue.pull(null);
  83 + this.model = model;
  84 + callBack.callBack(model);
  85 + return this;
  86 + }
  87 +
  88 + /**
  89 + * <li>@Description:数据反压
  90 + * <p>
  91 + * 针对临时任务队列，如果业务逻辑处理失败，手动反压该数据，下次重新执行
  92 + * <li>
  93 + * <li>创建人：maliang
  94 + * <li>创建时间：2017年3月13日
  95 + * <li>修改人：
  96 + * <li>修改时间：
  97 + */
  98 + public void backPressure() {
  99 + this.push(model, new CenterCallback() {
  100 + boolean tag = false;
  101 +
  102 + @Override
  103 + public void callBack(Object object) {
  104 + tag = (Boolean) object;
  105 +
  106 + if (!tag) {
  107 + backPressure();
  108 + }
  109 + }
  110 + });
  111 + }
  112 +
  113 + /**
  114 + * <li>@Description:该方法为回执方法，将会调用目标接口，返回提示对应消息处理成功
  115 + * <p>
  116 + * 使用方式 pull(type,callBack).ackGet();
  117 + * <p>
  118 + * 目前该方法对PULL 的消息回执才有实际的意义
  119 + * <li>@return
  120 + * <li>创建人：maliang
  121 + * <li>创建时间：2017年3月13日
  122 + * <li>修改人：
  123 + * <li>修改时间：
  124 + */
  125 + public String ackGet(Map<String, String> params) {
  126 + if (model != null && StringUtils.isNotBlank(model.getCallBack())) {
  127 + return HttpUtils.REMOTE.get(model.getCallBack(), params);
  128 + }
  129 + return null;
  130 + }
  131 +
  132 + public String ackPost(Map<String, String> params) {
  133 + if (model != null && StringUtils.isNotBlank(model.getCallBack())) {
  134 + return HttpUtils.REMOTE.post(model.getCallBack(), params);
  135 + }
  136 + return null;
  137 + }
  138 +
  139 + /**
  140 + * <li>@Description:该方法主要检测是否重复执行，对应数据
  141 + * <p>
  142 + * TODO
  143 + * <p>
  144 + * 使用方式 push(model,callBack).validata();
  145 + * <li>@return
  146 + * <li>创建人：maliang
  147 + * <li>创建时间：2017年3月13日
  148 + * <li>修改人：
  149 + * <li>修改时间：
  150 + */
  151 + public Work validata() {
  152 + return this;
  153 + }
  154 +
  155 + }
  156 +
  157 + public SyncnQueue getQueue() {
  158 + return queue;
  159 + }
  160 +
  161 + public void setQueue(SyncnQueue queue) {
  162 + this.queue = queue;
  163 + }
  164 +
  165 +}
core.sdk/src/main/java/com/lyms/synch/entity/Model.java View file @ d5dba59
  1 +package com.lyms.synch.entity;
  2 +
  3 +/**
  4 + * <li>@ClassName: Model
  5 + * <li>@Description: 模块信息
  6 + * <li>@author maliang
  7 + * <li>@date 2017年3月13日
  8 + * <li>
  9 + */
  10 +public enum Model {
  11 + USER("用户信息");
  12 +
  13 + private String des;
  14 +
  15 + private Model(String des) {
  16 + this.des = des;
  17 + }
  18 +
  19 + public String getDes() {
  20 + return this.des;
  21 + }
  22 +}
core.sdk/src/main/java/com/lyms/synch/entity/ModelType.java View file @ d5dba59
  1 +package com.lyms.synch.entity;
  2 +
  3 +import java.util.EnumSet;
  4 +
  5 +/**
  6 + * <li>@ClassName: ModelType
  7 + * <li>@Description: 数据模型类型
  8 + * <li>@author maliang
  9 + * <li>@date 2017年3月13日
  10 + * <li>
  11 + */
  12 +public enum ModelType {
  13 +
  14 + ADD, DEL, UPDATE, GET, ACK;
  15 +
  16 + public static ModelType exitsType(String typeName) {
  17 + for (ModelType modelType : EnumSet.allOf(ModelType.class)) {
  18 + if (modelType.name().equalsIgnoreCase(typeName)) {
  19 + return modelType;
  20 + }
  21 + }
  22 + return null;
  23 + }
  24 +
  25 + public static boolean isAdd(String type) {
  26 + return ADD.equals(exitsType(type));
  27 + }
  28 +
  29 + public static boolean isACK(String type) {
  30 + return ACK.equals(exitsType(type));
  31 + }
  32 +
  33 + public static boolean isDel(String type) {
  34 + return DEL.equals(exitsType(type));
  35 + }
  36 +
  37 + public static boolean isUpdate(String type) {
  38 + return UPDATE.equals(exitsType(type));
  39 + }
  40 +
  41 + public static boolean isGet(String type) {
  42 + return GET.equals(exitsType(type));
  43 + }
  44 +
  45 + public static boolean isGet(ModelType type) {
  46 + return GET.equals(type);
  47 + }
  48 +
  49 + public static void main(String[] args) {
  50 + System.out.println(ModelType.isAdd("add"));
  51 + }
  52 +
  53 +}
core.sdk/src/main/java/com/lyms/synch/entity/SyncnModel.java View file @ d5dba59
  1 +package com.lyms.synch.entity;
  2 +
  3 +import java.io.Serializable;
  4 +
  5 +import com.lyms.util.JsonUtils;
  6 +
  7 +/**
  8 + * <li>@ClassName: SynceModel
  9 + * <li>@Description: 数据模型
  10 + * <li>@author maliang
  11 + * <li>@date 2017年3月13日
  12 + * <li>
  13 + */
  14 +public class SyncnModel implements Serializable {
  15 +
  16 + /**
  17 + */
  18 + private static final long serialVersionUID = -5121448640215880284L;
  19 +
  20 + /**
  21 + * 消息模型ID
  22 + */
  23 + private String id;
  24 +
  25 + /**
  26 + * 远程地址,远程访问的服务地址
  27 + */
  28 + private String remote;
  29 +
  30 + /**
  31 + * 回调地址,获取数据处理的时候该参数才有实际意义
  32 + */
  33 + private String callBack;
  34 +
  35 + /**
  36 + * get | post
  37 + */
  38 + private String method;
  39 +
  40 + /**
  41 + * 操作类型
  42 + */
  43 + private String type;
  44 +
  45 + /**
  46 + * 模块
  47 + */
  48 + private String model;
  49 +
  50 + /**
  51 + * 数据信息
  52 + */
  53 + private Object data;
  54 +
  55 + /**
  56 + * 操作时间
  57 + */
  58 + private Long ts;
  59 +
  60 + public String getId() {
  61 + return id;
  62 + }
  63 +
  64 + public void setId(String id) {
  65 + this.id = id;
  66 + }
  67 +
  68 + public String getType() {
  69 + return type;
  70 + }
  71 +
  72 + public void setType(String type) {
  73 + this.type = type;
  74 + }
  75 +
  76 + public String getModel() {
  77 + return model;
  78 + }
  79 +
  80 + public void setModel(String model) {
  81 + this.model = model;
  82 + }
  83 +
  84 + public Object getData() {
  85 + return data;
  86 + }
  87 +
  88 + public void setData(Object data) {
  89 + this.data = data;
  90 + }
  91 +
  92 + public Long getTs() {
  93 + return ts;
  94 + }
  95 +
  96 + public void setTs(Long ts) {
  97 + this.ts = ts;
  98 + }
  99 +
  100 + public String getRemote() {
  101 + return remote;
  102 + }
  103 +
  104 + public void setRemote(String remote) {
  105 + this.remote = remote;
  106 + }
  107 +
  108 + public String getCallBack() {
  109 + return callBack;
  110 + }
  111 +
  112 + public void setCallBack(String callBack) {
  113 + this.callBack = callBack;
  114 + }
  115 +
  116 + public String getMethod() {
  117 + return method;
  118 + }
  119 +
  120 + public void setMethod(String method) {
  121 + this.method = method;
  122 + }
  123 +
  124 + public static SyncnModel conver(String body) {
  125 + return JsonUtils.jsonToBean(body, SyncnModel.class);
  126 + }
  127 +
  128 +}
core.sdk/src/main/java/com/lyms/synch/entity/package-info.java View file @ d5dba59
  1 +/**
  2 + * <li>@ClassName: package-info
  3 + * <li>@Description: 数据模型
  4 + * <li>@author maliang
  5 + * <li>@date 2017年3月13日
  6 + * <li>
  7 + */
  8 +package com.lyms.synch.entity;
core.sdk/src/main/java/com/lyms/synch/package-info.java View file @ d5dba59
  1 +/**
  2 + * <li>@ClassName: package-info
  3 + * <li>@Description: 数据同步中心
  4 + * <li>@author maliang
  5 + * <li>@date 2017年3月13日
  6 + * <li>
  7 + */
  8 +package com.lyms.synch;
  9 +
  10 +// SyncnCenter.push(object);
  11 +// SyncnCenter.pull(object).ack();
  12 +// SyncnCenter.ack(object);
core.sdk/src/main/java/com/lyms/synch/queue/SyncnQueue.java View file @ d5dba59
  1 +package com.lyms.synch.queue;
  2 +
  3 +import java.io.Serializable;
  4 +
  5 +import org.springframework.data.redis.core.RedisTemplate;
  6 +
  7 +import com.lyms.spring.redis.operation.RedisLock;
  8 +import com.lyms.synch.entity.ModelType;
  9 +import com.lyms.synch.entity.SyncnModel;
  10 +
  11 +/**
  12 + * <li>@ClassName: SyncnQueue
  13 + * <li>@Description: 数据队列
  14 + * <li>@author maliang
  15 + * <li>@date 2017年3月13日
  16 + * <li>
  17 + */
  18 +public class SyncnQueue {
  19 +
  20 + public RedisTemplate<Serializable, Serializable> template;
  21 +
  22 + /**
  23 + * 构建临时任务队列，固定任务队列锁
  24 + * <p>
  25 + * 区分锁，避免高并发下锁释放的错误
  26 + */
  27 + public static RedisLock Fix_Lock;
  28 +
  29 + public static RedisLock Tmp_Lock;
  30 +
  31 + /**
  32 + * 针对数据获取的时候添加锁
  33 + */
  34 + private static final String LOCK_TMP_NAME = "TMP_lock";
  35 +
  36 + private static final String LOCK_FIXATION_NAME = "FIXATION_lock";
  37 +
  38 + /**
  39 + * 等待获取锁的时间10 S
  40 + */
  41 + private static final Long WAIT_LOCK_TIME = 10000L;
  42 +
  43 + /**
  44 + * 固定任务
  45 + */
  46 + private static final String FIXATION_WORK = "FIXATION_WORK";
  47 +
  48 + /**
  49 + * 临时任务
  50 + */
  51 + private static final String TMP_WORK = "TMP_WORK";
  52 +
  53 + /**
  54 + * 回执消息任务
  55 + */
  56 + @Deprecated
  57 + private static final String ACK_WORK = "ACK_WORK";
  58 +
  59 + public SyncnQueue(RedisTemplate<Serializable, Serializable> template) {
  60 + this.template = template;
  61 + Fix_Lock = new RedisLock(template);
  62 + Tmp_Lock = new RedisLock(template);
  63 + }
  64 +
  65 + /**
  66 + * <li>@Description:固定任务 {@link ModelType#GET}
  67 + * <p>
  68 + * 固定任务执行效率跟任务链表的长度有关系
  69 + * <li>@param model
  70 + * <li>@return
  71 + * <li>创建人：maliang
  72 + * <li>创建时间：2017年3月13日
  73 + * <li>修改人：
  74 + * <li>修改时间：
  75 + */
  76 + private Boolean pushFixationWork(SyncnModel model) {
  77 + Long tag = template.opsForList().rightPush(FIXATION_WORK, model);
  78 + return tag >= 1;
  79 + }
  80 +
  81 + @SuppressWarnings("unchecked")
  82 + private SyncnModel pullFixationModel() {
  83 + // 添加锁
  84 + try {
  85 + if (Fix_Lock.acquireLock(LOCK_FIXATION_NAME, WAIT_LOCK_TIME)) {
  86 + SyncnModel obj = (SyncnModel) template.opsForList().leftPop(FIXATION_WORK);
  87 + this.pushFixationWork(obj);
  88 + return obj;
  89 + }
  90 + } catch (Exception e) {
  91 + } finally {
  92 + try {
  93 + if (Fix_Lock != null) {
  94 + Fix_Lock.releaseLock();
  95 + }
  96 + } catch (Exception e) {
  97 + e.printStackTrace();
  98 + }
  99 + }
  100 + return null;
  101 +
  102 + }
  103 +
  104 + /**
  105 + * <li>@Description:临时任务
  106 + * <p>
  107 + * {@link ModelType#ADD ModelType#UPDATE ModelType#DEL}
  108 + * <li>@param model
  109 + * <li>@return
  110 + * <li>创建人：maliang
  111 + * <li>创建时间：2017年3月13日
  112 + * <li>修改人：
  113 + * <li>修改时间：
  114 + */
  115 + private boolean pushTempWork(SyncnModel model) {
  116 + Long tag = template.opsForList().rightPush(TMP_WORK, model);
  117 + return tag >= 1;
  118 + }
  119 +
  120 + /**
  121 + * <li>@Description:临时队列数据弹出
  122 + * <li>@return
  123 + * <li>创建人：maliang
  124 + * <li>创建时间：2017年3月13日
  125 + * <li>修改人：
  126 + * <li>修改时间：
  127 + */
  128 + private SyncnModel pullTempModel() {
  129 +
  130 + try {
  131 + if (Tmp_Lock.acquireLock(LOCK_TMP_NAME, WAIT_LOCK_TIME)) {
  132 + Serializable object = template.opsForList().leftPop(TMP_WORK);
  133 + return object != null ? (SyncnModel) object : null;
  134 + }
  135 + } catch (Exception e) {
  136 + e.printStackTrace();
  137 + } finally {
  138 + if (Tmp_Lock != null) {
  139 + try {
  140 + Tmp_Lock.releaseLock();
  141 + } catch (Exception e) {
  142 + e.printStackTrace();
  143 + }
  144 + }
  145 + }
  146 + return null;
  147 + }
  148 +
  149 + /**
  150 + * <li>@Description:将任务信息添加到队列中,任务分为固定任务，临时任务,目前拉取数据的任务为固定任务
  151 + * {@link ModelType#GET}
  152 + * <li>@param model
  153 + * <li>@return
  154 + * <li>创建人：maliang
  155 + * <li>创建时间：2017年3月13日
  156 + * <li>修改人：
  157 + * <li>修改时间：
  158 + */
  159 + public Boolean push(SyncnModel model) {
  160 + if (template == null || model == null)
  161 + return Boolean.FALSE;
  162 + // 设置任务添加时间
  163 + model.setTs(System.currentTimeMillis());
  164 + String modelType = model.getType();
  165 + if (ModelType.isGet(modelType)) {
  166 + return this.pushFixationWork(model);
  167 + } else {
  168 + return this.pushTempWork(model);
  169 + }
  170 + }
  171 +
  172 + /**
  173 + * <li>@Description:获取队列中的数据
  174 + * <p>
  175 + * {@link ModelType#GET} 该参数将会获取固定任务列表中的数据{@link ModelType} 其他参数将会获取临时列表中数据
  176 + * <li>@param type
  177 + * <li>@return
  178 + * <li>创建人：maliang
  179 + * <li>创建时间：2017年3月13日
  180 + * <li>修改人：
  181 + * <li>修改时间：
  182 + */
  183 + public SyncnModel pull(ModelType type) {
  184 + if (ModelType.isGet(type)) {
  185 + return this.pullFixationModel();
  186 + } else {
  187 + return this.pullTempModel();
  188 + }
  189 + }
  190 +
  191 + public RedisTemplate<Serializable, Serializable> getTemplate() {
  192 + return template;
  193 + }
  194 +
  195 + public void setTemplate(RedisTemplate<Serializable, Serializable> template) {
  196 + this.template = template;
  197 + }
  198 +
  199 +}
core.sdk/src/main/java/com/lyms/synch/queue/package-info.java View file @ d5dba59
  1 +/**
  2 + * <li>@ClassName: package-info
  3 + * <li>@Description: 数据中心缓冲队列
  4 + * <li>@author maliang
  5 + * <li>@date 2017年3月13日
  6 + * <li>
  7 + */
  8 +package com.lyms.synch.queue;
core.sdk/src/main/java/com/lyms/util/JsonUtils.java View file @ d5dba59
  1 +package com.lyms.util;
  2 +
  3 +import java.util.Map;
  4 +
  5 +import com.alibaba.fastjson.JSON;
  6 +import com.alibaba.fastjson.JSONObject;
  7 +
  8 +/**
  9 + * <li>@ClassName: JsonUtils
  10 + * <li>@Description: json 转化工具
  11 + * <li>@author maliang
  12 + * <li>@date 2017年3月13日
  13 + * <li>
  14 + */
  15 +public class JsonUtils {
  16 +
  17 + public static <T> String beanToJson(T t) {
  18 + return JSON.toJSONString(t);
  19 + }
  20 +
  21 + /*
  22 + * @SuppressWarnings("unchecked") public static <T> Map<String, Object>
  23 + * beanToMap(T t) { System.out.println(beanToJson(t)); return (Map<String,
  24 + * Object>) JSON.parse(beanToJson(t)); }
  25 + */
  26 +
  27 + public static <T> JSONObject beanToMap(T t) {
  28 + return JSON.parseObject(beanToJson(t));
  29 + }
  30 +
  31 + public static <T> T jsonToBean(String json, Class<T> clazz) {
  32 + return JSON.parseObject(json, clazz);
  33 + }
  34 +
  35 + public static <T> T mapToBean(Map<String, Object> map, Class<T> clazz) {
  36 + return JSON.parseObject(JSON.toJSONString(map), clazz);
  37 + }
  38 +
  39 +}
core.sdk/src/main/java/com/lyms/util/remote/AbsRemote.java View file @ d5dba59
... ... @@ -108,12 +108,13 @@
108 108 */
109 109 public HttpEntity postEntity(Map<String, String> params) {
110 110 MultipartEntityBuilder reqEntity = MultipartEntityBuilder.create();
111   -
112   - ContentType content = ContentType.create(MimeTypeUtils.TEXT_PLAIN_VALUE, CharsetUtils.UTF_8);
113   - for (Entry<String, String> entry : params.entrySet()) {
114   - String key = entry.getKey();
115   - String value = entry.getValue();
116   - reqEntity.addTextBody(key, value, content);
  111 + if (params != null) {
  112 + ContentType content = ContentType.create(MimeTypeUtils.TEXT_PLAIN_VALUE, CharsetUtils.UTF_8);
  113 + for (Entry<String, String> entry : params.entrySet()) {
  114 + String key = entry.getKey();
  115 + String value = entry.getValue();
  116 + reqEntity.addTextBody(key, value, content);
  117 + }
117 118 }
118 119 return reqEntity.build();
119 120 }
core.sdk/src/main/java/com/lyms/util/remote/HttpRemote.java View file @ d5dba59
... ... @@ -9,32 +9,81 @@
9 9 import org.apache.http.client.methods.HttpGet;
10 10 import org.apache.http.client.methods.HttpPost;
11 11 import org.apache.http.client.protocol.HttpClientContext;
  12 +import org.apache.http.entity.SerializableEntity;
12 13  
  14 +import com.lyms.synch.entity.SyncnModel;
  15 +
13 16 /**
14 17 * http,https 远程调用
  18 + *
15 19 * @author maliang
16 20 *
17 21 */
18   -public class HttpRemote extends AbsRemote{
  22 +public class HttpRemote extends AbsRemote {
19 23  
20   - private HttpRemote(){}
21   -
22   - public static HttpRemote build(){
  24 + private HttpRemote() {
  25 + }
  26 +
  27 + public static HttpRemote build() {
23 28 return new HttpRemote();
24 29 }
25   -
  30 +
  31 + public String post(SyncnModel model) {
  32 + if (model == null)
  33 + return null;
  34 +
  35 + String url = model.getRemote();
  36 + HttpClientContext httpContext = HttpClientContext.create();
  37 + HttpPost post = new HttpPost(url);
  38 + HttpEntity entity = null;
  39 + CloseableHttpResponse response = null;
  40 + try {
  41 +
  42 + // ContentType content =
  43 + // ContentType.create(MimeTypeUtils.TEXT_PLAIN_VALUE,
  44 + // CharsetUtils.UTF_8);
  45 +
  46 + // entity = new StringEntity(JsonUtils.beanToJson(model), content);
  47 +
  48 + entity = new SerializableEntity(model);
  49 + // post.setHeader("Content-Type", "text/plain; charset=utf-8");
  50 +
  51 + post.setEntity(entity);
  52 + response = httpclient.execute(post, httpContext);
  53 + entity = response.getEntity();
  54 + httpContext.getCookieStore().clear();
  55 + return responseUTF8(entity);
  56 + } catch (ClientProtocolException e) {
  57 + e.printStackTrace();
  58 + } catch (IOException e) {
  59 + e.printStackTrace();
  60 + } finally {
  61 + try {
  62 + if (response != null) {
  63 + response.close();
  64 + }
  65 + } catch (IOException e) {
  66 + e.printStackTrace();
  67 + }
  68 + }
  69 + return null;
  70 +
  71 + }
  72 +
26 73 /**
27 74 * get 请求
28   - * <p>该方法参数只接受String类型参数
  75 + * <p>
  76 + * 该方法参数只接受String类型参数
  77 + *
29 78 * @param url
30 79 * @param params
31 80 * @return
32 81 */
33   - public String get(String url,Map<String,String> params) {
  82 + public String get(String url, Map<String, String> params) {
34 83 HttpClientContext httpContext = HttpClientContext.create();
35   - //处理参数,模拟http get 请求连接方式
36   - url = super.handlerReqParams(url,params);
37   - LOG.debug("remote : {}",url);
  84 + // 处理参数,模拟http get 请求连接方式
  85 + url = super.handlerReqParams(url, params);
  86 + LOG.debug("remote : {}", url);
38 87 HttpGet get = new HttpGet(url);
39 88 HttpEntity entity = null;
40 89 CloseableHttpResponse response = null;
41 90  
42 91  
43 92  
44 93  
45 94  
... ... @@ -58,26 +107,27 @@
58 107 }
59 108 return null;
60 109 }
61   -
  110 +
62 111 /**
63 112 * post 请求
64   - * <p>目前参数只接受String类型参数
  113 + * <p>
  114 + * 目前参数只接受String类型参数
  115 + *
65 116 * @param url
66 117 * @param params
67 118 * @return
68 119 */
69   - public String post(String url,Map<String,String> params) {
  120 + public String post(String url, Map<String, String> params) {
70 121 HttpClientContext httpContext = HttpClientContext.create();
71 122 HttpPost post = new HttpPost(url);
72 123 HttpEntity entity = null;
73 124 CloseableHttpResponse response = null;
74 125 try {
75   -
76 126 entity = super.postEntity(params);
77   - if(entity!=null) {
  127 + if (entity != null) {
78 128 post.setEntity(entity);
79 129 }
80   -
  130 +
81 131 response = httpclient.execute(post, httpContext);
82 132 entity = response.getEntity();
83 133 httpContext.getCookieStore().clear();
... ... @@ -97,8 +147,6 @@
97 147 }
98 148 return null;
99 149 }
100   -
101   -
102   -
  150 +
103 151 }
core.sdk/src/main/java/com/lyms/web/controller/BaseController.java View file @ d5dba59
1 1 package com.lyms.web.controller;
2 2  
3 3 import java.io.IOException;
  4 +import java.io.ObjectInputStream;
4 5 import java.io.PrintWriter;
5 6 import java.text.SimpleDateFormat;
6 7 import java.util.Date;
... ... @@ -23,6 +24,7 @@
23 24 import com.baomidou.mybatisplus.plugins.pagination.Pagination;
24 25 import com.lyms.common.StringEscapeEditor;
25 26 import com.lyms.exception.BusinessException;
  27 +import com.lyms.synch.entity.SyncnModel;
26 28 import com.lyms.util.HttpUtils;
27 29 import com.lyms.util.StrUtils;
28 30 import com.lyms.web.bean.AjaxResult;
... ... @@ -50,6 +52,13 @@
50 52 * 防止XSS攻击
51 53 */
52 54 binder.registerCustomEditor(String.class, new StringEscapeEditor(true, false));
  55 + }
  56 +
  57 + public SyncnModel conver(HttpServletRequest request) throws IOException, ClassNotFoundException {
  58 + request.setCharacterEncoding("utf-8");
  59 + ObjectInputStream dis = new ObjectInputStream(request.getInputStream());
  60 + SyncnModel model = (SyncnModel) dis.readObject();
  61 + return model;
53 62 }
54 63  
55 64 /**