一、ActiveMQ的介绍。

　　ActiveMQ 是Apache出品，最流行的，能力强劲的开源消息总线。ActiveMQ 是一个完全支持JMS1.1和J2EE 1.4规范的 JMS Provider实现。

1、主要特点：

　　（1）. 多种语言和协议编写客户端。语言: Java, C, C++, C#, Ruby, Perl, Python, PHP。应用协议: OpenWire,Stomp REST,WS Notification,XMPP,AMQP。

　　（2）.完全支持JMS1.1和J2EE 1.4规范 (持久化,XA消息,事务)。

　　（3）.ActiveMQ可以很容易内嵌到使用Spring的系统里面。

　　（4）.支持多种传送协议:in-VM,TCP,SSL,NIO,UDP,JGroups,JXTA。

　　（5）. 支持通过JDBC和journal提供高速的消息持久化。

　　（6）. 从设计上保证了高性能的集群,客户端-服务器,点对点。

　　（7） 支持Ajax。

　　（8）. 支持与Axis的整合。

　　（9）. 可以很容易得调用内嵌JMS provider,进行测试。

2、ActiveMQ的消息形式。

2.1.消息的传递有两种类型：

（1）点对点：一个生产者和一个消费者一一对应。

（2）发布/订阅模式，即一个生产者产生消息并进行发送后，可以由多个消费者进行接收。

2.2、JMS定义了五种不同的消息正文格式，以及调用的消息类型，允许你发送并接收以一些不同形式的数据，提供现有消息格式的一些级别的兼容性。

　　StreamMessage — Java原始值的数据流。

　　MapMessage–一套名称-值对。

　　TextMessage–一个字符串对象。

　　ObjectMessage–一个序列化的 Java对象。

　　BytesMessage–一个字节的数据流。

二、ActiveMQ的安装。

下载地址：https://activemq.apache.org/

 

1、安装环境。

1.1、需要JDK。

1.2、安装Linux系统。生产环境都是Linux系统。

2、安装步骤。

2.1、把ActiveMQ 的压缩包上传到Linux系统。

2.2、解压缩 。

2.3、启动。

（1）使用bin目录下的activemq命令启动：[root@localhost bin]# ./activemq start

（2）关闭：[root@localhost bin]# ./activemq stop

（3）查看状态：[root@localhost bin]# ./activemq status

（4）进入管理后台：http://192.168.25.168:8161/admin  用户名：admin   密码：admin

三、ActiveMQ的使用方法。

 

 

 1、Queue。

1.1、生产者（Product）：生产消息，发送端。在Pom.xml文件中引入依赖，这里使用的是activemq-all-5.11.2.jar。

（1）创建 ConnectionFactory 对象，需要指定服务端ip及端口号。

（2）、使用ConnectionFactory对象创建一个Connection对象。

（3）开启连接，调用Connection对象的 start() 方法。

（4）创建Session，使用Connection对象创建一个Session对象。

（5）使用Session对象创建一个Destination对象（queue、topic），这里创建的是Queue对象。

（6）使用Session对象创建一个Product对象。

（7）创建一个Message对象，创建一个TestMessage对象。

（8）使用Product发送消息。

（9）关闭资源。

    @Test
    public void testQueueProducer() throws Exception {
        // 第一步：创建ConnectionFactory对象，需要指定服务端ip及端口号。
        //brokerURL服务器的ip及端口号
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.25.168:61616");
        // 第二步：使用ConnectionFactory对象创建一个Connection对象。
        Connection connection = connectionFactory.createConnection();
        // 第三步：开启连接，调用Connection对象的start方法。
        connection.start();
        // 第四步：使用Connection对象创建一个Session对象。
        //第一个参数：是否开启事务。true：开启事务，第二个参数忽略。
        //第二个参数：当第一个参数为false时，才有意义。消息的应答模式。1、自动应答2、手动应答。一般是自动应答。
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        // 第五步：使用Session对象创建一个Destination对象（topic、queue），此处创建一个Queue对象。
        //参数：队列的名称。
        Queue queue = session.createQueue("test-queue");
        // 第六步：使用Session对象创建一个Producer对象。
        MessageProducer producer = session.createProducer(queue);
        // 第七步：创建一个Message对象，创建一个TextMessage对象。
        /*TextMessage message = new ActiveMQTextMessage();
        message.setText("hello activeMq,this is my first test.");*/
        TextMessage textMessage = session.createTextMessage("hello activeMq,this is my first test.");
        // 第八步：使用Producer对象发送消息。
        producer.send(textMessage);
        // 第九步：关闭资源。
        producer.close();
        session.close();
        connection.close();
    }

1.2、消费者（Consumer）：接收消息。

（1）创建一个ConnectionFactory对象。

（2）从ConnectionFactory对象中获取一个Connection对象。

（3）开启连接。调用Connection对象的start方法。

（4）使用Connection对象创建一个Session对象。

（5）使用Session对象创建一个Destination对象，和发送端保持一致Queue，并且队列的名称一致。

（6）使用Session对象创建一个Consumer对象。

（7）接收消息。

（8）打印消息。

（9）关闭资源。

    @Test
    public void testQueueConsumer() throws Exception {
        // 第一步：创建一个ConnectionFactory对象。
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.25.168:61616");
        // 第二步：从ConnectionFactory对象中获得一个Connection对象。
        Connection connection = connectionFactory.createConnection();
        // 第三步：开启连接。调用Connection对象的start方法。
        connection.start();
        // 第四步：使用Connection对象创建一个Session对象。
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        // 第五步：使用Session对象创建一个Destination对象。和发送端保持一致queue，并且队列的名称一致。
        Queue queue = session.createQueue("test-queue");
        // 第六步：使用Session对象创建一个Consumer对象。
        MessageConsumer consumer = session.createConsumer(queue);
        // 第七步：接收消息。
        consumer.setMessageListener(new MessageListener() {
            
            @Override
            public void onMessage(Message message) {
                try {
                    TextMessage textMessage = (TextMessage) message;
                    String text = null;
                    //取消息的内容
                    text = textMessage.getText();
                    // 第八步：打印消息。
                    System.out.println(text);
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }
        });
        //等待键盘输入
        System.in.read();
        // 第九步：关闭资源
        consumer.close();
        session.close();
        connection.close();
    }

2、Topic。

2.1、生产者（Product）。

（1）创建ConnectionFactory对象，需要指定服务端ip及端口号。

（2）使用ConnectionFactory对象创建一个Connection对象。

（3）开启连接。调用Connection对象的 start 方法。

（4）使用Connection对象创建一个Session对象。

（5）使用Session对象创建一个Destination对象（topic、queue），这里创建一个Topic对象。

（6）使用Session对象创建一个Product对象。

（7）创建一个Message，创建一个TextMessage对象。

（8）使用Product对象发送消息。

（9）关闭资源。

    @Test
    public void testTopicProducer() throws Exception {
        // 第一步：创建ConnectionFactory对象，需要指定服务端ip及端口号。
        // brokerURL服务器的ip及端口号
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.25.168:61616");
        // 第二步：使用ConnectionFactory对象创建一个Connection对象。
        Connection connection = connectionFactory.createConnection();
        // 第三步：开启连接，调用Connection对象的start方法。
        connection.start();
        // 第四步：使用Connection对象创建一个Session对象。
        // 第一个参数：是否开启事务。true：开启事务，第二个参数忽略。
        // 第二个参数：当第一个参数为false时，才有意义。消息的应答模式。1、自动应答2、手动应答。一般是自动应答。
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        // 第五步：使用Session对象创建一个Destination对象（topic、queue），此处创建一个topic对象。
        // 参数：话题的名称。
        Topic topic = session.createTopic("test-topic");
        // 第六步：使用Session对象创建一个Producer对象。
        MessageProducer producer = session.createProducer(topic);
        // 第七步：创建一个Message对象，创建一个TextMessage对象。
        /*
         * TextMessage message = new ActiveMQTextMessage(); message.setText(
         * "hello activeMq,this is my first test.");
         */
        TextMessage textMessage = session.createTextMessage("hello activeMq,this is my topic test");
        // 第八步：使用Producer对象发送消息。
        producer.send(textMessage);
        // 第九步：关闭资源。
        producer.close();
        session.close();
        connection.close();
    }

2.2、消费者（Consumer）：接收消息。

（1）创建一个ConnectionFactory对象。

（2）使用ConnectionFactory对象创建一个Connection对象。

（3）开启连接。调用Connection对象的 start 方法开启连接。

（4）创建Session对象。使用Connection对象创建一个Session对象。

（5）使用Session对象创建一个Destination对象。和发送端保持一个的topic，并且话题的名称保持一致。

（6）使用Session对象创建一个Consumer对象。

（7）接收消息。

（8）打印消息。

（9）关闭资源。

    @Test
    public void testTopicConsumer() throws Exception {
        // 第一步：创建一个ConnectionFactory对象。
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.25.168:61616");
        // 第二步：从ConnectionFactory对象中获得一个Connection对象。
        Connection connection = connectionFactory.createConnection();
        // 第三步：开启连接。调用Connection对象的start方法。
        connection.start();
        // 第四步：使用Connection对象创建一个Session对象。
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        // 第五步：使用Session对象创建一个Destination对象。和发送端保持一致topic，并且话题的名称一致。
        Topic topic = session.createTopic("test-topic");
        // 第六步：使用Session对象创建一个Consumer对象。
        MessageConsumer consumer = session.createConsumer(topic);
        // 第七步：接收消息。
        consumer.setMessageListener(new MessageListener() {

            @Override
            public void onMessage(Message message) {
                try {
                    TextMessage textMessage = (TextMessage) message;
                    String text = null;
                    // 取消息的内容
                    text = textMessage.getText();
                    // 第八步：打印消息。
                    System.out.println(text);
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }
        });
        System.out.println("topic的消费端03。。。。。");
        // 等待键盘输入
        System.in.read();
        // 第九步：关闭资源
        consumer.close();
        session.close();
        connection.close();
    }

原文地址：http://www.cnblogs.com/sfwu/p/16879961.html

1. 本站所有资源来源于用户上传和网络，如有侵权请邮件联系站长！ 2. 分享目的仅供大家学习和交流，请务用于商业用途! 3. 如果你也有好源码或者教程，可以到用户中心发布，分享有积分奖励和额外收入！ 4. 本站提供的源码、模板、插件等等其他资源，都不包含技术服务请大家谅解！ 5. 如有链接无法下载、失效或广告，请联系管理员处理！ 6. 本站资源售价只是赞助，收取费用仅维持本站的日常运营所需！ 7. 如遇到加密压缩包，默认解压密码为"gltf",如遇到无法解压的请联系管理员！ 8. 因为资源和程序源码均为可复制品，所以不支持任何理由的退款兑现，请斟酌后支付下载 声明：如果标题没有注明"已测试"或者"测试可用"等字样的资源源码均未经过站长测试.特别注意没有标注的源码不保证任何可用性