java - ActiveMQ 中的大量 SSL 客户端

Question

我在使用故障转移协议(protocol)、SSL (NIO) 和 simpleAuthenticationPlugin 的 ActiveMQ 中遇到大量 (1000) 个客户端的问题。运行一段时间后，broker 就停止了。我一直在寻找日志文件中的一些错误，但我什么也没找到——它是空的。

服务器 - ubuntu 15.04，java 版本“1.7.0_80”基本架构信息

我有一个分布式系统。它由 C# 客户端(节点)和一个 Java 客户端(管理器)组成。我正在使用安装在管理器工作的同一主机上的 ActiveMQ 5.12.1。通信仅发生在 C# 客户端和管理器之间。 C# 客户端之间不通信。

每个客户端(C# 和 java)使用一个队列进行监听。当管理器想要向节点发送消息时，它会使用一个唯一的队列来完成。当一个节点想要与管理器通信时，它会向管理器的队列发送一条消息。管理器每分钟轮询每个节点。 Manager有2个线程，一个用来接收消息。二次拉取节点信息

代码如下:

ActiveMQConnectionFactory   connectionFactory = new ActiveMQConnectionFactory("failover://(nio+ssl://localhost:61617)");
        connectionFactory.setUserName("user");
        connectionFactory.setPassword("password");
        Connection connection = connectionFactory.createConnection();
        connection.start();
        connection.setExceptionListener(this);
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        Destination destination = session.createQueue("manager");
        MessageConsumer consumer = session.createConsumer(destination);
        Message message= null;
        while (!stop) {
            try{
            message = consumer.receive();
            if (message instanceof TextMessage) {
                TextMessage textMessage = (TextMessage) message;
                String text = textMessage.getText();
                processMessage(text);
            }
            }
            catch (Exception e){
                Logger.getInstance().log("Exception in message receive loop continue");
            }
        }

发送消息的管理器代码:

ActiveMQConnectionFactory   connectionFactory = new ActiveMQConnectionFactory("failover://(nio+ssl://localhost:61617)");
        connectionFactory.setUserName("user");
        connectionFactory.setPassword("password");
        Connection connection = connectionFactory.createConnection();
        connection.start();
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        while(!stop) {
            for (everynode) {
                    Destination destination = session.createQueue(uniqueNodeQueue);
                    MessageProducer producer = session.createProducer(destination);
                    producer.setDeliveryMode(DeliveryMode.PERSISTENT);
                    TextMessage message = session.createTextMessage("in");
                    producer.send(message);
                    producer.close();
            }
            Thread.sleep(time);
        }

        // Clean up
        session.close();
        connection.close();

C#客户端使用Apache.NMS.ActiveMQ；

brokerUri="failover://(ssl://" +Server + ":61617?transport.acceptInvalidBrokerCert=true";
public NonDurableQueueSubscriber(string queueName, string brokerUri)
    {
        this.queueName = queueName;
        ConnectionFactory cf = new ConnectionFactory(brokerUri);
        cf.Password = "password";
        cf.UserName = "user";
        this.connectionFactory = cf;
        this.connection = this.connectionFactory.CreateConnection();
        if (this.connection.IsStarted)
            connection.Stop();
        this.connection.Start();
        this.session = connection.CreateSession();
        ActiveMQQueue queue = new ActiveMQQueue(queueName);
        this.consumer = this.session.CreateConsumer(queue, "2 > 1", false);
        this.consumer.Listener += new MessageListener(OnMessage);
    }
public void OnMessage(IMessage message)
    {
        ITextMessage textMessage = message as ITextMessage;
        if (this.OnMessageReceived != null)
        {
            this.OnMessageReceived(textMessage.Text);
        }
    }
static void onMessageReceived(string message)
    {
        NonDurableQueuePublisher mypublisher = new NonDurableQueuePublisher(queueManager, activemqBrokerUrl);
            mypublisher.SendMessage(info());
            mypublisher.Dispose();
    }
public NonDurableQueuePublisher(String queueName, string brokerUri)
    {
        this.queueName = queueName;
        ConnectionFactory cf = new ConnectionFactory(brokerUri);
        cf.Password = "password";
        cf.UserName = "user";
        this.connectionFactory = cf;
        this.connection = connectionFactory.CreateConnection();
        this.connection.Start();
        this.session = connection.CreateSession();
        ActiveMQQueue queue = new ActiveMQQueue(queueName);
        this.producer = session.CreateProducer(queue);
    }
public void SendMessage(String msg)
    {
        if (!this.isDisposed)
        {
            ITextMessage txtMessage = session.CreateTextMessage(msg);
            //txtMessage.NMSPersistent = false;
            producer.Send(txtMessage);
        }
        else
        {
            throw new ObjectDisposedException(this.GetType().FullName);
        }
    }

activeMQ.xml 配置activeMQ代理

<beans
  xmlns="http://www.springframework.org/schema/beans"
  xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd

http://activemq.apache.org/schema/core 

http://activemq.apache.org/schema/core/activemq-core.xsd">


<bean class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer">
    <property name="locations">
        <value>file:${activemq.conf}/credentials.properties</value>
    </property>
</bean>

<bean id="logQuery" class="io.fabric8.insight.log.log4j.Log4jLogQuery"
      lazy-init="false" scope="singleton"
      init-method="start" destroy-method="stop">
</bean>

<broker xmlns="http://activemq.apache.org/schema/core" brokerName="localhost" deleteAllMessagesOnStartup="true" dataDirectory="${activemq.data}">

    <destinationPolicy>
        <policyMap>
          <policyEntries>
            <policyEntry topic=">" >

              <pendingMessageLimitStrategy>
                <constantPendingMessageLimitStrategy limit="1000"/>
              </pendingMessageLimitStrategy>
            </policyEntry>
          </policyEntries>
        </policyMap>
    </destinationPolicy>


    <managementContext>
        <managementContext createConnector="false"/>
    </managementContext>


    <persistenceAdapter>
        <kahaDB directory="${activemq.data}/kahadb"/>
    </persistenceAdapter>

      <systemUsage>
        <systemUsage>
            <memoryUsage>
                <memoryUsage percentOfJvmHeap="70" />
            </memoryUsage>
            <storeUsage>
                <storeUsage limit="100 gb"/>
            </storeUsage>
            <tempUsage>
                <tempUsage limit="50 gb"/>
            </tempUsage>
        </systemUsage>
    </systemUsage>



<plugins>
<simpleAuthenticationPlugin>
<users>
<authenticationUser username="user" password="password" groups="users,admins"/>
</users>
</simpleAuthenticationPlugin>
</plugins>
<sslContext>
        <sslContext keyStore="/ssl/broker.ks"
                    keyStorePassword="password"/>
    </sslContext>
    <transportConnectors>
    <transportConnector name="nio+ssl" uri="nio+ssl://0.0.0.0:61617?maximumConnections=20000&amp;wireFormat.maxFrameSize=104857600"/>

    </transportConnectors>


    <shutdownHooks>
        <bean xmlns="http://www.springframework.org/schema/beans" class="org.apache.activemq.hooks.SpringContextHook" />
    </shutdownHooks>

</broker>

<import resource="jetty.xml"/>

一些我在使用NIO时截取的有趣日志:

2015-11-25 17:20:31,495 | ERROR | Could not accept connection from null: java.io.IOException: javax.net.ssl.SSLException: Inbound closed before receiving peer's close_notify: possible truncation attack? | org.apache.activemq.broker.TransportConnector | ActiveMQ BrokerService[localhost] Task-616
2015-11-25 17:20:31,495 | WARN  | Transport Connection to: tcp://192.168.1.4:51939 failed: javax.net.ssl.SSLException: Inbound closed before receiving peer's close_notify: possible truncation attack? | org.apache.activemq.broker.TransportConnection.Transport | ActiveMQ NIO Worker 151
2015-11-25 17:20:25,493 | ERROR | Could not accept connection from null: java.io.IOException: javax.net.ssl.SSLException: Unsupported record version Unknown-0.0 | org.apache.activemq.broker.TransportConnector | ActiveMQ BrokerService[localhost] Task-787
2015-11-25 17:20:19,309 | ERROR | Could not accept connection from null: java.io.IOException: javax.net.ssl.SSLException: Unsupported record version Unknown-0.0 | org.apache.activemq.broker.TransportConnector | ActiveMQ BrokerService[localhost] Task-669
2015-11-25 17:19:53,051 | WARN  | Transport Connection to: tcp://192.168.1.23:51587 failed: org.apache.activemq.transport.InactivityIOException: Channel was inactive (no connection attempt made) for too (>30000) long: tcp://192.168.1.23:51587 | org.apache.activemq.broker.TransportConnection.Transport | ActiveMQ InactivityMonitor Worker

在 NIO 之前我只尝试过 SSL，结果是一样的 → broker shutdown。

有人有什么想法吗？

Answer 1

您似乎超出了在 activemq.xml 中配置的最大连接数限制。您可以通过 JMX 检查连接数来确认这一点。

您需要研究为什么您的设置会创建如此多的打开连接，以及为什么它没有关闭它们(如果有意的话)。 C# 代码对我来说看起来有点不可靠(在任何地方都没有 close calls)，但由于我看不到它的全部，所以它可能没问题。

这似乎不是 ActiveMQ 问题，而是应用程序问题。