美文网首页
第一课 ActiveMQ的安装及使用

第一课 ActiveMQ的安装及使用

作者: Arroganter | 来源:发表于2018-08-27 08:39 被阅读6次

    ActiveMq简单介绍
    一般常见的消息中间件有:RabbitMQ,ActiveMq,RocketMQ(阿里)等,都称之为MQ(Message Queue,消息队列),这里介绍ActiveMQ。
    ActiveMQ是Apache出品,最流行的,能力强劲的开源消息总线。从设计上保证了高性能的集群,客户端-服务器,点对点。完全支持JMS1.1和J2EE1.4规范的JMSProvider实现,尽管JMS规范出台已经是很久的事情了,但是JMS在当今的J2EE应用中间仍然扮演着特殊的地位。
    特性:
    ⒈多种语言和协议编写客户端。语言:Java,C,C++,C#,Ruby,Perl,Python,PHP。应用协议:OpenWire,StompREST,WSNotification,XMPP,AMQP

    ⒉完全支持JMS1.1和J2EE1.4规范(持久化,XA消息,事务)

    ⒊对Spring的支持,ActiveMQ可以很容易内嵌到使用Spring的系统里面去,而且也支持Spring2.0的特性

    ⒋通过了常见J2EE服务器(如Geronimo,JBoss4,GlassFish,WebLogic)的测试,其中通过JCA1.5resourceadaptors的配置,可以让ActiveMQ可以自动的部署到任何兼容J2EE1.4商业服务器上

    ⒌支持多种传送协议:in-VM,TCP,SSL,NIO,UDP,JGroups,JXTA

    ⒍支持通过JDBC和journal提供高速的消息持久化

    ⒎从设计上保证了高性能的集群,客户端-服务器,点对点

    ⒏支持Ajax

    ⒐支持与Axis的整合

    ⒑可以很容易得调用内嵌JMSprovider,进行测试

    消息形式:
    一种是点对点的,即一个生产者和一个消费者一一对应;
    另一种是发布/订阅模式,即一个生产者产生消息并进行发送后,可以由多个消费者进行接收。
    JMS定义了五种不同的消息正文格式,以及调用的消息类型,允许你发送并接收以一些不同形式的数据,提供现有消息格式的一些级别的兼容性。
      • StreamMessage – Java原始值的数据流
      • MapMessage–一套名称-值对
      • TextMessage–一个字符串对象
      • ObjectMessage–一个序列化的 Java对象
      • BytesMessage–一个字节的数据流

    ActiveMq的安装启动
    ActiveMQ的安装启动很简单。
    下载:进入http://activemq.apache.org/下载ActiveMQ
    这里在linux下进行安装。
    1、将下载的ActiveMQ压缩包上传到linux(我是上传到了/usr目录下),用远程连接工具(比如ssh)。
    2、解压缩 :tar zxvf apache-activemq-5.15.4-bin.tar.gz
    3、启动:进入bin目录 ./activemq start
    查看是否启动成功,./activemq status
    如果没有启动成功,那么就要修改hosts文件了,配置机器名和本机(127.0.0.1)的映射关系。比如本机名是laowang(可以使用命令/etc/sysconfig/network查看机器名称)。那么需要做如下的修改:

    [root@laowang bin]# cat /etc/hosts
    127.0.0.1   localhost localhost.localdomain localhost4 localhost4.localdomain4
    ::1         localhost localhost.localdomain localhost6 localhost6.localdomain6
    127.0.0.1           laowang.localdomain itheima32//加上这一行
    
    

    然后浏览器访问:192.168.25.128:8161(8161是固定端口号)

    image.png

    image.png
    那么说明启动成功了。
    注:点击Manage ActiveMqQ broker输入用户名密码为admin。
    ActiveMQ在java中的使用


    image.png
    image.png

    对照着上面两种模式图,编写代码的时候后会更容易理解。
    创建个maven项目,导入的包是

    <dependency>
    <groupId>org.apache.activemq</groupId>
    <artifactId>activemq-all</artifactId>
    <version>5.14.0</version>
    </dependency>
    包版本超过5.14.0会和mybatis冲突

    1、点对点模式:生产者

    /*
     * 测试点对点模式:生产者
     */
    @Test
    public void testQueueProducer() throws JMSException{
        //1.创建一个连接工厂对象,需要指定服务的ip及端口号
        ActiveMQConnectionFactory factory = 
                new ActiveMQConnectionFactory("tcp://192.168.25.128:61616");
        //2.使用工厂对象来创建一个Connection对象
        Connection connection = factory.createConnection();
        //3.开启连接,调用Connection对象的start方法
        connection.start();
        //4.创建一个Session对象
        //第一个参数:是否开启事务,如果开启事务true带二个参数无意义,一般不开启事务false
        //第二个参数:应答模式。一般自动应答或者手动应答。一般自动应答
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        //5.使用Session对象创建一个Destination对象。两种形式queue,topic,现在应该使用queue
        Queue queue = session.createQueue("测试消息队列First");
        //6.使用Session对象创建一个Producer对象
        MessageProducer producer = session.createProducer(queue);
        //7.创建一个Message对象,可以使用TextMessage
        Message message = session.createTextMessage("hello activemq!");
        //8.发送消息
        producer.send(message);
        //9.关闭资源
        producer.close();
        session.close();
        connection.close();
    }
    

    运行方法后,会发送一个消息,可以在页面查看。

    image.png
    image.png

    一条挂起的消息,一条入队的消息,也能看到目的地跟消息详情,以及消息被持久化。

    2、点对点模式:消费者
    前面生产了一条消息,这里就测试消费者去消费这条消息。

    /*
     * 测试点对点模式:消费者
     */
    @Test
    public void testQueueConsumer() throws Exception{
        //1.创建连接工厂
        ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://192.168.25.128:61616");
        //2.创建连接
        Connection connection = factory.createConnection();
        //3.开启连接
        connection.start();
        //4.创建session对象
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        //5.创建目的地,需要跟生产者发送消息的目的地一致不然不能接收到消息
        Queue queue = session.createQueue("测试消息队列First");
        //使用Session对象创建一个消费者对象
        MessageConsumer consumer = session.createConsumer(queue);
        //接收消息
        consumer.setMessageListener(new MessageListener() {
    
            public void onMessage(Message message) {
    
                TextMessage textMessage = (TextMessage)message;
                try {
                    String text = textMessage.getText();
                    System.out.println(text);
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }
        });
        //等待接收消息,消费完了就没了
        System.in.read();//等待我们敲击键盘才会向下执行
        //关闭资源
        consumer.close();
        session.close();
        connection.close();
    }
    

    运行后

    image.png
    image.png

    image.png
    被挂起的消息已经被消费了,一个消费者,一个入队消息一个出队消息。如果再发生消息的话控制台会持续(也就是消费者能持续接收生产者发送的消息)输出直至按回下回车键才结束。这个时候消费者数量也变为0了。

    3、发布/订阅模式:生产者
    跟点对点模式的生产者很类似,主要是创建的目的地为Topic而不是Queue了。

    /*
     * 发布订阅模式:生产者
     */
    @Test
    public void testTopicProducer() throws Exception{
        //1.创建连接工厂
        ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://192.168.25.128:61616");
        //2.创建连接
        Connection connection = factory.createConnection();
        //3.开启连接
        connection.start();
        //4.创建session对象
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        //5.创建目的地Destination,该模式创建的是Topic
        Topic topic = session.createTopic("测试消息队列First");
        //6.创建生产者
        MessageProducer producer = session.createProducer(topic);
        //7.创建消息
        TextMessage message = session.createTextMessage("hello topic");
        //8.发送消息
        producer.send(message);
        //9.关闭资源
        producer.close();
        session.close();
        connection.close();
    }
    

    运行后去页面点击Topics查看

    image.png

    0个消费者,该模式默认消息不被持久化,但是可以修改为持久化,只有一条入队消息。
    4、发布/订阅模式:消费者

    /*
     * 发布订阅模式:消费者
     */
    @SuppressWarnings("unused")
    @Test
    public void testTopicConsumer() throws Exception{
        //1.创建连接工厂
        ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://192.168.25.128:61616");
        //2.创建连接
        Connection connection = factory.createConnection();
        //3.开启连接
        connection.start();
        //4.创建session对象
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        //5.创建目的地Destination
        Topic topic = session.createTopic("测试消息队列First");
        //6.创建消费者
        MessageConsumer consumer = session.createConsumer(topic);
        //7.接收消息
        consumer.setMessageListener(new MessageListener() {
    
            @Override
            public void onMessage(Message message) {
                TextMessage textMessage = (TextMessage)message;
                //获取信息并打印
                try {
                    String text = textMessage.getText();
                    System.out.println(text);
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }
        });
        System.out.println("----消费者1准备就绪----");
        System.in.read();
        //9.关闭资源
        consumer.close();
        session.close();
        connection.close();
    }
    

    虽然前面发布了一条消息但是既然不能被持久化,那么这里自然就不能去接收该消息了。
    所以需要先执行该方法,使消费者处于持续能接收消息的状态,这样生产者每次发送消息都能被接收到。
    这里该测试方法运行三次。打印依次为:
    System.out.println(“—-消费者1准备就绪—-“);
    System.out.println(“—-消费者2准备就绪—-“);
    System.out.println(“—-消费者3准备就绪—-“);
    就等于有三个消费者了。这个时候再执行测试生产者的方法(执行三次,等于发了三次消息),就能发现三个消费者都接收到了三条消息。
    查看控制台:

    ----消费者1准备就绪----
    hello topic
    hello topic
    hello topic
    ----消费者2准备就绪----
    hello topic
    hello topic
    hello topic
    ----消费者3准备就绪----
    hello topic
    hello topic
    hello topic
    再看页面:


    image.png

    4条入队消息(其中一条没被消费),9条出队消息(3个消费者每个消费3条消息)。

    Apache ActiveMQ 各个版本所依赖的JDK版本

    https://blog.csdn.net/yanfeipei/article/details/75267384

    相关文章

      网友评论

          本文标题:第一课 ActiveMQ的安装及使用

          本文链接:https://www.haomeiwen.com/subject/pxhiiftx.html