如何解决Java Socket通信技术收发线程互斥

相关推荐

　　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.*

　　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.}

　　272.}

【如何解决Java Socket通信技术收发线程互斥】相关文章：

PHP中如何使用socket进行通信10-27

Java多线程的实现方式11-03

java多线程面试题201707-27

2024年java多线程面试题及答案09-04

sun认证考试辅导：java关于多线程的部分操作07-25

PHP socket的配置09-28