jiangjiazhi / center

Commit 2fee7384f2e2c5df231982c93324b2854f5abc8e

Authored by maliang about 9 years ago

Exists in master

修改提交方式，编写测试用例

Showing 6 changed files with 155 additions and 196 deletions

center.manager/src/main/resources/xml/app-sycn.xml
center.manager/src/main/resources/xml/app-syncn.xml
center.manager/src/test/java/center/manager/test/user/WorkTest.java
core.sdk/src/main/java/com/lyms/sycn/SycnCallback.java
core.sdk/src/main/java/com/lyms/sycn/SycnCenter.java
core.sdk/src/main/java/com/lyms/sycn/SycnHandler.java

	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.sycn.SycnCenter" factory-method="install" >
	13	+ <property name="queue" ref="queue" />
	14	+ </bean>
	15	+
	16	+ <!-- 使用的队列信息 -->
	17	+ <bean id="queue" class="com.lyms.sycn.queue.SycnQueue">
	18	+ <!-- 依赖redis template -->
	19	+ <constructor-arg index="0" ref="redisTemplate"/>
	20	+ </bean>
	21	+</beans>

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>

...	...	@@ -4,15 +4,9 @@
4	4	import org.junit.Test;
5	5	import org.springframework.beans.factory.annotation.Autowired;
6	6
7		-import com.alibaba.fastjson.JSON;
8		-import com.lyms.cm.entity.sys.SysUsers;
9		-import com.lyms.sycn.ParamsAdpter;
10	7	import com.lyms.sycn.SycnCallback;
11	8	import com.lyms.sycn.SycnCenter;
12		-import com.lyms.sycn.SycnCenter.Work;
13	9	import com.lyms.sycn.channel.ChannelData;
14		-import com.lyms.util.HttpUtils;
15		-import com.lyms.util.JsonUtils;
16	10	import com.lyms.util.StrUtils;
17	11
18	12	/**
19	13
20	14
21	15
22	16
23	17
24	18
...	...	@@ -27,140 +21,54 @@
27	21	@Autowired
28	22	public SycnCenter center;
29	23
	24	+ /**
	25	+ * <li>@Description:测试往固定任务队列中添加任务
	26	+ * <li>
	27	+ * <li>创建人：maliang
	28	+ * <li>创建时间：2017年3月16日
	29	+ * <li>修改人：
	30	+ * <li>修改时间：
	31	+ */
30	32	@Test
31		- public void push() {
	33	+ public void pushFix() {
	34	+ String id = StrUtils.uuid();
32	35	String remote = "http://127.0.0.1:9090/hospital.web/remote/sycn";
33	36	String remoteClazz = "com.lyms.hospital.service.sys.impl.SyncDataBasicServiceImpl";
34	37	String remoteMethod = "selectOne";
35		- Object data = null;// new Object[] { user, null };
36		-
37		- boolean loop = true;
38		-
39		- ChannelData push = new ChannelData(remote, remoteClazz, remoteMethod, null, loop);
40		-
41		- center.buildWork().push(push, new SycnCallback() {
42		- @Override
43		- public void callBack(Object object) {
44		- System.out.println("push callBack : " + object);
45		- }
46		- });
	38	+ center.pushFix(id, remote, remoteClazz, remoteMethod, null);
47	39	}
48	40
	41	+ /**
	42	+ * <li>@Description:测试往固定任务队列中添加任务
	43	+ * <li>
	44	+ * <li>创建人：maliang
	45	+ * <li>创建时间：2017年3月16日
	46	+ * <li>修改人：
	47	+ * <li>修改时间：
	48	+ */
49	49	@Test
50		- public void workTest() throws InterruptedException {
51		- new Thread(new Runnable() {
52		-
53		- @Override
54		- public void run() {
55		- tmp();
56		- }
57		- }).start();
58		-
59		- /*
60		- * new Thread(new Runnable() {
61		- *
62		- * @Override public void run() { try { fix(); } catch
63		- * (InterruptedException e) { e.printStackTrace(); } } }).start();
64		- */
65		- //
66		- while (true) {
67		- push();
68		- Thread.sleep(30000);
69		- }
70		-
	50	+ public void pushTmp() {
	51	+ String id = StrUtils.uuid();
	52	+ String remote = "http://127.0.0.1:9090/hospital.web/remote/sycn";
	53	+ String remoteClazz = "com.lyms.hospital.service.sys.impl.SyncDataBasicServiceImpl";
	54	+ String remoteMethod = "selectOne";
	55	+ // 参数构建如下注释方式
	56	+ // ParamsAdpter adpter =
	57	+ // ParamsAdpter.builder().push(null).push(null).push(null);;
	58	+ center.pushFix(id, remote, remoteClazz, remoteMethod, null);
71	59	}
72	60
73	61	@Test
74		- public void tmp() {
75		- while (true) {
76		- final Work work = center.buildWork();
77		- try {
78		- // final AckObj ack = new AckObj();
79		-
80		- work.pullTmp(new SycnCallback() {
81		- @Override
82		- public void callBack(Object object) {
83		- if (object == null)
84		- return;
85		- if (object instanceof ChannelData) {
86		- ChannelData model = (ChannelData) object;
87		- String result = HttpUtils.REMOTE.post(model);
88		- System.out.println(result);
89		- model = JsonUtils.jsonToBean(result, ChannelData.class);
90		- System.out.println("回执: " + ToStringBuilder.reflectionToString(model));
91		- if (model != null) {
92		- if (!model.getAck()) {
93		- work.backPressure();
94		- } else {
95		- System.out.println("远程数据处理成功");
96		- }
97		- }
98		- }
99		- }
100		- });
101		- } catch (Exception e) {
102		- e.printStackTrace();
103		- // 异常数据反压
104		- System.out.println("数据反压");
105		- work.backPressure();
106		- }
107		- }
108		- }
109		-
110		- @Test
111		- public void fix() throws InterruptedException {
112		- // 获取固定任务队列任务
113		- // 包含ack 回调方法
114		- // while (true) {
115		- Work work2 = center.buildWork();
116		- try {
117		- // final AckObj ack = new AckObj();
118		- work2.pullFix(new SycnCallback() {
119		- @Override
120		- public void callBack(Object object) {
121		- if (object != null && object instanceof ChannelData) {
122		- // remote
123		- ChannelData model = (ChannelData) object;
124		- // 处理实际业务
125		- System.out.println(ToStringBuilder.reflectionToString(model));
126		- System.out.println(model.getRemote() + "====>fix");
127		- String result = HttpUtils.REMOTE.post(model);
128		- System.out.println(result);
129		- // ack.setData(model.getData());
130		- }
	62	+ public void pull() {
	63	+ center.pull(true, new SycnCallback() {
	64	+ @Override
	65	+ public void callBack(Object object) {
	66	+ if (object instanceof ChannelData) {
	67	+ ChannelData data = (ChannelData) object;
	68	+ System.out.println(ToStringBuilder.reflectionToString(data));
131	69	}
132		- });// .ackGet(params);// .ackGet(null)
133		- } catch (Exception e) {
134		- e.printStackTrace();
135		- }
136		- // Thread.sleep(15000);
137		- // }
138		- }
139		-
140		- public static void buildParams() throws ClassNotFoundException {
141		- String uid = StrUtils.uuid();
142		-
143		- ChannelData data = ChannelData.emtpy();
144		- data.setId(uid);
145		- data.setRemote(null);
146		- data.setRemoteClazz("com.lyms.cm.service.sys.impl.SysUsersServiceImpl");
147		- data.setRemoteMethod("addUser");
148		- SysUsers user = new SysUsers();
149		- user.setId(uid);
150		- user.setName("拉取信息测试");
151		- // Object params = new Object[] { user, null };
152		-
153		- // 设置参数
154		- ParamsAdpter params = ParamsAdpter.builder().push(user).push(null);
155		- data.setData(params.toJsonString());
156		-
157		- String str = JSON.toJSONString(data);
158		- System.out.println(str);
159		-
160		- }
161		-
162		- public static void main(String[] args) throws ClassNotFoundException {
163		- buildParams();
	70	+ }
	71	+ });
164	72	}
165	73
166	74	}

1	1	package com.lyms.sycn;
2	2
	3	+import org.apache.ibatis.javassist.expr.Instanceof;
	4	+
3	5	/**
4	6	* <li>@ClassName: CenterCallback
5	7	* <li>@Description: 数据回调
...	...	@@ -8,6 +10,14 @@
8	10	* <li>
9	11	*/
10	12	public interface SycnCallback {
	13	+ /**
	14	+ * <li>@Description:回调方法
	15	+ * <li>@param object Instanceof {@link Instanceof ChannelData}
	16	+ * <li>创建人：maliang
	17	+ * <li>创建时间：2017年3月16日
	18	+ * <li>修改人：
	19	+ * <li>修改时间：
	20	+ */
11	21	public void callBack(Object object);
12	22	}

...	...	@@ -43,6 +43,93 @@
43	43	}
44	44
45	45	/**
	46	+ * <li>@Description:在3次没有成功将会放弃,如果需要对失败处理，使用
	47	+ * {@link Work#push(ChannelData, SycnCallback)}
	48	+ * <p>
	49	+ * 该方法使用尽量异步操作，避免阻塞
	50	+ * <li>@return
	51	+ * <li>创建人：maliang
	52	+ * <li>创建时间：2017年3月16日
	53	+ * <li>修改人：
	54	+ * <li>修改时间：
	55	+ */
	56	+ public boolean pushFix(String id, String remote, String remoteClazz, String remoteMethod, ParamsAdpter adpter) {
	57	+ ChannelData data = new ChannelData(id, remote, remoteClazz, remoteMethod, adpter.toJsonString(), true);
	58	+
	59	+ int tag = 0;
	60	+
	61	+ do {
	62	+ try {
	63	+ buildWork().push(data, new SycnCallback() {
	64	+ @Override
	65	+ public void callBack(Object object) {
	66	+ // nothing to do
	67	+ }
	68	+ });
	69	+ return true;
	70	+ } catch (Exception e) {
	71	+ e.printStackTrace();
	72	+ tag++;
	73	+ }
	74	+ } while (tag < 3);
	75	+
	76	+ return false;
	77	+ }
	78	+
	79	+ /**
	80	+ *
	81	+ * <li>@Description: 获取数据,等价于下面两个方法
	82	+ * <p>
	83	+ * {@link Work#pullFix(SycnCallback)}
	84	+ * <p>
	85	+ * {@link Work#pullTmp(SycnCallback)}
	86	+ * <li>@param loop 从固定任务队列中获取任务
	87	+ * <li>@param callBack
	88	+ * <li>创建人：maliang
	89	+ * <li>创建时间：2017年3月16日
	90	+ * <li>修改人：
	91	+ * <li>修改时间：
	92	+ */
	93	+ public void pull(boolean loop, SycnCallback callBack) {
	94	+ if (loop)
	95	+ buildWork().pullFix(callBack);
	96	+ else
	97	+ buildWork().pullTmp(callBack);
	98	+ }
	99	+
	100	+ /**
	101	+ * <li>@Description:在3次没有成功将会放弃,如果需要对失败处理，使用
	102	+ * {@link Work#push(ChannelData, SycnCallback)}
	103	+ * <p>
	104	+ * 该方法使用尽量异步操作，避免阻塞
	105	+ * <li>@return
	106	+ * <li>创建人：maliang
	107	+ * <li>创建时间：2017年3月16日
	108	+ * <li>修改人：
	109	+ * <li>修改时间：
	110	+ */
	111	+ public boolean pushTmp(String id, String remote, String remoteClazz, String remoteMethod, ParamsAdpter adpter) {
	112	+ ChannelData data = new ChannelData(id, remote, remoteClazz, remoteMethod, adpter.toJsonString(), false);
	113	+ int tag = 0;
	114	+ do {
	115	+ try {
	116	+ buildWork().push(data, new SycnCallback() {
	117	+ @Override
	118	+ public void callBack(Object object) {
	119	+ // nothing to do
	120	+ }
	121	+ });
	122	+ return true;
	123	+ } catch (Exception e) {
	124	+ e.printStackTrace();
	125	+ tag++;
	126	+ }
	127	+ } while (tag < 3);
	128	+
	129	+ return false;
	130	+ }
	131	+
	132	+ /**
46	133	* <li>@Description:构建一个工作器
47	134	* <li>@return
48	135	* <li>创建人：maliang

...	...	@@ -114,51 +114,5 @@
114	114	}
115	115	return null;
116	116	}
117		-
118		- /**
119		- * <li>@Description:TODO(方法描述)
120		- * <li>@param t
121		- * <li>@param loop
122		- * <li>@return
123		- * <li>创建人：maliang
124		- * <li>创建时间：2017年3月16日
125		- * <li>修改人：
126		- * <li>修改时间：
127		- */
128		- public static boolean pushFix(String id, String remote, String remoteClazz, String remoteMethod,
129		- ParamsAdpter adpter) {
130		-
131		- ChannelData data = new ChannelData(id, remote, remoteClazz, remoteMethod, adpter.toJsonString(), true);
132		- try {
133		- center.buildWork().push(data, new SycnCallback() {
134		- @Override
135		- public void callBack(Object object) {
136		- // nothing to do
137		- }
138		- });
139		- } catch (Exception e) {
140		- e.printStackTrace();
141		- return false;
142		- }
143		- return true;
144		- }
145		-
146		- public static boolean pushTmp(String id, String remote, String remoteClazz, String remoteMethod,
147		- ParamsAdpter adpter) {
148		- ChannelData data = new ChannelData(id, remote, remoteClazz, remoteMethod, adpter.toJsonString(), false);
149		- try {
150		- center.buildWork().push(data, new SycnCallback() {
151		- @Override
152		- public void callBack(Object object) {
153		- // nothing to do
154		- }
155		- });
156		- } catch (Exception e) {
157		- e.printStackTrace();
158		- return false;
159		- }
160		- return true;
161		- }
162		-
163	117	}