基于Mina同步通信中存在的BUG的一种解决方案

需求：
有一个命令序列，包含三个命令

• 1.调整命令
• 2.测量命令
• 3.获取命令

每个命令，对应的服务端都可能会返回一个字符串。
现在的需求是：
step1.发送命令1，如果服务端没返回结果，重新发送，如果返回了，则执行step2
step2.发送命令2，如果服务端没返回结果，重新发送，如果返回了，则执行step3
step3.发送命令3，如果服务端没返回结果，重新发送，如果返回了，则执行step1

针对这个需求，我之前也在问答模块里提过问，但是没人回答，请看下面的：
http://www.iteye.com/problems/86196

现在通信框架是基于Mina2的，Mina其实提供了同步通信操作

 session.getConfig().setUseReadOperation(true);  

但是实际用起来，却发现无法达到同步效果。网上也有类似的疑问，具体表现为：如果发送一条命令后，在给定的超时时间内，无法获取服务端相应，则后续的通讯将阻塞。简单的说就是如果通讯一旦产生超时，则无法恢复通讯。
       
我搜了一下网上的资源，也有人提到过这个问题：

http://bochengwen.iteye.com/blog/982062

进一步搜索发现在Mina网站上，还有外国人提过这个问题，但是一直都是Open状态。
https://issues.apache.org/jira/browse/DIRMINA-777?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel

引用

• Status: Open Open
• Resolution: Unresolved
• Fix Version/s: 2.0.5

我当时也参与了评论，但是用的是中文文字，后来管理员直接把我的评论屏蔽了，而且还说：

Emmanuel Lecharny 写道

Emmanuel Lecharny added a comment - 19/Jul/12 07:26
Please, *do not* add comments in a foreign language... (even if those are related to the problem, google translate is not that good in chinese...)

晕死啊，老外就是老外。

看样子是准备2.0.5这个版本中fixed掉。对于这个bug，网上有网友提出了临时解决方案

((AbstractIoSession) session).offerReadFuture(null);// 针对同步实现的bug  

其实这样做并不能达到效果，具体原因我不想多说。我目前的解决方法是

String request = "XXXX"; 
					
/**发送数据到服务端**/

session.write(IoBuffer.wrap(request.getBytes()));  
/**
 * 获取Client.class的锁定，进入wait set,等待
 * 直到在Decoder中获取到了服务端响应的数据，notify，或者超时
 */					
synchronized(Client.class){
        try {
		Client.class.wait(3000L);
	} catch (InterruptedException e) {
		Thread.currentThread().interrupt();
	}
}					
/**下面的代码就是接收到数据后或者wait超时后执行的**/
codes...

在Decoder类中

 @Override  
    public void decode(IoSession session, IoBuffer in, ProtocolDecoderOutput out)  
            throws Exception {  
    	synchronized (Client.class) {
			Client.class.notifyAll();//唤醒
		}
othercode...
    }  

具体的notify的时机，应该根据需求而定的，例如我只要监听服务端返回数据就notiry的话，直接写在这里就行了，如果说还要判断数据是正确的才能发送下一条命令，那么显然放在这里notify是不合适的。
附：串口通讯程序

package comm;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.ConcurrentHashMap;

import javax.comm.CommPortIdentifier;
import javax.comm.SerialPort;
import javax.comm.SerialPortEvent;
import javax.comm.SerialPortEventListener;


/**
 * java串口通讯
 *
 * @author  administrator
 * @since   comm 1.0.0
 * @created 2012-8-2
 */
public class Comm {
	private SerialPort sPort;

	private OutputStream os ;

	private Map<String,List<String>> commands;

	private Recieved rec;
	
	public Comm(){
		commands = new ConcurrentHashMap<String, List<String>>();
		List<String> command1 = new ArrayList<String>();
		//初始化commands...
	}

	public boolean init(){
		try{
			CommPortIdentifier portId = CommPortIdentifier.getPortIdentifier("COM3");
			sPort = (SerialPort) portId.open("TPS", 1000);
			sPort.setOutputBufferSize(8);
			sPort.setSerialPortParams(9600,SerialPort.DATABITS_8,SerialPort.STOPBITS_1,SerialPort.PARITY_NONE);
			rec = new Recieved(sPort);
			sPort.addEventListener(rec);
			sPort.notifyOnDataAvailable(true); 
			sPort.enableReceiveTimeout(1000);
			os = sPort.getOutputStream();
		}catch(Exception e){
			return false;
		}
		return true;


	}

	public void start(){
		Timer t = new Timer();
		t.schedule(new TimerTask(){
			@Override
			public void run() {
				final Iterator<Entry<String,List<String>>> it = commands.entrySet().iterator();
				while(it.hasNext()){
					Entry<String,List<String>> e = it.next();
					for(String s:e.getValue()){
						System.out.println(e.getKey()+","+s);
						try {
							os.write(s.getBytes());
						} catch (IOException e1) {
							e1.printStackTrace();
						}
						synchronized (Comm.class) {
							try {
								Comm.class.wait(15000);
							} catch (InterruptedException e1) {
								Thread.currentThread().interrupt();
							}
						}
					}
				}
			}
			
		}, 1000,2000);
		
	}

	public static void main(String[] args) {
		Comm c = new Comm();
		c.init();
		c.start();
	}
}

class Recieved  implements SerialPortEventListener {
	private InputStream is;
	private SerialPort sPort;
	private StringBuffer msg = new StringBuffer();
	public Recieved(SerialPort sPort){
		this.sPort = sPort;
	}
	
	@Override
	public void serialEvent(SerialPortEvent e) {
		try {
			is = sPort.getInputStream();
		} catch (IOException e2) {
			e2.printStackTrace();
		}
		
		if (e.getEventType()==SerialPortEvent.DATA_AVAILABLE) { 
			try {
				while(is.available() > 0){
					msg.append((char)is.read());
				}
			} catch (IOException e1) {
				e1.printStackTrace();
			}
		
			if(msg.toString().startsWith("%R1P")&&msg.toString().endsWith("\r\n")){
				System.out.println("recieved:"+msg);
				msg = new StringBuffer("");
				synchronized (Comm.class){
					Comm.class.notifyAll();
				}
			}
			
		}  
		
	}
}

评论

1 楼 scholers 2013-02-27  

这个方法也不错，有兴趣换工作不？