【重大故障】消息堆积、消息丢失？

冷不防 2023-03-03 06:23 9阅读 0赞

## 前言 ##

总算尘埃落定了，记录下这两天的血泪史吧。算是教训也算是经验了。

原始需求：接收第三方推送过来的数据，进行解析入库。一个很简单的需求，当时对端提供的是 socket 进行推送，每条消息的以 `<START>`开始，`<END>`结束，但是他们推送的是一直往流里面写数据，所以我们需要分割出消息，然后再对消息进行解析入库。

如下图：是我通过

nc -l port -> nc.log

监听获取到对端推送的消息日志，简化后的消息。

![\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-5N5OddhX-1596090024114)(【重大故障】消息丢失？.assets/image-20200730111023436.png)\]][img-5N5OddhX-1596090024114_.assets_image-20200730111023436.png]

最开始采用原生的 socket ,进行接收。

直接创建的 一个 serverSocket 并轮寻监听，接收对端推送的数据，对端实际上只会创建一个连接，然后通过这个连接一直推送数据，所以这里的 accept 方法不会阻塞。

public void start(int port) throws Exception{
    
            ServerSocket serverSocket = new ServerSocket(port);
            System.out.println("socket start on port："+port);
    
            init();
            Socket accept = serverSocket.accept();
            //监听轮寻
            while (true){
                try {
                    if(accept.isClosed()||accept.isOutputShutdown()){
                        log.info("创建新的accept....");
                        accept = serverSocket.accept();
                    }
                    //接收到请求
                    log.info("新一轮接收。。。");
                    process(accept);
                }catch (Exception e){
                    log.error("{}",e);
                }
    
            }
    
        }

核心在 process 方法中。通过 BufferedReader 接收 scoket 输入流中的数据。一行行的读取， 进行消息的截取。然后将消息交给线程池去执行解析操作。这里接着读取下一条消息，为了方便，我还特意将消息打印了出来。

private void process(Socket accept)throws Exception {
    
            StringBuffer messageBuffer=new StringBuffer();
    
            BufferedReader br=new BufferedReader(new InputStreamReader(accept.getInputStream()));
    
            String info=null;
            while(!((info=br.readLine())==null)){
                try {
    
                    //log.info(info);
                    if(info.equals("<END><START>")){
                        String message=messageBuffer.toString()+"<END>";
                        exexutor(message);
                        messageBuffer=new StringBuffer("<START>\n");
                        log.info("begin-----------------------------");
                        log.info(message);
                        log.info("end-----------------------------");
                    }else if(info.equals("<END>")){
                        String message=messageBuffer.toString()+"<END>";
                        exexutor(message);
                        messageBuffer=new StringBuffer();
                        log.info("begin***************************");
                        log.info(message);
                        log.info("end***************************");
                    }else {
                        messageBuffer.append(info+"\n");
                    }
                }catch (Exception e){
                    log.error("{}",e);
    
                }
            }
        }

线程池的配置：

//创建一个线程池
        private void crateTreadPool() {
            // 定义一个线程池
            int corePoolSize = 20;
            int maximumPoolSize =100;
            long keepAliveTime = 100L;
            TimeUnit unit = TimeUnit.SECONDS;
            BlockingQueue<Runnable> workQueue = new ArrayBlockingQueue<>(500);
            ThreadFactory threadFactory = Executors.defaultThreadFactory();
            RejectedExecutionHandler handler = new ThreadPoolExecutor.AbortPolicy();
    
            threadPoolExecutor=new ThreadPoolExecutor(
                    corePoolSize,
                    maximumPoolSize,
                    keepAliveTime,
                    unit,
                    workQueue,
                    threadFactory,
                    handler
            );
        }

结果放到服务器上运行，刚开始都好好的，过了一段时间，就会出现日志打不赢了，延迟了好几分钟，并且在持续增长。这下就知道了上面从 BufferedReader 取消息处理，处理的速度跟不上推送的速度，导致缓冲区越来越大，速度也来越慢，最终导致了服务卡死。

然后了两步改进，采用 nio 接收，以及不印数据接收的报文log。

/**
         * 在初始化中要做一下如下操作：
         * 1、开启多路复用器
         * 2、开启服务通道
         * 3、设置为非阻塞
         * 4、绑定端口
         * 5、标记选择器状态为可接受，表示可以接受通道注册到选择器上。
         */
        public void init(int port) {
            try {
                System.out.println("init......"+port);
                //开启多路复用器
                selector = Selector.open();
                //开启通道
                ServerSocketChannel channel = ServerSocketChannel.open();
                //设置为非阻塞
                channel.configureBlocking(false);
                //绑定端口
                channel.bind(new InetSocketAddress(port));
                //标记选择器状态为可接受
                /**
                 * SelectionKey.OP_ACCEPT   —— 接收连接继续事件，表示服务器监听到了客户连接，服务器可以接收这个连接了
                 * SelectionKey.OP_CONNECT  —— 连接就绪事件，表示客户与服务器的连接已经建立成功
                 * SelectionKey.OP_READ     —— 读就绪事件，表示通道中已经有了可读的数据，可以执行读操作了（通道目前有数据，可以进行读操作了）
                 * SelectionKey.OP_WRITE    —— 写就绪事件，表示已经可以向通道写数据了（通道目前可以用于写操作）
                 */
                channel.register(selector, SelectionKey.OP_ACCEPT);
    
                System.out.println("init finished......");
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        public void process(){
            //轮寻
            while(true){
                try {
                    //通道选中的个数，至少有一个通道被选中才会执行。
                    selector.select();
                    Set<SelectionKey> selectionKeys = selector.selectedKeys();
                    Iterator<SelectionKey> iterator = selectionKeys.iterator();
                    while(iterator.hasNext()){
                        //获取key
                        SelectionKey key = iterator.next();
                        //从迭代器中取出这个key
                        iterator.remove();
                        //执行
                        ServerHandlerNio serverHandlerNio = new ServerHandlerNio(selector, key, realtimeFailureMessageParse);
                        threadPoolExecutor.execute(serverHandlerNio);
                    }
    
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }

public class ServerHandlerNio extends Thread{
    
        //定义一个选择器
        private Selector selector;
    
        private RealtimeFailureMessageParse realtimeFailureMessageParse;
    
        private static Logger log = LoggerFactory.getLogger(ServerHandlerNio.class);
    
    
        //定义读写缓冲区
        private ByteBuffer readBuffer=ByteBuffer.allocateDirect(1024*100);
    
        private SelectionKey key;
    
        public ServerHandlerNio(Selector selector, SelectionKey key,RealtimeFailureMessageParse realtimeFailureMessageParse){
            this.selector=selector;
            this.key=key;
            this.realtimeFailureMessageParse=realtimeFailureMessageParse;
        }
    
        @Override
        public void run() {
            try {
                process();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    
    
        private void process() throws Exception {
            try{
                //判断key 是否有效
                if (key.isValid()) {
                    try {
                        //6.判断是否可以连接
                        if (key.isAcceptable()) {
                            accept(key);
                        }
                    } catch (CancelledKeyException e) {
                        //出现异常断开连接
                        key.cancel();
                    }
    
    
                    try {
                        //7.判断是否可读
                        if (key.isReadable()) {
                            read(key);
                        }
                    } catch (CancelledKeyException e) {
                        //出现异常断开连接
                        key.cancel();
                    }
    
                    try {
                        //8.判断是否可写
                        if (key.isWritable()) {
                            write(key);
                        }
                    } catch (CancelledKeyException e) {
                        //出现异常断开连接
                        key.cancel();
                    }
                }
            }catch (ClosedChannelException e){
                key.cancel();
            }
        }
    
        /**
         *给通道中写数据。从buffer 中给通道写数据。
         * @param key
         */
        private void write(SelectionKey key) throws IOException {
            SocketChannel channel = (SocketChannel) key.channel();
            //重新标记为可读
            channel.register(selector,SelectionKey.OP_READ);
        }
    
        /**
         *使用通道读取数据。主要就是将通道中的数据读取到读缓存中。
         * @param key
         */
        private void read(SelectionKey key) throws IOException {
            readBuffer.clear();
    
            SocketChannel channel = (SocketChannel)key.channel();
    
            int len = channel.read(readBuffer);
    
            //如果通道没有数据
            if(len==-1){
                //关闭通道
                key.channel().close();
                //关闭key
                key.cancel();
                return;
            }
    
            //Buffer中有游标,游标不会重置,需要我们调用flip重置. 否则读取不一致
            readBuffer.flip();
            //创建有效字节长度数组
            byte[] bytes = new byte[readBuffer.remaining()];
            //读取buffer中数据保存在字节数组
            readBuffer.get(bytes);
    
            String clientMessage = new String(bytes, "UTF-8");
            //System.out.println("accepted client message are "+clientMessage);
    
            onMessage(clientMessage);
    
            //注册通道,标记为可读操作
            channel.register(selector,SelectionKey.OP_WRITE);
        }
    
    
        public void onMessage( String message) {
            //处理消息
            if(messageHandle(message)){
                //成功
                log.info("消息接收成功");
            }else{
                //失败
                log.error("消息接收失败：数据格式不正确！");
            }
        }
    
        /**
         * 处理消息
         * @param message
         * @return
         */
        private boolean messageHandle(String message) {
            try {
                return realtimeFailureMessageParse.parse(message);
            } catch (Exception e) {
                e.printStackTrace();
            }
            return false;
        }
    
    
    
        /**
         *设置通道接受客户端数据，并设置通道为可读。
         * @param key
         */
        private void accept(SelectionKey key) throws IOException {
            //1.获取通道
            ServerSocketChannel socketChannel = (ServerSocketChannel) key.channel();
            //阻塞方法，获取客户端的请求
            SocketChannel channel = socketChannel.accept();
            if(channel!=null){
                //设置为非阻塞
                channel.configureBlocking(false);
                //设置对应客户端的通道标记,设置次通道为可读时使用
                channel.register(selector,SelectionKey.OP_READ);
            }
        }
    
    
    
    }

采用 nio 的方式，代码逻辑写得太复杂了，仅仅是接收就这么费劲，我简单的测了一下，发现还是不行。

继续改进，采用了 netty 进行接收处理。

public void start(String host,int port) {
            //负责监听连接
            EventLoopGroup boss = new NioEventLoopGroup();
            ServerBootstrap boot = new ServerBootstrap();
            try {
                boot.group(boss)
                        .option(ChannelOption.SO_REUSEADDR, true)
                        .option(ChannelOption.SO_BACKLOG,1024)
                        .childOption(ChannelOption.TCP_NODELAY,true)
                        .channel(NioServerSocketChannel.class)
                        .localAddress(host, port)
                        .childHandler(
                                new ChannelInitializer<NioSocketChannel>() {
    
                                    @Override
                                    protected void initChannel(NioSocketChannel socketChannel)
                                            throws Exception {
                                        log.info(
                                                "有一个新的客户端连接到服务器,ip={},port={}",
                                                socketChannel.remoteAddress().getHostName(),
                                                socketChannel.remoteAddress().getPort());
                                        ByteBuf byteBuf = Unpooled.copiedBuffer("<END>", StandardCharsets.UTF_8);
                                        socketChannel.pipeline().addLast(new DelimiterBasedFrameDecoder(1024*50, byteBuf));
                                        socketChannel.pipeline().addLast(new StringDecoder(StandardCharsets.UTF_8));
                                        socketChannel.pipeline().addLast(new IdleStateHandler(4,2,0, TimeUnit.SECONDS));
                                        socketChannel.pipeline().addLast(new MyNettyHandle());
                                    }
                                });
                ChannelFuture channel = boot.bind().sync();
                channel.channel().closeFuture().sync();
            } catch (Exception e) {
                log.error("服务器运行中发生异常!", e);
            } finally {
                boss.shutdownGracefully();
            }
        }

自己的处理器代码如下：

@Component
    public class MyNettyHandle extends ChannelInboundHandlerAdapter {
    
    
        private static Logger log = LoggerFactory.getLogger(SocketServer.class);
    
    
        private static ExecutorService threadPoolExecutor= CrateTreadPool.crateTreadPool() ;//= Executors.newFixedThreadPool(100);
    
    
    
        private static RealtimeFailureMessageParse realtimeFailureMessageParse;
    
        @Autowired
        public void setRealtimeFailureMessageParse(RealtimeFailureMessageParse realtimeFailureMessageParse) {
            this.realtimeFailureMessageParse = realtimeFailureMessageParse;
        }
    
    
    
    
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            String message = (String) msg+"<END>";
            //int index = message.indexOf("EventTime");
            //log.info("接收到客户端数据:{}", message.substring(index,index+30));
            RequestProcessor requestProcessor=new RequestProcessor(message,realtimeFailureMessageParse);
            threadPoolExecutor.execute(requestProcessor);
        }
    
        private void sendMsg(ChannelHandlerContext ctx, String reply) {
            reply = reply + "\r\n";
            ByteBuf byteBuf = Unpooled.copiedBuffer(reply.getBytes(StandardCharsets.UTF_8));
            ctx.writeAndFlush(byteBuf);
        }
    
        private boolean check(String msg) {
            return (msg.startsWith("<START>") && msg.endsWith("<END>"));
        }
    
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            ctx.close();
            log.error("当前客户端已经断开连接!{}",cause.getMessage());
        }
    }

我刚开始是开日志跑的，还是发现有消息堆积，后来我通过测试，不进行任何的处理，接收消息后，我就打印核心消息进行对比。

int index = message.indexOf("EventTime");
    log.info("接收到客户端数据:{}", message.substring(index,index+30));

发现速度是可以跟得上的，所以这边将不必要的日志都进行清理，不再打印日志。

但是这样做了之后我们发现，数据解析后入库的操作还是不行，每条消息都需要进行插入或者更新操作。这样效率就太低了，也会导致消息堆积处理不完。

后面采用自定义队列，进行批量入库或者更新操作。

自定义队列方案：采用 ConcurrentHashMap 和 Collections.synchronizedList 实现线程安全。

public static Map<String, List<RealtimeFailure>> realtimeFailureUpdateMap=new ConcurrentHashMap<>();
    Collections.synchronizedList(new ArrayList<>())

接收到消息后，交给线程池分派线程进行解析封装成入库对象，每分钟生成一个list。

将对象存入到当前 list 中。如下：

String key = MapPool.simpleDateFormat.format(new Date());
    List<RealtimeFailure> orDefault = MapPool.realtimeFailureInsertMap.getOrDefault(key, Collections.synchronizedList(new ArrayList<>()));
    orDefault.add(realtimeFailure);
    MapPool.realtimeFailureInsertMap.put(key,orDefault);

当然插入和更新的队列是分开的。

然后再开一个单独的线程，一直从队列中取最久的一条消息进行读取批量入库。

public  void process() {
            Object[] objects = MapPool.realtimeFailureInsertMap.keySet().toArray();
            if(objects.length>1){
                try{
                    Arrays.sort(objects);
                    Object key = objects[0];
                    List<RealtimeFailure> realtimeFailures = MapPool.realtimeFailureInsertMap.get(key);
                    //移除
                    MapPool.realtimeFailureInsertMap.remove(key);
                    //入库
                    log.info("begin入库:"+key+" size:"+realtimeFailures.size());
                    realtimeFailureService.insertSelective(realtimeFailures);
                    log.info("入库完成 ："+key+" size:"+realtimeFailures.size());
    
                    if(MapPool.realtimeFailureUpdateMap.containsKey(key)){
                        List<RealtimeFailure> realtimeFailures1 = MapPool.realtimeFailureUpdateMap.get(key);
                        MapPool.realtimeFailureUpdateMap.remove(key);
                        realtimeFailureService.updateSelective(realtimeFailures1);
                        log.info("更新完成 ："+key+" size:"+realtimeFailures1.size());
    
                    }else {
                        Thread.sleep(100);
                }
                }catch (Exception e){
                    log.error("{}",e);
                }
            }
        }
        
        public void run() throws Exception {
            while (true){
                process();
            }
    
        }

批量入库和跟新的话，采用mybatis 的 foreach 就可以实现。

最终效果：基本上是一分钟处理一次，处理速度也可以，解析、插入和更新 耗时 1 秒左右。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3NzkwMDEx_size_16_color_FFFFFF_t_70]

存在的问题：虽然现在问题是解决了，但是现在整个系统都是非常脆弱的，服务停掉或者重启就会丢失消息。这个主要也是 socket 消息机制的问题。对端推送的消息是实时的，不会重复推送。然后我们这边停掉服务，正在处理的消息也会丢失。所以必然会存在消息丢失的情况。

要解决这种问题，必须得用消息中间件了，这个就不是一时半会能弄好的。不过打算用 kafka 进行改进，对端对推送消息，我们从 kafka 中消费消息，基本上不会存在消息丢失的情况，但是还是要注意消费的速度，必须大于他们推送的速度，不然就会造成消息的堆积。

[img-5N5OddhX-1596090024114_.assets_image-20200730111023436.png]: /images/20230208/af1e06cc007b4a8eb9f3d94e1f4a9c5c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3NzkwMDEx_size_16_color_FFFFFF_t_70]: /images/20230208/37216d1295c34cb39543d98dc4e4eaee.png