zoukankan      html  css  js  c++  java
  • activemq

    1.下载ActiveMQ

    去官方网站下载:http://activemq.apache.org/

    2.运行ActiveMQ

    解压缩apache-activemq-5.5.1-bin.zip,然后双击apache-activemq-5.5.1inactivemq.bat运行ActiveMQ程序。

    启动ActiveMQ以后,登陆:http://localhost:8161/admin/,系统会自动执行启动过程,当然一般安装失败的情况是没有装JVM环境,启动成功应该是这样

    3、打开浏览器输入http://localhost:8161/admin/默认配置是这个,当然你也可以更改这个配置

    4、至此,服务端启动完毕

    —————————————————————————————————完美分割———————————————————————————————————

    创建一个Queue,命名为FirstQueue

    创建Eclipse项目并运行

    创建project:ActiveMQ-5.5,并导入apache-activemq-5.5.1lib目录下需要用到的jar文件,项目结构如下图所示:

    Sender.java

    package com.newer.tanlei;
    
    import javax.jms.Connection;
    import javax.jms.ConnectionFactory;
    import javax.jms.DeliveryMode;
    import javax.jms.Destination;
    import javax.jms.MessageProducer;
    import javax.jms.Session;
    import javax.jms.TextMessage;
    import org.apache.activemq.ActiveMQConnection;
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    public class Sender {
        private static final int SEND_NUMBER = 5;
    
        public static void main(String[] args) {
            // ConnectionFactory :连接工厂,JMS 用它创建连接
            ConnectionFactory connectionFactory;
            // Connection :JMS 客户端到JMS Provider 的连接
            Connection connection = null;
            // Session: 一个发送或接收消息的线程
            Session session;
            // Destination :消息的目的地;消息发送给谁.
            Destination destination;
            // MessageProducer:消息发送者
            MessageProducer producer;
            // TextMessage message;
            // 构造ConnectionFactory实例对象,此处采用ActiveMq的实现jar
            connectionFactory = new ActiveMQConnectionFactory(
            		ActiveMQConnection.DEFAULT_USER,
            		ActiveMQConnection.DEFAULT_PASSWORD,
                    "tcp://localhost:61616");
            try {
                // 构造从工厂得到连接对象
                connection = connectionFactory.createConnection();
                // 启动
                connection.start();
                // 获取操作连接
                session = connection.createSession(Boolean.TRUE,
                        Session.AUTO_ACKNOWLEDGE);
                // 获取session注意参数值xingbo.xu-queue是一个服务器的queue,须在在ActiveMq的console配置
                destination = session.createQueue("FirstQueue");
                // 得到消息生成者【发送者】
                producer = session.createProducer(destination);
                // 设置不持久化,此处学习,实际根据项目决定
                producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
                // 构造消息,此处写死,项目就是参数,或者方法获取
                sendMessage(session, producer);
                session.commit();
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                try {
                    if (null != connection)
                        connection.close();
                } catch (Throwable ignore) {
                }
            }
        }
    
        public static void sendMessage(Session session, MessageProducer producer)
                throws Exception {
            for (int i = 1; i <= SEND_NUMBER; i++) {
                TextMessage message = session
                        .createTextMessage("ActiveMq 发送的消息" + i);
                // 发送消息到目的地方
                System.out.println("发送消息:" + "ActiveMq 发送的消息" + i);
                producer.send(message);
            }
        }
    }
    

    Receiver.java

     

    package com.newer.tanlei;
    
    import javax.jms.Connection;
    import javax.jms.ConnectionFactory;
    import javax.jms.Destination;
    import javax.jms.MessageConsumer;
    import javax.jms.Session;
    import javax.jms.TextMessage;
    import org.apache.activemq.ActiveMQConnection;
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    public class Receiver {
        public static void main(String[] args) {
            // ConnectionFactory :连接工厂,JMS 用它创建连接
            ConnectionFactory connectionFactory;
            // Connection :JMS 客户端到JMS Provider 的连接
            Connection connection = null;
            // Session: 一个发送或接收消息的线程
            Session session;
            // Destination :消息的目的地;消息发送给谁.
            Destination destination;
            // 消费者,消息接收者
            MessageConsumer consumer;
            connectionFactory = new ActiveMQConnectionFactory(
                    ActiveMQConnection.DEFAULT_USER,
                    ActiveMQConnection.DEFAULT_PASSWORD,
                    "tcp://localhost:61616");
            try {
                // 构造从工厂得到连接对象
                connection = connectionFactory.createConnection();
                // 启动
                connection.start();
                // 获取操作连接
                session = connection.createSession(Boolean.FALSE,
                        Session.AUTO_ACKNOWLEDGE);
                // 获取session注意参数值xingbo.xu-queue是一个服务器的queue,须在在ActiveMq的console配置
                destination = session.createQueue("FirstQueue");
                consumer = session.createConsumer(destination);
                while (true) {
                    //设置接收者接收消息的时间,为了便于测试,这里谁定为500s
                    TextMessage message = (TextMessage) consumer.receive(500000);
                    if (null != message) {
                        System.out.println("收到消息" + message.getText());
                    } else {
                        break;
                    }
                }
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                try {
                    if (null != connection)
                        connection.close();
                } catch (Throwable ignore) {
                }
            }
        }
    }
    

      

     

    注意事项

    1. 最后接收者跟发送者在不同的机器上测试
    2. 项目所引用的jar最后在ActiveMQ下的lib中找,这样不会出现版本冲突。

    测试过程

    因为是在单机上测试,所以需要开启两个eclipse,每一个eclipse都有自身的workspace。我们在eclipse1中运行Receiver,在eclipse2中运行Sender。

    刚开始eclipse1中运行Receiver以后console介面没有任何信息,在eclipse2中运行Sender以后,eclipse2中的console显示如下信息:

    发送消息:ActiveMq 发送的消息1
    发送消息:ActiveMq 发送的消息2
    发送消息:ActiveMq 发送的消息3
    发送消息:ActiveMq 发送的消息4
    发送消息:ActiveMq 发送的消息5

    而回到eclipse1中发现console界面出现如下信息:

    收到消息ActiveMq 发送的消息1
    收到消息ActiveMq 发送的消息2
    收到消息ActiveMq 发送的消息3
    收到消息ActiveMq 发送的消息4
    收到消息ActiveMq 发送的消息5

     

    今天发现测试并不需要开启两个eclipse,在一个eclipse下页可以启动多个程序,并且有多个console,在上面的Receiver.java中,设置一个较大的时间,比如receive(500000),如下代码所示:

    TextMessage message = (TextMessage) consumer.receive(500000);
    

    这个时候运行Receiver.java的话,会使得这个Receiver.java一直运行500秒,在eclipse中可以发现:  

     

    点击那个红色方块可以手动停止运行程序。

    运行玩receiver以后我们在运行sender,在运行完sender以后,我们要切换到receiver的console,如下图所示:

     

     Number Of Consumers  消费者 这个是消费者端的消费者数量

    Number Of Pending Messages 等待消费的消息 这个是当前未出队列的数量。可以理解为总接收数-总出队列数
    Messages Enqueued 进入队列的消息  进入队列的总数量,包括出队列的。 这个数量只增不减
    Messages Dequeued 出了队列的消息  可以理解为是消费这消费掉的数量
    这个要分两种情况理解
    在queues里它和进入队列的总数量相等(因为一个消息只会被成功消费一次),如果暂时不等是因为消费者还没来得及消费。
    在 topics里 它因为多消费者从而导致数量会比入队列数高。
    简单的理解上面的意思就是
    当有一个消息进入这个队列时,等待消费的消息是1,进入队列的消息是1。
    当消息消费后,等待消费的消息是0,进入队列的消息是1,出队列的消息是1.
    在来一条消息时,等待消费的消息是1,进入队列的消息就是2.


    没有消费者时  Pending Messages   和 入队列数量一样
    有消费者消费的时候 Pedding会减少 出队列会增加
    到最后 就是 入队列和出队列的数量一样多
    以此类推,进入队列的消息和出队列的消息是池子,等待消费的消息是水流。 

  • 相关阅读:
    [ Luogu 3398 ] 仓鼠找sugar
    [ JLOI 2014 ] 松鼠的新家
    AtcoderGrandContest 005 F. Many Easy Problems
    Codeforces 388 D. Fox and Perfect Sets
    Codeforces 1037 H. Security
    「学习笔记」wqs二分/dp凸优化
    「NOI2017」游戏
    「SCOI2014」方伯伯的商场之旅
    「SCOI2015」情报传递
    「SCOI2016」美味
  • 原文地址:https://www.cnblogs.com/tanlei-sxs/p/9796904.html
Copyright © 2011-2022 走看看