首页 > > 程序设计 > Java技术 >

ActiveMQ学习总结------实战操作（上）02

2019-10-12 08:38:49来源：博客园 阅读 ()

新老客户大回馈,云服务器低至5折

ActiveMQ学习总结------实战操作（上）02

相信大家通过上一篇博文已经对ActiveMQ有了一个大致的概念了，

那么本篇博文将带领大家一步一步去实战操作我们的ActiveMQ

本篇主要内容：

　　1.ActiveMQ术语及API介绍

　　2.ActiveMQ 文本消息处理

　　3.ActiveMQ 对象消息处理

既然我们要学习如何实战操作了，那么久不可不知它的一些术语和API的用于

即使觉得枯燥，我们也要大致看一眼，过后相信你还会再来看的！因为很有用

 

 

一 ActiveMQ 的术语（这么碉堡的东西怎么会没有写让人觉得更碉堡的词汇呢？）

1 Destination

目的地，JMS Provider（消息中间件）负责维护，用于对 Message 进行管理的对象。
MessageProducer 需要指定 Destination 才能发送消息，MessageReceiver 需要指定 Destination
才能接收消息。

 

2 Producer

消息生成者，负责发送 Message 到目的地。

 

3 Consumer | Receiver

消息消费者，负责从目的地中消费【处理|监听|订阅】Message。

 

4 Message

消息，消息封装一次通信的内容。

 

 

 

二 ActiveMQ API简介

下述API都是接口类型，由定义在javax.jms包中

都是JMS（Java Message Service）标准接口定义

 

1.ConnectionFactory

　　链接工厂，用于创建链接的工厂类型

2.Connection

　　链接，用于建立访问ActiveMQ连接的类型，由链接工厂ConnectionFactory创建

3.Session

　　会话，一次持久有效有状态的访问，由链接创建

4.Destination & Queue

　　目的地。用于描述本次访问ActiveMQ的消息访问目的地，即ActiveMQ服务中的具体队列，由会话创建

　　interface Queue extends Destination

5.MessageProducer

　　消息生成者、在一次有效会话中，用于发送消息给ActiveMQ服务的工具，由会话创建

6.MessageConsumer

　　消息消费者【消息订阅者|消息处理着】，在一次有效会话中，用于从ActiveMQ服务中获取消息的工具，由会话创建

7.Message

　　消息。通过消息生成者向ActiveMQ服务发送消息时使用的数据载体对象或消费者从ActiveMQ服务中获取消息时使用的数据载体对象。是所有消息【文本消息|对象消息等】具体类型的顶级接口，可以通过会话创建或通过会话从ActiveMQ服务中获取

 

 

 

 

三 ActiveMQ处理文本消息

注：*本案例将以producer（消息生产者）和consumer（消息发送者）连个模块为例

 

准备工作：

　　确保你的ActiveMQ已经开启，并且所依赖的端口已经关闭（建议直接关闭防火墙 service iptables stop）

　　Maven环境

 

1 消息生产者

1.1创建模块

 

 

1,2在POM文件添加ActiveMQ的依赖（根据个人的ActiveMQ版本自行查找）

 <dependencies>
        <dependency>
            <groupId>org.apache.activemq</groupId>
            <artifactId>activemq-all</artifactId>
            <version>5.9.0</version>
        </dependency>
    </dependencies>

 

1.3 编写消息的生产者

package cn.arebirth.mq;

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

public class ActiveMQTextProducer {
    
    public void sendTextActiveMQ(String msg) {
        //定义链接工厂
        ConnectionFactory connectionFactory = null;

        //定义链接对象
        Connection connection = null;

        //定义会话
        Session session = null;

        //目的地
        Destination destination = null;

        //定义消息的发送者
        MessageProducer producer = null;

        //定义消息
        Message message = null;


        try {
            /**
             * userName:访问ActiveMQ服务的用户名，默认为admin。
             * password：访问ActiveMQ服务的密码，默认为admin
             * 用户名和用户密码都可以通过ActiveMQ安装目录的oonf目录下的jetty-ream.properties文件进行修改
             *
             * borkerURL：访问ActiveMQ服务的路径地址。
             *  路径结构为：协议名://主机地址:端口号
             *  在conf/activemq.xml文件中可以找到修改
             *  在上一篇文章中都有介绍
             */
            connectionFactory = new ActiveMQConnectionFactory("admin", "admin", "tcp://169.254.18.20:61616");

            //创建链接对象
            connection = connectionFactory.createConnection();

            //启动连接
            connection.start();

            /**
             * 参数一：transacted 是否使用事务 可选值为:true|false
             *      ture:使用事务 设置第二个参数变量这为 Session.SESSION_TRANSACTION 交由session管理
             *      false:不使用事务，则设置我们的参数即可
             *
             * acknowledgeMode:
             * * Session.AUTO_ACKNOWLEDGE:自动消息确认机制
             * * Session.CLIENT_ACKNOWLEDGE:客户端确认机制
             * * Session.DUPS_OK_ACKNOWLEDGE:有副本的客户端确认消息机制
             这里设置 这两个参数的含义为：
             不使用事务，并由Session自动确认提交
             * 这里对此不作过多讲解，初学者跟着敲，其后在慢慢了解原理
             *
             */
            session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

            //创建目的地，目的地名称即队列的名称，也就是保存我们消息地方的名称。消息的消费者需要通过此名称访问对应的队列
            //名称不固定
            destination = session.createQueue("hello-mq");

            //创建消息的生产者 需要指定目的地（也就是把Destination传入参数）
            producer = session.createProducer(destination);

            //创建消息对象 并传入我们需要放入队列的消息
            message = session.createTextMessage(msg);

            //发送消息
            producer.send(message);

        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            //回收消息发送者资源
            if (producer != null) {
                try {
                    producer.close();
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }
            if (session != null) {
                try {
                    session.close();
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }
            if (connection != null) {
                try {
                    connection.close();
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }
        }


    }
}

 

 

2. 消息消费者

2.1创建模块

 

 

 

2,2在POM文件添加ActiveMQ的依赖（根据个人的ActiveMQ版本自行查找）

 <dependency>
            <groupId>org.apache.activemq</groupId>
            <artifactId>activemq-all</artifactId>
            <version>5.9.0</version>
        </dependency>

 

2.3编写消息的消费者

package cn.arebirth.mq;

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

public class ActiveMQTextConsumer {

    public void receiveTextActiveMQ() {
        //定义链接工厂
        ConnectionFactory connectionFactory = null;

        //定义链接对象
        Connection connection = null;

        //定义会话
        Session session = null;

        //目的地
        Destination destination = null;

        //定义消息的消费者（接收者）
        MessageConsumer consumer = null;

        //定义消息
        Message message = null;


        try {
            /**
             * userName:访问ActiveMQ服务的用户名，默认为admin。
             * password：访问ActiveMQ服务的密码，默认为admin
             * 用户名和用户密码都可以通过ActiveMQ安装目录的oonf目录下的jetty-ream.properties文件进行修改
             *
             * borkerURL：访问ActiveMQ服务的路径地址。
             *  路径结构为：协议名://主机地址:端口号
             *  在conf/activemq.xml文件中可以找到修改
             *  在上一篇文章中都有介绍
             */
            connectionFactory = new ActiveMQConnectionFactory("admin", "admin", "tcp://169.254.18.20:61616");

            //创建链接对象
            connection = connectionFactory.createConnection();

            //启动连接
            connection.start();

            /**
             * 参数一：transacted 是否使用事务 可选值为:true|false
             *      ture:使用事务 设置第二个参数变量这为 Session.SESSION_TRANSACTION 交由session管理
             *      false:不使用事务，则设置我们的参数即可
             *
             * acknowledgeMode:
             * * Session.AUTO_ACKNOWLEDGE:自动消息确认机制
             * * Session.CLIENT_ACKNOWLEDGE:客户端确认机制
             * * Session.DUPS_OK_ACKNOWLEDGE:有副本的客户端确认消息机制
             这里设置 这两个参数的含义为：
             不使用事务，并由Session自动确认提交
             * 这里对此不作过多讲解，初学者跟着敲，其后在慢慢了解原理
             *
             */
            session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

            //创建目的地，目的地名称即队列的名称，也就是保存我们消息地方的名称。消息的消费者需要通过此名称访问对应的队列
            //名称不固定
            destination = session.createQueue("hello-mq");

            //创建消息的消费者 需要指定目的地（也就是把Destination传入参数）
            consumer = session.createConsumer(destination);

            //创建消息对象 由消费者接收消息
            message = consumer.receive();

            //处理消息
            String msg = ((TextMessage) message).getText();
            System.out.println("ActiveMQ say:" + msg);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            //回收消息接收者资源
            if (consumer != null) {
                try {
                    consumer.close();
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }
            if (session != null) {
                try {
                    session.close();
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }
            if (connection != null) {
                try {
                    connection.close();
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }
        }

    }
}

 

 

3 测试成果

定义测试类，来测试我们的成果！

 

3.1 定义producer测试类

package cn.arebirth.mq;

public class ProducerTest {
    public static void main(String[] args) {
        ActiveMQTextProducer producer = new ActiveMQTextProducer();
        producer.sendTextActiveMQ("Hello,ActiveMQ!");
    }
}

当我们执行完这段代码后，我们打开我们的ActiveMQ的控制面板（首先你要启动你的ActiveMQ，上一篇有介绍到！）

 

 

接下来，当我们看到此图片内容即说明producer成果运行！（hello-mq我们自定义的队列名称！）

 

 

 

 

3.2 定义 Consumer测试类

package cn.arebirth.mq;

public class ConsumerTest {
    public static void main(String[] args) {
        ActiveMQTextConsumer consumer = new ActiveMQTextConsumer();
        consumer.receiveTextActiveMQ();
    }
}

 

运行结果出现如下内容即成果！

 

 

 

 

 

当我们仔细观察ActiveMQ控制台的时候会发现，

　　producer每产生一条消息对应的Number Of Pending Message和Message Enqueued就+1

　　consumer每次启动消费消息的时候，Number Of pending Message就会-1  Message Dequeued +1 

为什么呢？哈哈  相信大家已经大概猜出来了或者看英文的意思　　

介绍下：

  • Name:消息的名字
  • Number Of Pending Message:为消费的消息数量
  • Number Of Consumers:消费者的数量
  • Message Enqueued:队列共有过多少消息
  • Message Dequeued:消费者消费了多少条消息

 

鼓捣完文本消息，那么有的人会说，只能鼓捣文本？？错了，接下来我们搞实用的对象！（不是你和我，是我们和电脑 -. -）

 

 

三 ActiveMQ处理对象消息

有了上面处理文本消息的基础，我们将很容易掌握处理对象消息的能力！这里面的步骤，我们将略省一部分，和上面及其相似

 

1 对象创建（处理对象消息，那么一定是要有个数据载体的对象啦，我们自己创建一个即可     生产者和消费者均需要此对象！）

package cn.arebirth.pojo;

import java.io.Serializable;

/**
 * 一定要序列化！！
 */
public class User implements Serializable {
    private String username;
    private String pwd;
    private String content;

    @Override
    public String toString() {
        return "User{" +
                "username='" + username + '\'' +
                ", pwd='" + pwd + '\'' +
                ", content='" + content + '\'' +
                '}';
    }

    public User() {
    }

    public User(String username, String pwd, String content) {
        this.username = username;
        this.pwd = pwd;
        this.content = content;
    }

    public String getUsername() {
        return username;
    }

    public void setUsername(String username) {
        this.username = username;
    }

    public String getPwd() {
        return pwd;
    }

    public void setPwd(String pwd) {
        this.pwd = pwd;
    }

    public String getContent() {
        return content;
    }

    public void setContent(String content) {
        this.content = content;
    }
}

 

2 消息生产者

2.1 还是需要pom的jar包依赖的的，如果还在原来的环境上，则不需要添加啦！

<dependency>
            <groupId>org.apache.activemq</groupId>
            <artifactId>activemq-all</artifactId>
            <version>5.9.0</version>
        </dependency>

 

2.2 编写消息的生产者

package cn.arebirth.mq;

import cn.arebirth.pojo.User;
import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

public class ActiveMQObjectProducer {

    public void sendObjectActiveMQ(User user){
        //定义链接工厂
        ConnectionFactory connectionFactory = null;

        //定义链接对象
        Connection connection = null;

        //定义会话
        Session session = null;

        //目的地
        Destination destination = null;

        //定义消息的发送者
        MessageProducer producer = null;

        //定义消息
        Message message = null;


        try {
            /**
             * userName:访问ActiveMQ服务的用户名，默认为admin。
             * password：访问ActiveMQ服务的密码，默认为admin
             * 用户名和用户密码都可以通过ActiveMQ安装目录的oonf目录下的jetty-ream.properties文件进行修改
             *
             * borkerURL：访问ActiveMQ服务的路径地址。
             *  路径结构为：协议名://主机地址:端口号
             *  在conf/activemq.xml文件中可以找到修改
             *  在上一篇文章中都有介绍
             */
            connectionFactory = new ActiveMQConnectionFactory("admin", "admin", "tcp://169.254.18.20:61616");

            //创建链接对象
            connection = connectionFactory.createConnection();

            //启动连接
            connection.start();

            /**
             * 参数一：transacted 是否使用事务 可选值为:true|false
             *      ture:使用事务 设置第二个参数变量这为 Session.SESSION_TRANSACTION 交由session管理
             *      false:不使用事务，则设置我们的参数即可
             *
             * acknowledgeMode:
             * * Session.AUTO_ACKNOWLEDGE:自动消息确认机制
             * * Session.CLIENT_ACKNOWLEDGE:客户端确认机制
             * * Session.DUPS_OK_ACKNOWLEDGE:有副本的客户端确认消息机制
             这里设置 这两个参数的含义为：
             不使用事务，并由Session自动确认提交
             * 这里对此不作过多讲解，初学者跟着敲，其后在慢慢了解原理
             *
             */
            session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

            //创建目的地，目的地名称即队列的名称，也就是保存我们消息地方的名称。消息的消费者需要通过此名称访问对应的队列
            //名称不固定
            destination = session.createQueue("hello-mq-obj");

            //创建消息的生产者 需要指定目的地（也就是把Destination传入参数）
            producer = session.createProducer(destination);

            //创建消息对象 并传入我们需要放入队列的消息
            message = session.createObjectMessage(user);

            //发送消息
            producer.send(message);

        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            //回收消息发送者资源
            if (producer != null) {
                try {
                    producer.close();
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }
            if (session != null) {
                try {
                    session.close();
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }
            if (connection != null) {
                try {
                    connection.close();
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

 

 

3 消息消费者

3.1 还是需要pom的jar包依赖的的，如果还在原来的环境上，则不需要添加啦！

<dependency>
            <groupId>org.apache.activemq</groupId>
            <artifactId>activemq-all</artifactId>
            <version>5.9.0</version>
        </dependency>

 

3.2  编写消息的消费者

package cn.arebirth.mq;

import cn.arebirth.pojo.User;
import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

public class ActiveMQObjectConsumer {
    public void receiveObjectActiveMQ() {
        //定义链接工厂
        ConnectionFactory connectionFactory = null;

        //定义链接对象
        Connection connection = null;

        //定义会话
        Session session = null;

        //目的地
        Destination destination = null;

        //定义消息的消费者（接收者）
        MessageConsumer consumer = null;

        //定义消息
        Message message = null;


        try {
            /**
             * userName:访问ActiveMQ服务的用户名，默认为admin。
             * password：访问ActiveMQ服务的密码，默认为admin
             * 用户名和用户密码都可以通过ActiveMQ安装目录的oonf目录下的jetty-ream.properties文件进行修改
             *
             * borkerURL：访问ActiveMQ服务的路径地址。
             *  路径结构为：协议名://主机地址:端口号
             *  在conf/activemq.xml文件中可以找到修改
             *  在上一篇文章中都有介绍
             */
            connectionFactory = new ActiveMQConnectionFactory("admin", "admin", "tcp://169.254.18.20:61616");

            //创建链接对象
            connection = connectionFactory.createConnection();

            //启动连接
            connection.start();

            /**
             * 参数一：transacted 是否使用事务 可选值为:true|false
             *      ture:使用事务 设置第二个参数变量这为 Session.SESSION_TRANSACTION 交由session管理
             *      false:不使用事务，则设置我们的参数即可
             *
             * acknowledgeMode:
             * * Session.AUTO_ACKNOWLEDGE:自动消息确认机制
             * * Session.CLIENT_ACKNOWLEDGE:客户端确认机制
             * * Session.DUPS_OK_ACKNOWLEDGE:有副本的客户端确认消息机制
             这里设置 这两个参数的含义为：
             不使用事务，并由Session自动确认提交
             * 这里对此不作过多讲解，初学者跟着敲，其后在慢慢了解原理
             *
             */
            session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

            //创建目的地，目的地名称即队列的名称，也就是保存我们消息地方的名称。消息的消费者需要通过此名称访问对应的队列
            //名称不固定
            destination = session.createQueue("hello-mq-obj");

            //创建消息的消费者 需要指定目的地（也就是把Destination传入参数）
            consumer = session.createConsumer(destination);

            //创建消息对象 由消费者接收消息
            message = consumer.receive();

            //处理消息
            ObjectMessage objectMessage = (ObjectMessage) message;
            User user = (User) objectMessage.getObject();
            System.out.println("ActiveMQ say:" + user);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            //回收消息接收者资源
            if (consumer != null) {
                try {
                    consumer.close();
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }
            if (session != null) {
                try {
                    session.close();
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }
            if (connection != null) {
                try {
                    connection.close();
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }
        }

    }
}

 

 

 

4 创建测试类

4.1 Produce生产者测试代码

package cn.arebirth.mq;

import cn.arebirth.pojo.User;

public class ProducerTest {
    public static void main(String[] args) {
        ActiveMQObjectProducer producer = new ActiveMQObjectProducer();
        producer.sendObjectActiveMQ(new User("Arebirth","123","Hello,ActiveMQ!"));
    }
}

 

当我们执行完这段代码后，我们打开我们的ActiveMQ的控制面板（首先你要启动你的ActiveMQ，上一篇有介绍到！）

 

 会看到 队列中增加了我们的消息！

 

4.2 Consumer消费者测试代码

package cn.arebirth.mq;

public class ConsumerTest {
    public static void main(String[] args) {
        ActiveMQObjectConsumer consumer = new ActiveMQObjectConsumer();
        consumer.receiveObjectActiveMQ();
    }
}

当我们执行后，出现下面的内容即可！成功啦！

 

 在看下，控制面板，

 

 

是不是已经和上一个列子一样啦！

 

 

ps

　　本篇内容就到这里了，希望各位勤加敲代码，毕竟代码不是看出来的~

 


原文链接:https://www.cnblogs.com/arebirth/p/activemq02.html
如有疑问请与原作者联系

标签：

版权申明：本站文章部分自网络，如有侵权，请联系：west999com@outlook.com
特别注意：本站所有转载文章言论不代表本站观点，本站所提供的摄影照片，插画，设计作品，如需使用，请与原作者联系，版权归原作者所有

上一篇：通俗易懂spring之singleton和prototype

下一篇：面向过程与面向对象及面向对象三大特性

相关文章

IDC资讯： 主机资讯 注册资讯 托管资讯 vps资讯 网站建设

网站运营： 建站经验 策划盈利 搜索优化 网站推广 免费资源

网站联盟： 联盟新闻 联盟介绍 联盟点评 网赚技巧

行业资讯： 搜索引擎 网络游戏 电子商务 广告传媒

网络编程： Asp.Net编程 Asp编程 Php编程 Xml编程 Access Mssql Mysql 其它

服务器技术： Web服务器 Ftp服务器 Mail服务器 Dns服务器 安全防护

软件技巧： 其它软件 Word Excel Powerpoint Ghost Vista QQ空间 QQ FlashGet 迅雷

网页制作： FrontPages Dreamweaver Javascript css photoshop fireworks Flash

程序设计： Java技术 C/C++ VB delphi

网络知识： 网络协议 网络安全 网络管理 组网方案 Cisco技术

操作系统： Win2000 WinXP Win2003 Mac OS Linux FreeBSD

热门词条
最新资讯
热门关注
热门标签