使用java语言，利用多线程调用WebService进行数据处理

柔情只为你懂 2022-08-09 14:56 222阅读 0赞

因工作原因，需要将一个表（tbA）中的所有数据，根据user\_id，去请求webserive获取相关的数据，然后插入到另外的一张表(tbB)中，供他人使用。不过这个表中的数据不少有78万条左右，而这样的大批量数据操作，还不能白天执行。只能在夜里，等服务器负荷低的时候进行执行。考虑如果webservice的效率不高的时候，需要对数据进行分批执行操作。经过综合考虑，最后采用多线程技术（不过最后经过测试，效率还是不错78万条数据，使用10个线程操作，8个多小时就可以了）。

首先在tbA表中，追加一个字段，deal\_tag，这个字段为处理标志字段。该字段的数据默认为'0',如果需要分批处理，则可以设计部分数据中的该字段值为'1'。这个时候就可以读取了。考虑到服务器的负荷问题，采用的方案是读取一部分数据到缓冲区中，同时更改已经进入缓冲区中待处理的数据的deal\_tag字段值为‘B’，标识该数据已经进入缓冲区中。对数据的处理，需要请求webservice。也就有可能是因为因为种种原因，webservice没有正常的返回数据，或者出现异常，而为了不让异常影响数据的处理，则每次处理完一条数据，将该数据放入另外的缓冲区中，如果请求WebService成功，标识数据的字段为'F',请求成功。如果因为WebService或者其他的原因导致失败，则标志该数据字段的deal\_tag字段值为'E',等待下一次的调用调用处理。另外为了不增加数据库的负荷，等到缓冲区中的数据达到一定的数据，一次提交。而不采用处理一条数据，提交一条数据，这样效率太慢。核心代码如下（因不想透漏具体的表名，采用tbA,tbB,tbC代替）：

点击(此处)折叠或打开

1.  import java.sql.ResultSet;  
    
2.  import java.sql.SQLException;  
    
3.  import java.util.ArrayList;  
    
4.  import java.util.Collections;  
    
5.  import java.util.List;  
    
6.    
    
7.  import com.epro.DBUtility.CommonHelper;  
    
8.  import com.epro.DBUtility.DBHelper;  
    
9.  import com.epro.DBUtility.BatchData;  
    
10. public class FetchData implements Runnable \{   
    
11.   
    
12. private int READ\_COUNT = 500; //一次读取数据库多少条记录到缓冲区中  
    
13.   
    
14. private int FLASH\_BUFFER\_COUNT = 2000; //处理后的数据，多少条进行提交到数据库  
    
15.   
    
16. private List<UserInfo> userList = null;  
    
17. private List<String> userStateSqlList = null;  
    
18. private List<BatchData> commitDataList = null;  
    
19.   
    
20. public FetchData()\{   
    
21. userList = Collections.synchronizedList(new ArrayList<UserInfo>());  
    
22. userStateSqlList = new ArrayList<String>();  
    
23. commitDataList = new ArrayList<BatchData>();  
    
24. \}  
    
25.   
    
26. //读取数据  
    
27.   
    
28. private synchronized int readData()\{   
    
29. int result = 0;  
    
30. String strSql = "SELECT user\_id, serial\_number, cust\_name, eparchy\_code, " \+  
    
31. "detail\_install\_address, link\_phone, service\_code, cust\_type, rate, " \+  
    
32. "in\_date, deal\_tag " \+  
    
33. "FROM tbA " \+  
    
34. "WHERE deal\_tag in ('1','E') AND rownum <= " \+ READ\_COUNT;  
    
35.   
    
36. ResultSet rs = null;  
    
37. DBHelper db = new DBHelper();  
    
38. try \{   
    
39. rs = db.executeQuery(strSql);  
    
40. while(rs.next())\{   
    
41. UserInfo user = new UserInfo();  
    
42. user.setUserId(rs.getString("user\_id"));  
    
43. user.setSerialNumber(rs.getString("serial\_number"));  
    
44. user.setCustName(rs.getString("cust\_name"));  
    
45. user.setEparchyCode(rs.getString("eparchy\_code"));  
    
46. user.setDetailInstallAddress(rs.getString("detail\_install\_address"));  
    
47. user.setLinkPhone(rs.getString("link\_phone"));  
    
48. user.setServiceCode(rs.getString("service\_code"));  
    
49. user.setCustType(rs.getString("cust\_type"));  
    
50. user.setRate(rs.getString("rate"));  
    
51. user.setInDate(rs.getString("in\_date"));  
    
52. user.setDealTag(rs.getString("deal\_tag"));  
    
53.   
    
54. userList.add(user); //放入队列中  
    
55.   
    
56. \+\+result;  
    
57. \}  
    
58. \} catch (SQLException e) \{   
    
59. e.printStackTrace();  
    
60. \}finally\{   
    
61. try \{   
    
62. rs.close();  
    
63. \} catch (SQLException e) \{   
    
64. // TODO Auto-generated catch block  
    
65.   
    
66. e.printStackTrace();  
    
67. \}  
    
68. finally\{   
    
69. rs = null;  
    
70. \}  
    
71. db.Close();  
    
72. \}  
    
73. userIntoBuffer(); //将查询出来的数据，更改标志位为'B'  
    
74.   
    
75. return result;  
    
76. \}  
    
77.   
    
78. private void userIntoBuffer()\{   
    
79. int buffersize = this.userList.size();  
    
80. String sql = "";  
    
81. String userID = "";  
    
82. List<String> sqlList = new ArrayList<String>();  
    
83. if(buffersize > 0)\{   
    
84. DBHelper db = new DBHelper();  
    
85. for(int i=0; i<buffersize; i\+\+)\{   
    
86. userID = userList.get(i).getUserId();  
    
87. sql = this.getUserStateSql(userID, "B");  
    
88. sqlList.add(sql);  
    
89. \}  
    
90. db.doBatch(sqlList); //批量进行数据处理  
    
91.   
    
92. db.Close();  
    
93. \}  
    
94. \}  
    
95.   
    
96. private String getUserStateSql(String userID, String dealTag)\{   
    
97. String result = "";  
    
98. result = "UPDATE tbA SET deal\_tag = '" \+ dealTag \+  
    
99. "' WHERE user\_id = " \+ userID;  
    
100.   
    
101. return result;  
    
102. \}  
    
103.   
    
104. public synchronized UserInfo getUserInfo()\{   
    
105. UserInfo result = null;  
    
106. if(userList.size() > 0)\{   
    
107. result = userList.remove(0);  
    
108. \}  
    
109. return result;  
    
110. \}  
    
111.   
    
112. //开始进行数据的处理  
    
113.   
    
114. public void beginDealData()\{   
    
115. ResInfo resInfo = null;  
    
116. while(true)\{   
    
117. UserInfo user = this.getUserInfo();  
    
118. if(user != null && !"".equals(user.getUserId().trim()))\{  //当没有数据时，跳出循环  
    
119.   
    
120. //进行WebService请求，获取生产环境的资源数据，对数据进行解析，返回资源对象  
    
121.   
    
122. resInfo = getResInfo(user.getUserId());  
    
123. \}  
    
124. else if(user == null)\{   
    
125. break;  
    
126. \}  
    
127. else if(user != null && "".equals(user.getUserId().trim()))\{   
    
128. resInfo = new ResInfo();  
    
129. resInfo.out\_err\_id = "-1";  
    
130. resInfo.ln\_line\_flag = "0";  
    
131. resInfo.out\_err\_msg = "beginDealData is null.";  
    
132. System.out.println("user id is null.");  
    
133. \}  
    
134. else\{   
    
135. ;  
    
136. \}  
    
137. userFinish(user, resInfo); //将用户的数据和资源的数据进行合并操作  
    
138.   
    
139. \}  
    
140. int datacount = readData(); //读取表，看是否还有要处理的数据  
    
141.   
    
142. if(datacount > 0)\{   
    
143. beginDealData(); //递归调用，对需要处理的数据，进行继续读取  
    
144.   
    
145. \}  
    
146. this.flashBuffer(); //做最后一次的提交处理后的buffer操作  
    
147.   
    
148. System.out.println("finish at " \+ CommonHelper.getCurrentDateTime());  
    
149. \}  
    
150.   
    
151. public ResInfo getResInfo(String productID)\{   
    
152. ResInfoReader myReader = new ResInfoReader();  
    
153. return myReader.beginRequest(productID);  
    
154. \}  
    
155.   
    
156. //获取地区名称  
    
157.   
    
158. public String getAreaName(String areaCode)\{   
    
159. String result = "";  
    
160. String strSql = "SELECT tb1.area\_name area\_name from " \+  
    
161. "tbC tb1 " \+  
    
162. "where tb1.area\_code = ?";  
    
163. String\[\] param = \{  areaCode \};  
    
164. ResultSet rs = null;  
    
165. DBHelper db = new DBHelper();  
    
166. rs = db.executeQuery(strSql, param);  
    
167. try \{   
    
168. while(rs.next())\{   
    
169. result = rs.getString("area\_name");  
    
170. \}  
    
171. \} catch (SQLException e) \{   
    
172. // TODO Auto-generated catch block  
    
173.   
    
174. e.printStackTrace();  
    
175. \} finally\{   
    
176. try \{   
    
177. rs.close();  
    
178. \} catch (SQLException e) \{   
    
179. // TODO Auto-generated catch block  
    
180.   
    
181. e.printStackTrace();  
    
182. \}  
    
183. db.Close();  
    
184. \}  
    
185.   
    
186. return result;  
    
187. \}  
    
188.   
    
189. private synchronized int userFinish(UserInfo user, ResInfo resInfo)\{   
    
190. int result = 0;  
    
191. String strSql = "";  
    
192. String strAreaName = ""; //地区名称  
    
193.   
    
194. String\[\] param = null;  
    
195. if("".equals(user.getServiceCode()))\{   
    
196. strAreaName = this.getAreaName(user.getEparchyCode());  
    
197. \}  
    
198. else\{   
    
199. strAreaName = this.getAreaName(user.getServiceCode());  
    
200. \}  
    
201. strSql = "INSERT INTO tbB(USER\_ID, SERIAL\_NUMBER, " \+  
    
202. "CUST\_NAME, DEAL\_TYPE, EPARCHY\_CODE, DETAIL\_INSTALL\_ADDRESS, LINK\_PHONE, " \+  
    
203. "EXCH\_ID, LINEBOX\_SEQ, LINEBOX, LINEBOX\_ADDR, ROW\_ID, COL\_ID, VCOL\_SEQ, " \+  
    
204. "PCABLE, PCABLE\_SEQ, OCABLE, OCABLE\_SEQ, FPCONNECT\_ROW\_ID, FPCONNECT\_COL\_ID, " \+  
    
205. "FPCONNECT\_SEQ, FCONNECT, FCONNECT\_ADDR, FCONNECT\_NAME, FOCONNECT\_ROW\_ID, " \+  
    
206. "FOCONNECT\_COL\_ID, FOCONNECT\_SEQ, SPCONNECT\_ROW\_ID, SPCONNECT\_COL\_ID, " \+  
    
207. "SPCONNECT\_SEQ, SCONNECT, SCONNECT\_ADDR, SCONNECT\_NAME, SOCONNECT\_ROW\_ID, " \+  
    
208. "SOCONNECT\_COL\_ID, SOCONNECT\_SEQ, LN\_FAC\_CODE, LS\_SWITCH\_ID, LN\_SWITCH\_NAME, " \+  
    
209. "LS\_SWITCH\_MODE, LN\_FAC\_TYPE, LS\_NODE\_CODE, LS\_NODE\_ID, LS\_MAC\_CODE, LS\_FRAME\_CODE, " \+  
    
210. "LN\_DHCOL\_CODE, LS\_FRAME\_ID, LS\_SLOT\_ID, LS\_PORT\_ID, LN\_DHCOL\_SEQ, LN\_LINE\_ID, " \+  
    
211. "LN\_LINE\_FLAG, RELA\_PRODUCT\_NO, MEASURE\_NAME, MEASURE\_CODE, LINE\_TYPE, REMOVE\_FLAG, " \+  
    
212. "IN\_TIME, OUT\_TIME, EXCH\_NAME, TRADE\_ID, CUST\_TYPE, RATE) VALUES(?, ?, ?, ?, ?, ?, " \+  
    
213. "?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, " \+  
    
214. "?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, sysdate, null, " \+  
    
215. "?, ?, ?, ?)";  
    
216. param = new String\[\]\{   
    
217. user.getUserId(),  
    
218. user.getSerialNumber(),  
    
219. user.getCustName(),  
    
220. "",  
    
221. user.getEparchyCode(),  
    
222. user.getDetailInstallAddress(),  
    
223. user.getLinkPhone(),  
    
224. user.getServiceCode(),  
    
225. resInfo.linebox\_seq,  
    
226. resInfo.linebox,  
    
227. resInfo.linebox\_addr,  
    
228. resInfo.row\_id,  
    
229. resInfo.col\_id,  
    
230. resInfo.vcol\_seq,  
    
231. resInfo.pcable,  
    
232. resInfo.pcable\_seq,  
    
233. resInfo.ocable,  
    
234. resInfo.ocable\_seq,  
    
235. resInfo.fpconnect\_row\_id,  
    
236. resInfo.fpconnect\_col\_id,  
    
237. resInfo.fpconnect\_seq,  
    
238. resInfo.fconnect,  
    
239. resInfo.fconnect\_addr,  
    
240. resInfo.fconnect\_name,  
    
241. resInfo.foconnect\_row\_id,  
    
242. resInfo.foconnect\_col\_id,  
    
243. resInfo.foconnect\_seq,  
    
244. resInfo.fpconnect\_row\_id,  
    
245. resInfo.fpconnect\_col\_id,  
    
246. resInfo.fpconnect\_seq,  
    
247. resInfo.sconnect,  
    
248. resInfo.sconnect\_addr,  
    
249. resInfo.sconnect\_name,  
    
250. resInfo.soconnect\_row\_id,  
    
251. resInfo.soconnect\_col\_id,  
    
252. resInfo.soconnect\_seq,  
    
253. resInfo.ln\_fac\_code,  
    
254. resInfo.ls\_switch\_id,  
    
255. resInfo.ln\_switch\_name,  
    
256. resInfo.ls\_switch\_mode,  
    
257. resInfo.ln\_fac\_type,  
    
258. resInfo.ls\_node\_code,  
    
259. resInfo.ls\_node\_id,  
    
260. resInfo.ls\_mac\_code,  
    
261. resInfo.ls\_frame\_code,  
    
262. resInfo.ln\_dhcol\_code,  
    
263. resInfo.ls\_frame\_id,  
    
264. resInfo.ls\_slot\_id,  
    
265. resInfo.ls\_port\_id,  
    
266. resInfo.ln\_dhcol\_seq,  
    
267. resInfo.ln\_line\_id,  
    
268. resInfo.ln\_line\_flag,  
    
269. resInfo.rela\_product\_no,  
    
270. resInfo.measure\_name,  
    
271. resInfo.measure\_code,  
    
272. resInfo.line\_type,  
    
273. "0",  
    
274. strAreaName,  
    
275. "",  
    
276. user.getCustType(),  
    
277. user.getRate()  
    
278. \};  
    
279.   
    
280. if("0".equals(resInfo.out\_err\_id))\{   
    
281. BatchData data = new BatchData(strSql, param);  
    
282. commitDataList.add(data); //提交处理后的数据，到缓冲区  
    
283.   
    
284.   
    
285. result = 1;  
    
286. strSql = this.getUserStateSql(user.getUserId(), "F");  
    
287. \}  
    
288. else\{   
    
289. //当调用webservice后，不能正常的获取数据，则修改标志位为 'E'  
    
290.   
    
291. strSql = this.getUserStateSql(user.getUserId(), "E");  
    
292. \}  
    
293. userStateSqlList.add(strSql); //提交用户状态数据到缓冲区  
    
294.   
    
295. checkBuffer();  
    
296. //System.out.println("deal user id: " + user.getUserId() + " at " + CommonHelper.getCurrentDateTime());  
    
297.   
    
298. return result;  
    
299. \}  
    
300.   
    
301. private void checkBuffer()\{   
    
302. if(userStateSqlList.size() >= FLASH\_BUFFER\_COUNT)\{   
    
303. this.flashBuffer();  
    
304. \}  
    
305. \}  
    
306.   
    
307. private synchronized void flashBuffer()\{   
    
308. List<String> stateList = new ArrayList<String>();  
    
309. List<BatchData> commitList = new ArrayList<BatchData>();  
    
310. DBHelper db = new DBHelper();  
    
311.   
    
312. while(commitDataList.size()>0)\{   
    
313. BatchData data112 = commitDataList.remove(0);  
    
314. commitList.add(data112);  
    
315. \}  
    
316. if(commitList.size() > 0)\{   
    
317. db.doBatchData(commitList);  
    
318. \}  
    
319.   
    
320. while(userStateSqlList.size() > 0)\{   
    
321. stateList.add(userStateSqlList.remove(0));  
    
322. \}  
    
323. if(stateList.size() > 0)\{   
    
324. db.doBatch(stateList);  
    
325. \}  
    
326.   
    
327. db.Close();  
    
328. System.out.println("flash buffer at " \+ CommonHelper.getCurrentDateTime());  
    
329. \}  
    
330.   
    
331. public void run() \{   
    
332. this.beginDealData();  
    
333. \}  
    
334. \}

java调用WebService的代码如下：

点击(此处)折叠或打开

其中webservice的操作，其中需要jar包，请下载apache.axis，XStream包，并加入到库中。其中axis主要是用来进行webServie请求用，而XStream主要使用将返回的XML数据，翻译为java对象。

因为是java project项目，因此少不了main函数了。代码如下：

点击(此处)折叠或打开

1.  public class Shell \{   
    
2.  public static void main(String\[\] args)\{   
    
3.  boolean isthread = true;  
    
4.  FetchData fetch = new FetchData();  
    
5.  if(isthread)\{   
    
6.  new Thread(fetch).start();  
    
7.  new Thread(fetch).start();  
    
8.  new Thread(fetch).start();  
    
9.  new Thread(fetch).start();  
    
10. new Thread(fetch).start();  
    
11. new Thread(fetch).start();  
    
12. new Thread(fetch).start();  
    
13. new Thread(fetch).start();  
    
14. new Thread(fetch).start();  
    
15. new Thread(fetch).start();  
    
16. \}else\{   
    
17. ResInfo resInfo = fetch.getResInfo("productID");  
    
18. System.out.println("out\_err\_id: " \+ resInfo.out\_err\_id);  
    
19. System.out.println("out\_err\_msg: " \+ resInfo.out\_err\_msg);  
    
20. System.out.println("linebox: " \+ resInfo.linebox);  
    
21. System.out.println("linebox\_addr: " \+ resInfo.linebox\_addr);  
    
22. System.out.println("pcable: " \+ resInfo.pcable);  
    
23. System.out.println("ocable: " \+ resInfo.ocable);  
    
24. \}  
    
25. \}  
    
26. \}

我在main函数中，使用了10个线程。可以根据具体的需要，进行线程的添加，减少。