Disruptor 异常卡死 bug

女爷i 2022-02-13 15:51 540阅读 0赞

# bug原因 #

在发送消息时候，需要重写这个接口，而假如重写的方法里抛出异常且未捕获处理，则会导致程序卡死，不会往下执行  
disrupt 源码

package com.lmax.disruptor;
    
    public interface EventTranslatorVararg<T> {
        void translateTo(T var1, long var2, Object... var4);
    }

--------------------

## 问题代码 ##

public class Main { 
        public static void main(String[] args) { 
            Disruptor<Message> disruptor = new Disruptor<>(
                    Message::new,
                    1024,
                    (ThreadFactory) Thread::new);
            disruptor.handleEventsWith((EventHandler<Message>) (message, l, b) -> { 
                System.out.println("处理消息 " + message);
            });
            disruptor.start();
            // 生产
            RingBuffer<Message> ringBuffer = disruptor.getRingBuffer();
    
            for(int i = 0; i<10; i++){ 
                Message message = new Message(String.valueOf(i));
                ringBuffer.publishEvent((m, l) -> 
                	throw new RuntimeException();
                );// 抛出异常不捕获，发现程序不能终止
            }
    
            System.out.println("hi");	//hi 不会输出
            disruptor.shutdown();//关闭 disruptor，方法会堵塞，直至所有的事件都得到处理；
        }
        
    }

## 结果： ##

主线程退出，但程序仍继续运行，未停止

## 解决： ##

在调用时处理异常

public class Main { 
        public static void main(String[] args) { 
            Disruptor<Message> disruptor = new Disruptor<>(
                    Message::new,
                    1024,
                    (ThreadFactory) Thread::new);
            disruptor.handleEventsWith((EventHandler<Message>) (message, l, b) -> { 
                System.out.println("处理消息 " + message);
            });
            disruptor.start();
            // 生产
            RingBuffer<Message> ringBuffer = disruptor.getRingBuffer();
    
            for(int i = 0; i<10; i++){ 
                Message message = new Message(String.valueOf(i));
                ringBuffer.publishEvent((m, l, m2) -> { 
                    try { 
                        throw new RuntimeException();
                    }catch (Exception e){ 
                        e.printStackTrace();
                    }
                });// 处理异常，发现程序打印异常且可以结束
            }
    
            System.out.println("hi");
            disruptor.shutdown();//关闭 disruptor，方法会堵塞，直至所有的事件都得到处理；
        }
        
    }

## Message.java ##

public class Message { 
        String id;
    
        public Message(String id) { 
            this.id = id;
        }
    
        public Message() { 
        }
    
        public String getId() { 
            return id;
        }
    
        public void setId(String id) { 
            this.id = id;
        }
    
        @Override
        public String toString() { 
            return "Message{" +
                    "id='" + id + '\'' +
                    '}';
        }
    
    }