ActiveMQ--NIO协议配置

一时失言乱红尘 2021-08-13 15:53 379阅读 0赞

### NIO协议配置 ###

修改配置文件activemq.xml:

如果您不特别指定ActiveMQ的网络监听端口，那么这些端口将使用BIO网络IO模型。（OpenWire、STOMP、AMQP…）。

所以为了首先提高单节点的网络吞吐性能，我们需要明确指定Active的网络IO模型，如下所示：

RUI格式头以"nio"开头，表示这个端口使用以TCP协议为基础的NIO网络IO模型。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70]

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 1]

### 生产者 ###

package pers.zhang.activemq;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    import javax.jms.*;
    
    public class JmsProduce { 
    
        //public static final String ACTIVEMQ_URL = "tcp://localhost:61616";
        public static final String ACTIVEMQ_URL = "nio://localhost:61618";
        public static final String QUEUE_NAME = "queue01";
    
        public static void main(String[] args) throws JMSException { 
            //1.创建连接工厂,按照给定的url地址，采用默认用户名和密码
            ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
            //2.通过连接工厂，获得连接connection，并启动访问
            Connection connection = activeMQConnectionFactory.createConnection();
            connection.start();
            //3.创建回话session
            //两个参数 1.事务 2.签收
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            //4.创建目的地（具体是队列还是主题topic）
            Queue queue = session.createQueue(QUEUE_NAME);
            //创建消息的生产者
            MessageProducer messageProducer = session.createProducer(queue);
            //通过使用messageProducer生产3条消息发送到MQ的队列里面
            for (int i = 0; i < 3; i++) { 
                //7.创建消息
                TextMessage textMessage = session.createTextMessage("msg---" + i);//可以理解为字符串
                //8.通过messageProducer发送给MQ
                messageProducer.send(textMessage);
            }
            //9.关闭资源
            messageProducer.close();
            session.close();
            connection.close();;
            System.out.println("消息发布到MQ完成！");
        }
    }

### 消费者 ###

package pers.zhang.activemq;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    import javax.jms.*;
    import java.io.IOException;
    
    public class JmsConsumer { 
    
        //public static final String ACTIVEMQ_URL = "tcp://localhost:61616";
        public static final String ACTIVEMQ_URL = "nio://localhost:61618";
        public static final String QUEUE_NAME = "queue01";
    
        public static void main(String[] args) throws JMSException, IOException { 
            //1.创建连接工厂,按照给定的url地址，采用默认用户名和密码
            ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
            //2.通过连接工厂，获得连接connection，并启动访问
            Connection connection = activeMQConnectionFactory.createConnection();
            connection.start();
            //3.创建回话session
            //两个参数 1.事务 2.签收
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            //4.创建目的地（具体是队列还是主题topic）
            Queue queue = session.createQueue(QUEUE_NAME);
            //5.创建消费者
            MessageConsumer messageConsumer = session.createConsumer(queue);
    
            //第一种方式：receive（）
            // while(true){ 
            // TextMessage textMessage = (TextMessage) messageConsumer.receive();
            // if(textMessage != null){ 
            // System.out.println("*****消费者接收到消息：" + textMessage.getText());
            // }else{ 
            // break;
            // }
            // }
            // messageConsumer.close();
            // session.close();
            // connection.close();
    
    
    
            //第二中方式：监听
            messageConsumer.setMessageListener(new MessageListener() { 
                @Override
                public void onMessage(Message message) { 
                    if(message != null && message instanceof TextMessage){ 
                        TextMessage textMessage = (TextMessage) message;
                        try { 
                            System.out.println("监听器接收消息：" + textMessage.getText());
                        } catch (JMSException e) { 
                            e.printStackTrace();
                        }
                    }
                }
            });
            //保证控制台运行
            System.in.read();
            messageConsumer.close();
            session.close();
            connection.close();
        }
    }

# NIO增强 #

使一个端口既支持NIO网络IO模型，又支持多个协议。

使用auto关键字：  
使用"+"号来为端口设置多种特性：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 2]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70]: /images/20210813/6e33207127df4a809de99bd8d423f00a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 1]: /images/20210813/e3a1607c0e564d11a4259c39a655159d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 2]: /images/20210813/66824d60ee814f5a83e98dbfcdef8aed.png