知识大全 Java Socket通信技术收发线程互斥的解决方法

Posted 2022-07-19 消息

篇首语：最关情，折尽梅花，难寄相思。本文由小常识网(cha138.com)小编为大家整理，主要介绍了知识大全 Java Socket通信技术收发线程互斥的解决方法相关的知识，希望对你有一定的参考价值。

Java Socket通信技术收发线程互斥的解决方法 以下文字资料是由(全榜网网www.cha138.com)小编为大家搜集整理后发布的内容，让我们赶快一起来看一下吧！

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

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

　　 package bill svr;

　　 import java io IOException;

　　 import java io InputStream;

　　 import java io OutputStream;

　　 import InetSocketAddress;

　　 import Socket;

　　 import SocketException;

　　 import SocketTimeoutException;

　　 import java text SimpleDateFormat;

　　 import java util Date;

　　 import java util Properties;

　　 import java util Timer;

　　 import java util TimerTask;

　　 import ncurrent ConcurrentHashMap;

　　 import ncurrent TimeUnit;

　　 import ncurrent locks Condition;

　　 import ncurrent locks ReentrantLock;

　　 import apache log j Logger;

　　 /**

*title: socket通信包装类

*Description:

*CopyRight: CopyRight (c)

*Company:

*Create date:

*author sunnylocus<A mailto: >

</A> * v 初类

　　 * v 对命令收发逻辑及收发线程互斥机制进行了优化

　　处理命令速度由原来 ~ 个/秒提高到 ~ 个/秒

　　 */ public class SocketConnection

　　 private volatile Socket socket;

　　 private int timeout = * ; //超时时间初始值秒

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

　　 private boolean isNeorkConnect = false; //网络是否已连接

　　 private static String host = ;

　　 private static int port;

　　 static InputStream inStream = null;

　　 static OutputStream outStream = null;

　　 private static Logger log =Logger getLogger

　　(SocketConnection class);

　　 private static SocketConnection socketConnection = null;

　　 private static java util Timer heartTimer=null;

//private final Map<String Object> recMsgMap= Collections

synchronizedMap(new HashMap<String Object>());

private final ConcurrentHashMap<String Object> recMsgMap

= new ConcurrentHashMap<String Object>();

　　 private static Thread receiveThread = null;

　　 private final ReentrantLock lock = new ReentrantLock();

　　 private SocketConnection()

　　 Properties conf = new Properties();

　　 try

　　nf load(SocketConnection class getResourceAsStream

　　( nf ));

　　 this timeout = Integer valueOf(conf getProperty( timeout ));

　　 init(conf getProperty( ip ) Integer valueOf

　　(conf getProperty( port )));

　　 catch(IOException e)

　　 log fatal( socket初始化异常! e);

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

　　 /**

　　 * 单态模式

　　 public static SocketConnection getInstance()

　　 if(socketConnection==null)

　　 synchronized(SocketConnection class)

　　 if(socketConnection==null)

　　 socketConnection = new SocketConnection();

　　 return socketConnection;

　　 private void init(String host int port) throws IOException

　　 InetSocketAddress addr = new InetSocketAddress(host port);

　　 socket = new Socket();

　　 synchronized (this)

　　( 【准备与 +addr+ 建立连接】 );

　　nnect(addr timeout);

　　( 【与 +addr+ 连接已建立】 );

　　 inStream = socket getInputStream();

　　 outStream = socket getOutputStream();

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

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

　　 socket setKeepAlive(true);

　　 socket setTrafficClass( x | x );//高可靠性和最小延迟传输

　　 isNeorkConnect=true;

　　 receiveThread = new Thread(new ReceiveWorker());

　　 receiveThread start();

　　 SocketConnection host=host;

　　 SocketConnection port=port;

　　 if(!isLaunchHeartcheck)

　　 launchHeartcheck();

　　 /**

　　 * 心跳包检测

　　 private void launchHeartcheck()

　　 if(socket == null)

　　 throw new IllegalStateException( socket is not

　　established! );

　　 heartTimer = new Timer();

　　 isLaunchHeartcheck = true;

　　 heartTimer schedule(new TimerTask()

　　 public void run()

　　 String msgStreamNo = StreamNoGenerator getStreamNo( kq );

　　 int mstType = ;// 心跳包请求

　　 SimpleDateFormat dateformate = new SimpleDateFormat

　　( yyyyMMddHHmmss );

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

　　 int msgLength = ;//消息头长度

　　 String mandstr = +msgLength + mstType + msgStreamNo;

　　( 心跳检测包 > IVR +mandstr);

　　 int reconnCounter = ;

　　 while(true)

　　 String responseMsg =null;

　　 try

　　 responseMsg = readReqMsg(mandstr);

　　 catch (IOException e)

　　 log error( IO流异常 e);

　　 reconnCounter ++;

　　 if(responseMsg!=null)

( 心跳响应包 < IVR +responseMsg);

　　 reconnCounter = ;

　　 break;

　　 else

　　 reconnCounter ++;

　　 if(reconnCounter > ) //重连次数已达三次判定网络连接中断

　　重新建立连接连接未被建立时不释放锁

　　 reConnectToCTCC(); break;

　　 * * * * );

　　 /**

　　 * 重连与目标IP建立重连

　　 private void reConnectToCTCC()

　　 new Thread(new Runnable()

　　 public void run()

　　( 重新建立与 +host+ : +port+ 的连接 );

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

　　 heartTimer cancel();

　　 isLaunchHeartcheck=false;

　　 isNeorkConnect = false;

　　 receiveThread interrupt();

　　 try

　　 socket close();

　　 catch (IOException e ) log error( 重连时关闭socket连

　　接发生IO流异常 e );

　　 synchronized(this)

　　 for(; ;)

　　 try

　　 Thread currentThread();

　　 Thread sleep( * );

　　 init(host port);

　　 this notifyAll();

　　 break ;

　　 catch (IOException e)

　　 log error( 重新建立连接未成功 e);

　　 catch (InterruptedException e)

　　 log error( 重连线程中断 e);

　　 ) start();

　　 /**

　　 * 发送命令并接受响应

　　 * @param requestMsg

　　 * @return

　　 * @throws SocketTimeoutException

　　 * @throws IOException

　　 public String readReqMsg(String requestMsg) throws IOException

　　 if(requestMsg ==null)

　　 return null;

　　 if(!isNeorkConnect)

　　 synchronized(this)

　　 try

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

　　 if(!isNeorkConnect)

　　 throw new IOException( 网络连接中断！ );

　　 catch (InterruptedException e)

　　 log error( 发送线程中断 e);

　　 String msgNo = requestMsg substring( + );//读取流水号

　　 outStream = socket getOutputStream();

　　 outStream write(requestMsg getBytes());

　　 outStream flush();

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

　　 //注册等待接收消息

　　 recMsgMap put(msgNo msglock);

　　 try

　　 lock lock();

　　 msglock await(timeout TimeUnit MILLISECONDS);

　　 catch (InterruptedException e)

　　 log error( 发送线程中断 e);

　　 finally

　　 lock unlock();

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

　　 if(respMsg!=null &&(respMsg != msglock))

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

　　 return (String) respMsg;

　　 else

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

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

　　 public void finalize()

　　 if (socket != null)

　　 try

　　 socket close();

　　 catch (IOException e)

　　 e printStackTrace();

　　 //消息接收线程

　　 private class ReceiveWorker implements Runnable

　　 String intStr= null;

　　 public void run()

　　 while(!Thread interrupted())

　　 try

　　 byte[] headBytes = new byte[ ];

　　 if(inStream read(headBytes)== )

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

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

　　 return;

　　 byte[] tmp =new byte[ ];

　　 tmp = headBytes;

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

　　 if(tempStr==null || tempStr equals( ))

　　 log error( received message is null );

　　ntinue;

　　 intStr = new String(tmp);

　　 int totalLength =Integer parseInt(intStr);

　　 byte[] msgBytes = new byte[totalLength ];

　　 inStream read(msgBytes);

　　 String resultMsg = new String(headBytes)+ new

　　String(msgBytes);

　　 //抽出消息ID

　　 String msgNo = resultMsg substring( + );

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

　　 if(msglock ==null)

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

　　 recMsgMap remove(msgNo);

　　ntinue;

　　 recMsgMap put(msgNo resultMsg);

　　 try

　　 lock lock();

　　 msglock signalAll();

　　 finally

　　 lock unlock();

　　 catch(SocketException e)

　　 log error( 服务端关闭socket e);

　　 reConnectToCTCC();

　　 catch(IOException e)

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

　　 catch(NumberFormatException e)

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

cha138/Article/program/Java/hx/201311/25550