[编织消息框架][分层模型设计]会话与节点

public class QNode implements IRecycle {
    /**
     * session会话，记录通信层属性
     **/
    private QSession session;
    /**
     * message cb 维护消息回调
     **/
    private QCallbackManager callbackManager;
    /**
     * netty channel
     **/
    private Channel channel;

    private InetSocketAddress address;
}

public class QSession implements IRecycle {

    public static enum SESSION_KEY {
    ALIAS, OP_TIME, ID,
    }

    private long id;
    private Map<String, Object> values = new HashMap<>();
}

/***
 * 响应异步消息 针对业务上的逻辑处理 {@link QNode#send(Object, IQCallback)}
 * 
 * @author solq
 */
public class QCallbackManager implements IRecycle {
    private final static Logger LOGGER = LoggerFactory.getLogger(QCallbackManager.class);

    private final static ScheduledExecutorService pool = PoolUtil.createScheduledPool(QMConfig.getInstance().POOL_CLEAR_MESSAGE_CORE, "message clear");

    private Map<Long, IQCallback<?>> messageRecoreds = Collections.synchronizedMap(new HashMap<>());

    @SuppressWarnings({ "unchecked", "rawtypes" })
    private void buildTask(long sn, IQCallback cb) {
    messageRecoreds.put(sn, cb);
    Future<?> future = pool.schedule(new Runnable() {
        private long _sn = sn;

        @Override
        public void run() {
        IQCallback<?> _cb = messageRecoreds.remove(_sn);
        if (_cb == null) {
            return;
        }
        try {
            _cb.onReceiveError(QCode.MESSAGE_ERROR_TIMEOUT);
        } finally {
            _cb.recycle();
        }

        }
    }, QMConfig.getInstance().NETTY_MESSAGE_CALLBACK_CLEAR_INTERVAL, TimeUnit.MILLISECONDS);
    cb.setFuture(future);
    }

    public <T> IQCallback<T> doSend(QPacket sendPacket, IQCallback<T> cb) {
    if (cb == null) {
        return null;
    }
    sendPacket.setStatus(QPacket.MASK_RESPONSE);
    cb.setSendPacket(sendPacket);
    buildTask(sendPacket.getSn(), cb);
    return cb;
    }

    public void doSendError(QPacket sendPacket, short code) {
    final long key = sendPacket.getSn();
    IQCallback<?> cb = messageRecoreds.remove(key);
    if (cb == null) {
        if (LOGGER.isWarnEnabled()) {
        LOGGER.warn("发送失败 未找到回调 :" + key);
        }
        return;
    }
    try {
        cb.onSendError(code);
    } finally {
        cb.recycle();
    }
    }

    public void doReceiveSucceed(QPacket rePacket) {
    final long key = rePacket.getSn();
    IQCallback<?> cb = messageRecoreds.remove(key);
    if (cb == null) {
        if (LOGGER.isWarnEnabled()) {
        LOGGER.warn("响应成功 未找到回调 :" + key);
        }
        return;
    }
    try {
        short code = rePacket.toCode();
        cb.setCode(code);
        cb.onSucceed(code);
    } finally {
        cb.recycle();
    }
    }

    public void doReceiveError(QPacket rePacket) {
    final long key = rePacket.getSn();
    IQCallback<?> cb = messageRecoreds.remove(key);
    if (cb == null) {
        if (LOGGER.isWarnEnabled()) {
        LOGGER.warn("响应失败 未找到回调 :" + key);
        }
        return;
    }
    try {
        short code = rePacket.toCode();
        cb.setCode(code);
        cb.onReceiveError(code);
    } finally {
        cb.recycle();
    }
    }

    public int getMessageRecoredSize() {
    return messageRecoreds.size();
    }

    @Override
    public void recycle() {
    // 释放所有消息
    messageRecoreds.forEach((sn, cb) -> {
        cb.recycle();
    });
    messageRecoreds.clear();
    }

public abstract class IQCallback<T> implements IRecycle, QResult<T> {
    //响应成功回调
    abstract public void onSucceed(short code);

    // 默认什么也不用做
    public void onSendError(short code) {
    if (LOGGER.isWarnEnabled()) {
        LOGGER.warn("onSendError : {)", code);
    }
    this.code = code;
    }

    // 默认什么也不用做
    public void onReceiveError(short code) {
    if (LOGGER.isWarnEnabled()) {
        LOGGER.warn("onReceiveError : {)", code);
    }
    this.code = code;
    }
}

IQCallback 有三种响应消息处理

1.onSucceed 响应返回成功

2.onSendError 发送时失败

3.onReceiveError 响应返回失败

有的业务非常复杂，如果响应失败了可以根据返回码深度处理