mina框架及LinkedBlockingQueue实现阻塞收发

以你之姓@ 2022-10-02 11:55 269阅读 0赞

mina框架实现socket发送，先发送，后阻塞。接收则无限制。

配置：

@Component
    public class BoxTcpServer {
    	
    	private static final Logger log = LoggerFactory.getLogger(LoggerAspect.class);
    	
    	@Autowired
    	public BoxTcpServerHandler boxTcpServerHandler;
    	
    	@Value("8087")
    	private int socketPort;
        @Value("1000")
        private int max;
        
        @PostConstruct
        public void startServer() throws IOException {
            NioSocketAcceptor nioSocketAcceptor = new NioSocketAcceptor();
    
            // 设置最大客户端连接数
            nioSocketAcceptor.setBacklog(max);
    
            // 设置接收单行文本数据最大大小为1M
            TextLineCodecFactory textLineCodecFactory = new TextLineCodecFactory(Charset.forName("GBK"));
            textLineCodecFactory.setDecoderMaxLineLength(1024 * 1024);
            textLineCodecFactory.setEncoderMaxLineLength(1024 * 1024);
            // 过滤器
            nioSocketAcceptor.getFilterChain().addLast("logger", new LoggingFilter());
            nioSocketAcceptor.getFilterChain().addLast("codec", new ProtocolCodecFilter(textLineCodecFactory));
    
            // 处理器
            nioSocketAcceptor.setHandler(boxTcpServerHandler);
    
            int[] ports = new int[]{socketPort};
            String hostAddress = InetAddress.getLocalHost().getHostAddress();
            // 绑定端口并启动服务
            for (int port : ports) {
                nioSocketAcceptor.bind(new InetSocketAddress(port));
            }
    
            log.info("TCP服务启动，接收box数据！ip：[{}]",
                    hostAddress, ports.length);
        }
    
    }

逻辑如下：

@Component
    public class BoxTcpServerHandler extends IoHandlerAdapter {
    	
    	private static final Logger log = LoggerFactory.getLogger(LoggerAspect.class);
    	
    	@Autowired
    	private SysDAO sysDAO;
    	
    	private Map<String, IoSession> ipMap = new ConcurrentHashMap<>();
    	private Map<IoSession, LinkedBlockingQueue<String>> recvMap = new ConcurrentHashMap<>();
    	
    
    	@Override
    	public void exceptionCaught(IoSession session, Throwable cause)
    			throws Exception {
    		// TODO Auto-generated method stub
    		super.exceptionCaught(session, cause);
    	}
    
    	public String getMsg(String ip) {
    		IoSession ioSession = ipMap.get(ip);
    		String msg = null;
    		try {
    			msg = recvMap.get(ioSession).poll(20,TimeUnit.SECONDS);
    		} catch (InterruptedException e) {
    			// TODO Auto-generated catch block
    			return "1";
    		}
    		if(msg == null){
    			return "1";
    		}
    		return msg;
    		
    	}
    	@Override
    	public void messageReceived(IoSession session, Object message)
    			throws Exception {
    		String string = message.toString();
    		log.info("本次接受的数据：{}",string.trim());
    		if (string.startsWith("report:")) {
    			String str = string.substring(7);
    			reportEvent(str);
    		}else {
    			recvMap.get(session).offer(string);
    		}
    	}
    
    	public String sendMsg(String ip, String msg){
    		IoSession ioSession = ipMap.get(ip);
    		if(ioSession == null){
    			return "1";
    		}
    		log.info("本次发送的数据：{}",msg.trim());
    		ioSession.write(msg);
    	
    		return getMsg(ip);
    	}
    
    	@Override
    	public void sessionClosed(IoSession session) throws Exception {
    		InetSocketAddress inetSocketAddress = (InetSocketAddress)session.getRemoteAddress();
    		String ip = inetSocketAddress.getHostString();
    		ipMap.remove(ip);
    		log.info("{}已离线",ip);
    	}
    
    	@Override
    	public void sessionCreated(IoSession session) throws Exception {
    		InetSocketAddress inetSocketAddress = (InetSocketAddress)session.getRemoteAddress();
    		String ip = inetSocketAddress.getHostString();
    		ipMap.put(ip, session);
            recvMap.put(session,new LinkedBlockingQueue<String>(1));
    		log.info("{}已上线",ip);
    	}
    
    	@Override
    	public void sessionIdle(IoSession session, IdleStatus status)
    			throws Exception {
    		// TODO Auto-generated method stub
    		super.sessionIdle(session, status);
    	}
    
    	@Override
    	public void sessionOpened(IoSession session) throws Exception {
    		// TODO Auto-generated method stub
    		super.sessionOpened(session);
    	}
    }