问ActiveMQ中的大量SSL客户端
EN

Stack Overflow用户

提问于 2015-11-26 12:28:19

回答 1查看 1.3K关注 0票数 2

我在使用故障转移协议、安全套接层协议( SSL )和simpleAuthenticationPlugin的ActiveMQ中遇到了大量(1000)客户端的问题。运行了一段时间后，brocker就停止了，我一直在寻找日志文件中的一些错误，但是我没有找到任何东西--它是空的。

服务器-ubuntu15.04，java版"1.7.0_80“基本架构信息

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

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

代码如下：

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