消息中间件之ActiveMQ之一

Posted 2021-12-01 xuweiweiwoaini

1 ActiveMQ简介

1.1 ActiveMQ是什么？

• ActiveMQ是Apache推出的，一款开源的，完全支持JMS1.1和J2EE 1.4规范的JMS Provider实现的消息中间件（Message Oriented Middleware，MOM）。

1.2 ActiveMQ能做什么？

• 最主要的功能就是：实现JMS规范的Provider，用来帮助实现高可用、高性能、可伸缩、易用和安全的企业级面向消息服务的系统。

1.3 ActiveMQ的特点

• 完全支持JMS1.1和J2EE 1.4规范（持久化、XA消息、事务）。
• 支持多种传输协议：in-VM、TCP、SSL、NIO、UDP、JGroups、JXTA等。
• 可插拔的体系结构，可用灵活定制，如消息存储方式、安全管理等。
• 很容易和应用服务器继承使用。
• 多种语言和协议编写客户端。语言：java、C、C++、C#、Ruby、Python、php等。
• 从设计上保证了高性能的集群：客户端-服务器、点对点。
• 可以很容易的和Spring集成使用。
• 支持通过JDBC和Journal提供高速的消息持久化。
• 支持和Axis的整合。

 

2 消息中间件

2.1 消息中间件的基本功能

• 将信息以消息的形式，从一个应用程序传送到另一个或多个应用程序。

2.2 消息中间件的主要特点

• ①消息异步接收：类似手机短信的行为，消息的发送者不需要等待消息接收者的响应，减少软件多系统集成的耦合度。
• ②消息可靠接收：确保消息在中间件可靠保存，只有当接收方收到后才删除消息，多个消息也可以组成原子事务。

2.3 消息中间件的主要应用场景

• 在多个系统间进行整合和通讯的时候，通常会要求：
• ①可靠传输，数据不能丢失，有的时候，也会要求不能重复传输。
• ②异步传输，如果是同步发送数据，各个系统会出现互相等待，非常容易造成系统瓶颈。

2.4 目前比较知名的消息中间件

• IBM MQSeries
• BEA WebLogic JMS Server
• Oracle AQ
• Tibco
• SwiftMQ
• ActiveMQ
• RobbitMQ
• RocketMQ
• ……

 

3 ActiveMQ的安装和基本使用

3.1 注意

• 本人所使用的操作系统是CentOS7。

3.2 下载并安装ActiveMQ服务器端

• ActiveMQ的下载地址：http://activemq.apache.org
• 直接解压，然后拷贝到指定的目录即可。

3.3 启动运行

3.3.1 普通方式

• ①进入ActiveMQ的bin目录下。
• ②输入./activemq start

3.3.2 启动并指定日志文件

• ①进入ActiveMQ的bin目录下
• ②输入 ./activemq start > /tmp/activemq.log

3.4 检查是否已经启动

• ActiveMQ默认采用61616端口提供JMS服务，使用8161端口提供管理控制台服务，执行以下命令以便检查是否成功启动ActiveMQ服务。

3.4.1 查看61616端口是否打开

• netstat -anp | grep -i 61616

3.4.2 查看控制台输出或者日志文件

3.4.3 可以直接访问ActiveMQ的管理页面

• http://localhost:8161/admin
• 默认用户名/密码是admin/admin

3.5 关闭ActiveMQ

• ①在ActiveMQ的bin目录下，输入./activemq stop。
• ②根据端口找到进程号，杀死即可。

 

4 ActiveMQ的快速入门

4.1 导入pom

      <dependency>
            <groupId>org.apache.activemq</groupId>
            <artifactId>activemq-all</artifactId>
            <version>5.15.9</version>
        </dependency>
        <dependency>
            <groupId>org.apache.xbean</groupId>
            <artifactId>xbean-spring</artifactId>
            <version>4.13</version>
        </dependency>

4.2 快速入门

• 示例：生产者

package com.xuweiwei;

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

public class QueueSender {
    public static void main(String[] args) throws JMSException, InterruptedException {
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.78.50:61616");

        Connection connection = connectionFactory.createConnection();
        connection.start();

        Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);


        Destination destination = session.createQueue("xuweiwei-queue");

        MessageProducer producer = session.createProducer(destination);

        for (int i = 0; i < 10; i++) {
            TextMessage textMessage = session.createTextMessage("许威威就是好啊" + i);
            Thread.sleep(1000);
            //通过消费生产者发出消息
            producer.send(textMessage);
        }

        session.commit();
        session.close();
        connection.close();

    }


}

View Code

 

• 示例：消费者

package com.xuweiwei;

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

public class QueueReceiver {
    public static void main(String[] args) throws JMSException {
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.78.50:61616");

        Connection connection = connectionFactory.createConnection();
        connection.start();

        Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);

        Destination destination = session.createQueue("xuweiwei-queue");

        MessageConsumer consumer = session.createConsumer(destination);

        for (int i = 0; i < 10; i++) {
            TextMessage textMessage = (TextMessage) consumer.receive();
            session.commit();
            System.out.println("消费了" + textMessage.getText());
        }
        session.close();
        connection.close();

    }
}

View Code