mina框架采用可重入锁实现消息重发

怼烎@ 2022-06-01 10:49 223阅读 0赞

前段时间在用mina的时候，一开始采用session.getConfig().setUseReadOperation(true);结合readFutrue.read()同步读取客户端的返回值;使得在发送消息的时候就等候消息返回，但是这种方法并不好，session.getConfig().setUseReadOperation(true)并不会真正的同步。  
以下是之前实现的同步读取返回数据的方法

//设置为需要读取数据
    session.getConfig().setUseReadOperation(true);
    WriteFuture future = session.write(protocolMsgVO);
    //等待数据发送完成
    future.awaitUninterruptibly();
    if(future.isWritten()) {
    //等待读取数据
    ReadFuture readFuture = session.read(); 
    BaseProtocolMsgVO msgReceive = null;
    // 等待数据读取完成 等待20s 如果没有返回 关掉连接           
    if(readFuture.awaitUninterruptibly(20, TimeUnit.SECONDS)) {             
    msgReceive = (BaseProtocolMsgVO) readFuture.getMessage();
    } else {                
    
    log.info("重发次数：" + 1); 
    session.write(protocolMsgVO); 
    if(readFuture.awaitUninterruptibly(20, TimeUnit.SECONDS)) {             
    msgReceive = (BaseProtocolMsgVO) readFuture.getMessage();
    } else {                    
    
    log.info("重发次数：" + 2); 
    session.write(protocolMsgVO); 
    if(readFuture.awaitUninterruptibly(20, TimeUnit.SECONDS)) {                 
    msgReceive = (BaseProtocolMsgVO) readFuture.getMessage(); 
    } else {                    
    log.info("重发超过三次,断开连接"); 
    session.getConfig().setUseReadOperation(false); // 设置为不需要读取数据 
    // 关掉连接                 
    session.closeNow(); 
    throw new BusiException("读取设备响应数据超时"); }
    }}}

BaseProtocolMsgVO为自定义的消息实体类。目前想到的一个好方法是采用可重入锁ReenLock和Condition去等待通知。  
1.声明一个实体类，存放一个可重入锁，以及Condition和返回消息

public class ReenLockVO {
    
        private ReentrantLock lock;
        private Condition condition;
        private BaseProtocolMsgVO baseProtocolMsgVO;
    
        public ReenLockVO() {
            this.lock = new ReentrantLock();
            this.condition = lock.newCondition();
        }
        }

实例化的时候加载重入锁以及condition；  
2.发送消息时，

ReenLockVO lockVOs = new ReenLockVO();
    lockVO.put(deviceCode + protocolMsgVO.getMsgNo(), lockVOs);
    lockVOs.getLock().lock();
    session.write(protocolMsgVO);
    try {
        if(lockVOs.getCondition().await(20, TimeUnit.SECONDS)) {
            rBaseProtocolMsgVO = lockVOs.getBaseProtocolMsgVO();
        } else {
            log.info("重发次数：" + 1 + "重发消息：" + protocolMsgVO.toByteString());
            session.write(protocolMsgVO);
            if(lockVOs.getCondition().await(20, TimeUnit.SECONDS)) {
                rBaseProtocolMsgVO = lockVOs.getBaseProtocolMsgVO();
            } else {
                log.info("重发次数：" + 2 + "重发消息：" + protocolMsgVO.toByteString());
                session.write(protocolMsgVO);
                if(lockVOs.getCondition().await(20, TimeUnit.SECONDS)) {
                    rBaseProtocolMsgVO = lockVOs.getBaseProtocolMsgVO();
                } else {
                    log.info("重发次数超过三次，断开连接！");
                    session.closeNow();
                }
            }
        }
    } catch(InterruptedException e) {
        e.printStackTrace();
    } finally {
        if(lockVO.contains(deviceCode + protocolMsgVO.getMsgNo())) {
            lockVO.remove(deviceCode + protocolMsgVO.getMsgNo());
        }
        lockVOs.getLock().unlock();
    }

lockVO为ConcurrentHashMap,key存放客户端编号以及报文编号,value就放可重入锁的VO  
private static ConcurrentHashMap

lockVO.get(responseMsg.getDeviceCode() + msgVO.getMsgNo()).setBaseProtocolMsgVO(msgVO);
    lockVO.get(responseMsg.getDeviceCode() + msgVO.getMsgNo()).getLock().lock();
    try {
        lockVO.get(responseMsg.getDeviceCode() + msgVO.getMsgNo()).getCondition().signal();
    } catch(Exception e) {
        log.error(e.getStackTrace());
    } finally {
        lockVO.get(responseMsg.getDeviceCode() + msgVO.getMsgNo()).getLock().unlock();
    }

收到消息时给重入锁的VO设置消息，并通知读取，这样就实现了同步读取消息以及消息重发的机制。  
可重入锁和Condition的结合使用在消息队列里被广泛使用，这里也是仿照消息队列实现的通知等待。

这样写有什么问题的话请指教~！！