ActiveMQ断开，消费者重连一段时间后进程崩溃退出的问题

柔光的暖阳◎ 2022-11-21 04:16 754阅读 0赞

## 问题 ##

最近遇到一个问题（activemq版本为5.14.5），正常ActiveMQ断开后，因为有因为有心跳检测和重连机制，使用failover方式，消费者其实是会不断的尝试重连，进程应该是一直存在的。但是奇怪的是有的进程的消费者在mq断开后进程会直接挂掉，没有日志输出也不再重连了。  
主进程如下，消费者代码就是正常的设置监听器（MessageListener）的代码，就不贴了。

/**
     * @author ZZJ
     * @description:
     * @date 2020-9-9 9:46
     */
    public class ActiveMQTest {
    
        public static void main(String[] args) {
    
            ActiveMQConsumer activeMQConsumer = new ActiveMQConsumer();
            activeMQConsumer.consume();
        }
    }

## 原因 ##

通过对比发现，只有进程中仅消费者线程独立运行的进程会出现这样的问题，经过一段时间的查询后发现，这是ActiveMQ本身重连机制的一个问题。  
这里引用kimmking的回答：[链接][Link 1]  
简单来说，由于ActiveMQ断开后，socket心跳无法得到回应，一段时间后就会断开，但是正常来说，配置failover方式后，ActiveMQ消费者client会再启动一个线程不断地尝试重连。如

> failover:(tcp://broker1:61616,tcp://broker2:61616)?maxReconnectAttempts=1000&initialReconnectDelay=1000&maxReconnectDelay=300000"

在与ActiveMQ断开后，会不断地从brokerURL的列表中选择broker来进行重连，参数maxReconnectAttempts代表的是最大重连次数，initialReconnectDelay为首次重连时间，代表第一次重连为1s后，之后的重连时间每次都会翻倍，1s，2s，4s，8s这样，maxReconnectDelay为最大重连时间。  
但是该重连线程是守护线程（Daemon Thread），用个比较通俗的比喻，任何一个守护线程都是整个JVM中所有非守护线程的保姆：只要当前JVM实例中尚存在任何一个非守护线程没有结束，守护线程就全部工作；只有当最后一个非守护线程结束时，守护线程随着JVM一同结束工作。守护线程的作用是为其他线程的运行提供便利服务，最典型的应用就是 GC (垃圾回收器)，当没有非守护线程存在时，GC自然也就不需要了，进程也就退出了。  
因此该问题的原因也就很清楚了：如果仅有ActiveMQ消费者线程单独存在，在心跳未接受回应时，socket连接断开，此时虽然存在重连线程，但由于是守护线程，无法单独存在，JVM判断没有非守护的线程存在了，进程自然也就退出了。

## 解决 ##

**第一种方法**

引用kimmking的方法，设置重连线程reconnectTaskFactory为非守护线程

于路径为activemq-all-5.8.0.jar!\\org\\apache\\activemq\\transport\\failover下的FailoverTransport.java 添加

reconnectTaskFactory = new TaskRunnerFactory();
    reconnectTaskFactory.setDaemon(false); // to set daemon=false by kimmking
    reconnectTaskFactory.init();

该方法我并未尝试，因为修改jar包很麻烦，但是应该是可用的。

**第二种方法**  
这种方法就比较简单也比较蠢了，其实就是再多设置一个线程使得非守护线程一直存在，这个方法不是最好的，但是可以根据需要使用，比如设置这个线程去实现定时统计和输出消费者的消费成功数量等功能

/**
     * @author ZZJ
     * @description:
     * @date 2020-9-9 9:46
     */
    public class ActiveMQTest {
        public  static AtomicInteger count = new AtomicInteger();
        public static void main(String[] args) {
    
            ActiveMQConsumer activeMQConsumer = new ActiveMQConsumer();
            activeMQConsumer.consume();
    
            new Thread(() ->{
                while (true) {
                    System.out.println("I'm alive");
                    System.out.println("当前消费了" + count + "条消息");
                    try {
                        Thread.sleep(60000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }).start();
        }
    }

## 总结 ##

为了重现这个问题自己临时搭了一套环境测试，事实证明可以复现的问题就不再是问题的。  
ActiveMQ太老了，吞吐量一般，维护不行，网上的信息也不多还有可能丢消息，除非是很老的项目，不然还是推荐使用RocketMQ之类的消息队列

[Link 1]: https://www.zhihu.com/question/29764784