成都网站建设设计

将想法与焦点和您一起共享

如何解决JavaSocket通信技术收发线程互斥的问题

本篇内容介绍了“如何解决Java Socket通信技术收发线程互斥的问题”的有关知识,在实际案例的操作过程中,不少人都会遇到这样的困境,接下来就让小编带领大家学习一下如何处理这些情况吧!希望大家仔细阅读,能够学有所成!

创新互联长期为超过千家客户提供的网站建设服务,团队从业经验10年,关注不同地域、不同群体,并针对不同对象提供差异化的产品和服务;打造开放共赢平台,与合作伙伴共同营造健康的互联网生态环境。为曲阜企业提供专业的成都网站制作、成都网站设计,曲阜网站改版等技术服务。拥有十余年丰富建站经验和众多成功案例,为您定制开发。

Java Socket通信技术在很长的时间里都在使用,在不少的程序员眼中都有很多高的评价。那么下面我们就看看如何才能掌握这门复杂的编程语言,希望大家在今后的Java Socket通信技术使用中有所收获。

下面就是Java Socket通信技术在解决收发线程互斥的代码介绍。

  1. package com.bill99.svr;   

  2. import java.io.IOException;   

  3. import java.io.InputStream;   

  4. import java.io.OutputStream;   

  5. import java.net.InetSocketAddress;   

  6. import java.net.Socket;   

  7. import java.net.SocketException;   

  8. import java.net.SocketTimeoutException;   

  9. import java.text.SimpleDateFormat;   

  10. import java.util.Date;   

  11. import java.util.Properties;   

  12. import java.util.Timer;   

  13. import java.util.TimerTask;   

  14. import java.util.concurrent.ConcurrentHashMap;   

  15. import java.util.concurrent.TimeUnit;   

  16. import java.util.concurrent.locks.Condition;   

  17. import java.util.concurrent.locks.ReentrantLock;   

  18. import org.apache.log4j.Logger;   

  19. /**   

  20. *

    title: socket通信包装类

       

  21. *

    Description: 

       

  22. *

    CopyRight: CopyRight (c) 2009

       

  23. *

    Company: 99bill.com

       

  24. *

    Create date: 2009-10-14

       

  25. *author sunnylocus   

  26.  * v0.10 2009-10-14 初类   

  27. * v0.11 2009-11-12 对命令收发逻辑及收发线程互斥机制进行了优化,
    处理命令速度由原来8~16个/秒提高到25~32个/秒   

  28. */ public class SocketConnection {   

  29. private volatile Socket socket;   

  30. private int timeout = 1000*10; //超时时间,初始值10秒   

  31. private boolean isLaunchHeartcheck = false;//是否已启动心跳检测   

  32. private boolean isNetworkConnect = false; //网络是否已连接   

  33. private static String host = "";   

  34. private static int port;   

  35. static InputStream inStream = null;   

  36. static OutputStream outStream = null;   

  37. private static Logger log =Logger.getLogger
    (SocketConnection.class);   

  38. private static SocketConnection socketConnection = null;   

  39. private static java.util.Timer heartTimer=null;     

  40. //private final Map recMsgMap= Collections.
    synchronizedMap(new HashMap());   

  41. private final ConcurrentHashMap recMsgMap 
    = new ConcurrentHashMap();   

  42. private static Thread receiveThread = null;   

  43. private final ReentrantLock lock = new ReentrantLock();   

  44. private SocketConnection(){   

  45. Properties conf = new Properties();   

  46. try {   

  47. conf.load(SocketConnection.class.getResourceAsStream
    ("test.conf"));   

  48. this.timeout = Integer.valueOf(conf.getProperty("timeout"));   

  49. init(conf.getProperty("ip"),Integer.valueOf
    (conf.getProperty("port")));   

  50. } catch(IOException e) {   

  51. log.fatal("socket初始化异常!",e);   

  52. throw new RuntimeException("socket初始化异常,请检查配置参数");   

  53. }   

  54. }   

  55. /**   

  56. * 单态模式   

  57. */   

  58. public static SocketConnection getInstance() {   

  59. if(socketConnection==null) {   

  60. synchronized(SocketConnection.class) {   

  61. if(socketConnection==null) {   

  62. socketConnection = new SocketConnection();   

  63. return socketConnection;   

  64. }   

  65. }   

  66. }   

  67. return socketConnection;   

  68. }   

  69. private void init(String host,int port) throws IOException {   

  70. InetSocketAddress addr = new InetSocketAddress(host,port);   

  71. socket = new Socket();   

  72. synchronized (this) {   

  73. log.info("【准备与"+addr+"建立连接】");   

  74. socket.connect(addr, timeout);   

  75. log.info("【与"+addr+"连接已建立】");   

  76. inStream = socket.getInputStream();   

  77. outStream = socket.getOutputStream();   

  78. socket.setTcpNoDelay(true);//数据不作缓冲,立即发送   

  79. socket.setSoLinger(true, 0);//socket关闭时,立即释放资源   

  80. socket.setKeepAlive(true);   

  81. socket.setTrafficClass(0x04|0x10);//高可靠性和最小延迟传输   

  82. isNetworkConnect=true;   

  83. receiveThread = new Thread(new ReceiveWorker());   

  84. receiveThread.start();   

  85. SocketConnection.host=host;   

  86. SocketConnection.port=port;   

  87. if(!isLaunchHeartcheck)   

  88. launchHeartcheck();   

  89. }   

  90. }   

  91. /**   

  92. * 心跳包检测   

  93. */   

  94. private void launchHeartcheck() {   

  95. if(socket == null)   

  96. throw new IllegalStateException("socket is not 
    established!");   

  97. heartTimer = new Timer();   

  98. isLaunchHeartcheck = true;   

  99. heartTimer.schedule(new TimerTask() {   

  100. public void run() {   

  101. String msgStreamNo = StreamNoGenerator.getStreamNo("kq");   

  102. int mstType =9999;//999-心跳包请求   

  103. SimpleDateFormat dateformate = new SimpleDateFormat
    ("yyyyMMddHHmmss");   

  104. String msgDateTime = dateformate.format(new Date());   

  105. int msgLength =38;//消息头长度   

  106. String commandstr = "00" +msgLength + mstType + msgStreamNo;   

  107. log.info("心跳检测包 -> IVR "+commandstr);   

  108. int reconnCounter = 1;   

  109. while(true) {   

  110. String responseMsg =null;   

  111. try {   

  112. responseMsg = readReqMsg(commandstr);   

  113. } catch (IOException e) {   

  114. log.error("IO流异常",e);   

  115. reconnCounter ++;   

  116. }   

  117. if(responseMsg!=null) {   

  118. log.info("心跳响应包 <- IVR "+responseMsg);   

  119. reconnCounter = 1;   

  120. break;   

  121. } else {   

  122. reconnCounter ++;   

  123. }   

  124. if(reconnCounter >3) {//重连次数已达三次,判定网络连接中断,
    重新建立连接。连接未被建立时不释放锁   

  125. reConnectToCTCC(); break;   

  126. }   

  127. }   

  128. }   

  129. },1000 * 60*1,1000*60*2);   

  130. }   

  131. /**   

  132. * 重连与目标IP建立重连   

  133. */   

  134. private void reConnectToCTCC() {   

  135. new Thread(new Runnable(){   

  136. public void run(){   

  137. log.info("重新建立与"+host+":"+port+"的连接");   

  138. //清理工作,中断计时器,中断接收线程,恢复初始变量   

  139. heartTimer.cancel();   

  140. isLaunchHeartcheck=false;   

  141. isNetworkConnect = false;   

  142. receiveThread.interrupt();   

  143. try {   

  144. socket.close();   

  145. } catch (IOException e1) {log.error("重连时,关闭socket连
    接发生IO流异常",e1);}   

  146. //----------------   

  147. synchronized(this){   

  148. for(; ;){   

  149. try {   

  150. Thread.currentThread();   

  151. Thread.sleep(1000 * 1);   

  152. init(host,port);   

  153. this.notifyAll();   

  154. break ;   

  155. } catch (IOException e) {   

  156. log.error("重新建立连接未成功",e);   

  157. } catch (InterruptedException e){   

  158. log.error("重连线程中断",e);   

  159. }   

  160. }   

  161. }   

  162. }   

  163. }).start();   

  164. }   

  165. /**   

  166. * 发送命令并接受响应   

  167. * @param requestMsg   

  168. * @return   

  169. * @throws SocketTimeoutException   

  170. * @throws IOException   

  171. */   

  172. public String readReqMsg(String requestMsg) throws IOException {   

  173. if(requestMsg ==null) {   

  174. return null;   

  175. }   

  176. if(!isNetworkConnect) {   

  177. synchronized(this){   

  178. try {   

  179. this.wait(1000*5); //等待5秒,如果网络还没有恢复,抛出IO流异常   

  180. if(!isNetworkConnect) {   

  181. throw new IOException("网络连接中断!");   

  182. }   

  183. } catch (InterruptedException e) {   

  184. log.error("发送线程中断",e);   

  185. }   

  186. }   

  187. }   

  188. String msgNo = requestMsg.substring(8, 8 + 24);//读取流水号   

  189. outStream = socket.getOutputStream();   

  190. outStream.write(requestMsg.getBytes());   

  191. outStream.flush();   

  192. Condition msglock = lock.newCondition(); //消息锁   

  193. //注册等待接收消息   

  194. recMsgMap.put(msgNo, msglock);   

  195. try {   

  196. lock.lock();   

  197. msglock.await(timeout,TimeUnit.MILLISECONDS);   

  198. } catch (InterruptedException e) {   

  199. log.error("发送线程中断",e);   

  200. } finally {   

  201. lock.unlock();   

  202. }   

  203. Object respMsg = recMsgMap.remove(msgNo); //响应信息   

  204. if(respMsg!=null &&(respMsg != msglock)) {   

  205. //已经接收到消息,注销等待,成功返回消息   

  206. return (String) respMsg;   

  207. } else {   

  208. log.error(msgNo+" 超时,未收到响应消息");   

  209. throw new SocketTimeoutException(msgNo+" 超时,未收到响应消息");   

  210. }   

  211. }   

  212. public void finalize() {   

  213. if (socket != null) {   

  214. try {   

  215. socket.close();   

  216. } catch (IOException e) {   

  217. e.printStackTrace();   

  218. }   

  219. }   

  220. }   

  221. //消息接收线程   

  222. private class ReceiveWorker implements Runnable {   

  223. String intStr= null;   

  224. public void run() {   

  225. while(!Thread.interrupted()){   

  226. try {   

  227. byte[] headBytes = new byte[4];   

  228. if(inStream.read(headBytes)==-1){   

  229. log.warn("读到流未尾,对方已关闭流!");   

  230. reConnectToCTCC();//读到流未尾,对方已关闭流   

  231. return;   

  232. }   

  233. byte[] tmp =new byte[4];   

  234. tmp = headBytes;   

  235. String tempStr = new String(tmp).trim();   

  236. if(tempStr==null || tempStr.equals("")) {   

  237. log.error("received message is null");   

  238. continue;   

  239. }   

  240. intStr = new String(tmp);   

  241. int totalLength =Integer.parseInt(intStr);   

  242. //----------------   

  243. byte[] msgBytes = new byte[totalLength-4];   

  244. inStream.read(msgBytes);   

  245. String resultMsg = new String(headBytes)+ new 
    String(msgBytes);   

  246. //抽出消息ID   

  247. String msgNo = resultMsg.substring(8, 8 + 24);   

  248. Condition msglock =(Condition) recMsgMap.get(msgNo);   

  249. if(msglock ==null) {   

  250. log.warn(msgNo+"序号可能已被注销!响应消息丢弃");   

  251. recMsgMap.remove(msgNo);   

  252. continue;   

  253. }   

  254. recMsgMap.put(msgNo, resultMsg);   

  255. try{   

  256. lock.lock();   

  257. msglock.signalAll();   

  258. }finally {   

  259. lock.unlock();   

  260. }   

  261. }catch(SocketException e){   

  262. log.error("服务端关闭socket",e);   

  263. reConnectToCTCC();   

  264. } catch(IOException e) {   

  265. log.error("接收线程读取响应数据时发生IO流异常",e);   

  266. } catch(NumberFormatException e){   

  267. log.error("收到没良心包,String转int异常,异常字符:"+intStr);   

  268. }   

  269. }   

  270. }   

  271. }   

“如何解决Java Socket通信技术收发线程互斥的问题”的内容就介绍到这里了,感谢大家的阅读。如果想了解更多行业相关的知识可以关注创新互联网站,小编将为大家输出更多高质量的实用文章!


分享题目:如何解决JavaSocket通信技术收发线程互斥的问题
标题URL:http://chengdu.cdxwcx.cn/article/pjdgsc.html