zoukankan html css js c++ java

ActiveMQ学习总结（一）

自己写的网上商城项目中使用了ActiveMQ，虽然相比于RabbitMQ，kafka，RocketMQ等相比，ActiveMQ可能性能方面不是最好的选择，不过消息队列其实原理区别不大，这里对学过的关于消息队列的知识进行一下总结，并结合自己面试中关于这方面遇到的问题做一个整理，为后面秋招找工作做准备。这一篇主要介绍一下JMS，ActiveMQ安装及其常用接口，两种队列模式，如何集成到Spring项目，面试总结等。

Java Message Service，Java消息服务应用程序接口，是一个Java平台中关于面向消息中间件（MOM）的API，用于在两个程序之间，或分布式系统中发送消息，进行异步通信，JMS是一个与具体平台无关的API，绝大多数MOM提供商都对JMS提供支持。这是比较详细的关于JMS的定义，而比较直观的说，JMS是一组消息服务的API，也就是说JMS只有接口，其具体实现类交给了各种MOM厂家来做。
JMS使用场景，应用程序A部署在北京，应用程序B部署在上海，每当A触发某个事件之后，B向获取A中的一些信息，也可能有很多个B都想获取A中的信息。这种情况下，Java提供了最佳的解决方案-JMS。JMS同样适用于基于事件的应用程序，如聊天服务，他需要一种发布事件机制向所有与服务器连接的客户端发送消息。JMS与RMI不同，不需要接受者在线。也就是服务器发送完消息，这个事件就与他无关了。
JMS的优势：
- 异步，JMS天生就是异步的，客户端获取消息的时候，不需要主动发送请求，消息会自动发送给可用的客户端。
- 可靠，JMS保证消息只会被递送一次。大家都遇到过重复创建消息的问题，JMS可以帮你避免这个问题，但是不能杜绝，需要MOM厂家来做更加完备的机制来改善。
JMS常用的一些概念：
- Provider/MessageProvider：生产者
- Consumer/MessageConsumer：消费者
- PTP：Point To Point，点对点通信消息模型
- Pub/Sub：Publish/Subscribe，发布订阅消息模型
- Queue：队列，目标类型之一，和PTP结合
- Topic：主题，目标类型之一，和Pub/Sub结合
- ConnectionFactory：连接工厂，JMS用它创建连接
- Connnection：JMS Client到JMS Provider的连接
- Destination：消息目的地，由Session创建
- Session：会话，由Connection创建，实质上就是发送、接受消息的一个线程，因此生产者、消费者都是Session创建的

ActiveMQ简介

ActiveMQ是Apache出品的，最流行的，能力强劲的开源消息总线。ActiveMQ是一个完全支持JMS1.1和J2EE1.4规范的JMS Provider实现，JMS上面已经有了简单的介绍。

ActiveMQ的特点：

多种语言和协议编写客户端，语言包括Java、C、C++、C#、Ruby、Perl、Python、PHP，协议包括OpenWire、Stomp、REST、WS Notification、XMPP、AMQP
完全支持JMS1.1和J2EE1.4规范
对Spring的支持，使得ActiveMQ集成到Spring里面很方便
支持多种传送协议：in-VM、TCP、SSL、NIO、UDP、JGroups、JXTA
支持通过JDBC和journal提供高速的消息持久化
从设计上保证了高性能的集群，客户端-服务器点对点
支持Ajax
支持与Axis的整合

ActiveMQ的消息形式

点对点，也就是一个省车按着对一个消费者的一对一
发布/订阅模式，一个生产者产生消息并进行发送后，可以由多个消费者进行接收

JMS中定义了五种不同的消息正文格式以及调用的消息信息，允许你发送并接收以一些不同形式的数据：

* StreamMessage --- 数据流
* MapMessage --- key-value
* TextMessage --- 字符串
* ObjectMessage --- 序列化的Java对象
* BytesMessage --- 一个字节的数据流

ActiveMQ的安装

activemq.apache.org下载ActiveMQ
解压缩
启动， ./activemq start，关闭 ./activemq stop, 查看状态 ./activemq status

后台管理页面：http://192.168.25.168:8161/admin，用户名admin，密码admin

ActiveMQ的使用

添加jar包

<!--引入ActiveMQ的jar包-->
<dependency>
    <groupId>org.apache.activemq</groupId>
    <artifactId>activemq-all</artifactId>
</dependency>

使用Queue形式的消息队列

Producer

/**
 * ActiveMQ队列模式生产者
 * @throws JMSException
 */
@Test
public void testQueueProducer() throws JMSException {
    //1.创建一个连接工厂对象、需要指定IP和端口/消息服务端口为61616
    ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.25.128:61616");
    //2.使用连接工厂来创建连接
    Connection connection = connectionFactory.createConnection();
    //3.开启连接
    connection.start();
    //4.创建一个会话，
    //第一个参数为是否开启ActiveMQ的事务，一般不使用事务
    //如果开启事务，第二个参数自动忽略，不开启事务，第二个参数表示消息的应答模式，自动应答、手动应答
    Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    //5.使用Session对象来创建一个Destination对象，topic或者queue
    Queue queue = session.createQueue("test-Queue");
    //6.使用Session对象来创建一个生产者
    MessageProducer producer = session.createProducer(queue);
    //7.创建一个TextMessage对象
    TextMessage textMessage = new ActiveMQTextMessage();
    textMessage.setText("hello!");
    //8.发送消息
    producer.send(textMessage);
    //9.关闭资源
    producer.close();
    session.close();
    connection.close();
}

可以看出生产者和消费者之间传递的对象，由三个主要部分构成：消息头+消息属性+消息体。当然创建过程中也可以对消息进行持久化的选择等配置。

Consumer

/**
 * ActiveMQ队列模式消费者
 * @throws JMSException
 */
@Test
public void testQueueConsumer() throws JMSException, IOException {
    //1.创建一个连接工厂对象、需要指定IP和端口/消息服务端口为61616
    ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.25.128:61616");
    //2.使用连接工厂来创建连接
    Connection connection = connectionFactory.createConnection();
    //3.开启连接
    connection.start();
    //4.创建一个会话，
    //第一个参数为是否开启ActiveMQ的事务，一般不使用事务
    //如果开启事务，第二个参数自动忽略，不开启事务，第二个参数表示消息的应答模式，自动应答、手动应答
    Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    //5.使用Session对象来创建一个Destination对象，topic或者queue
    Queue queue = session.createQueue("test-Queue");
    //6.使用Session对象来创建一个消费者
    MessageConsumer consumer = session.createConsumer(queue);
    //7.接收消息
    consumer.setMessageListener(new MessageListener() {
        @Override
        public void onMessage(Message message) {
            TextMessage textMessage = (TextMessage) message;
            String text = null;
            try {
                text = textMessage.getText();
            } catch (JMSException e) {
                e.printStackTrace();
            }
            System.out.println(text);
        }
    });
    //8.关闭资源
    System.in.read();
    consumer.close();
    session.close();
    connection.close();
}

注释基本解释清楚了基本的流程，唯一要注意的是接收消息的时候，我们先查看setMessageListener这个方法,其接口定义如下：

1	void setMessageListener(MessageListener var1) throws JMSException;

也就是说需要传入一个实现了MessageListener接口的对象，而MessageListener接口如下只有一个onMessage方法：

package javax.jms;

public interface MessageListener {
    void onMessage(Message var1);
}

采用匿名类的方法来对test-Queue队列进行监控，只要有消息进来，就立即执行onMessage方法。

Topic模式

Provider

/**
 * ActiveMQ的订阅模式生产者
 */
@Test
public void testTopicProducer() throws Exception {
    ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.25.128:61616");
    Connection connection = connectionFactory.createConnection();
    connection.start();
    Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    Topic topic = session.createTopic("test-topic");
    MessageProducer producer = session.createProducer(topic);
    TextMessage activeMQ_topic = session.createTextMessage("activeMQ topic");
    producer.send(activeMQ_topic);

    producer.close();
    session.close();
    connection.close();
}

和队列模式相比，只是由session生成的消息队列模式编程了订阅发布模式，其他完全一样。

Consumer

/**
 * ActiveMQ的订阅者模式消费者
 */
@Test
public void testTopicConsumer() throws Exception{
    ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.25.1 大专栏  ActiveMQ学习总结（一）28:61616");
    Connection connection = connectionFactory.createConnection();
    connection.start();
    Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    Topic topic = session.createTopic("test-topic");
    MessageConsumer consumer = session.createConsumer(topic);
    consumer.setMessageListener(new MessageListener() {
        @Override
        public void onMessage(Message message) {
            if (message instanceof TextMessage) {
                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();
}

ActiveMQ在Spring中的使用

上一节介绍的在项目中直接使用消息队列的方式，可以看出存在很大的重复代码，而且步骤很多，将ActiveMQ整合到Spring中可以大大改善这两个问题。

引入JMS相关jar包

<dependency>
    <groupId>org.springframework</groupId>
    <artifactId>spring-jms</artifactId>
    <version>${spring.version}</version>
</dependency>
<dependency>
    <groupId>org.springframework</groupId>
    <artifactId>spring-context-support</artifactId>
    <version>${spring.version}</version>
</dependency>

配置ActiveMQ整合Spring，配置ConnectionFactory

<!--ConnectionFactory,JMS服务厂商提供的ConnectionFactory-->
<bean id="targetConnecctionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
    <constructor-arg name="brokerURL" value="tcp://192.168.25.128:61616"/>
</bean>
<!--spring对ConnectionFactory的封装-->
<bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
    <property name="targetConnectionFactory" ref="targetConnecctionFactory"/>
</bean>

配置生产者，使用JMSTemplate对象，发送消息

<!--配置JMSTemplate-->
<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
    <property name="connectionFactory" ref="connectionFactory"/>
</bean>

配置Destination

<!--消息的目的地-->
<bean id="test-queue" class="org.apache.activemq.command.ActiveMQQueue">
    <constructor-arg name="name" value="test-queue"/>
</bean>
<bean id="item-add-topic" class="org.apache.activemq.command.ActiveMQTopic">
    <constructor-arg name="name" value="item-add-topic"/>
</bean>

完整的配置文件：

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:context="http://www.springframework.org/schema/context" xmlns:p="http://www.springframework.org/schema/p"
       xmlns:aop="http://www.springframework.org/schema/aop" xmlns:tx="http://www.springframework.org/schema/tx"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-4.2.xsd
	http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-4.2.xsd
	http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-4.2.xsd http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-4.2.xsd
	http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util-4.2.xsd">

    <!--ConnectionFactory,JMS服务厂商提供的ConnectionFactory-->
    <bean id="targetConnecctionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
        <constructor-arg name="brokerURL" value="tcp://192.168.25.128:61616"/>
    </bean>
    <!--spring对ConnectionFactory的封装-->
    <bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
        <property name="targetConnectionFactory" ref="targetConnecctionFactory"/>
    </bean>
    <!--配置JMSTemplate-->
    <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
        <property name="connectionFactory" ref="connectionFactory"/>
    </bean>
    <!--消息的目的地-->
    <bean id="test-queue" class="org.apache.activemq.command.ActiveMQQueue">
        <constructor-arg name="name" value="test-queue"/>
    </bean>
    <bean id="item-add-topic" class="org.apache.activemq.command.ActiveMQTopic">
        <constructor-arg name="name" value="item-add-topic"/>
    </bean>
</beans>

测试代码

发送消息，步骤注释中写的很清楚了，

/**
 * 使用JMSTemplate来发送消息
 * @throws Exception
 */
@Test
public void testJmsTemplate() throws Exception {
    //初始化Spring容器
    ApplicationContext applicationContext = new ClassPathXmlApplicationContext("spring/applicationContext-activemq.xml");
    //从容器中获得模板对象
    JmsTemplate jmsTemplate = applicationContext.getBean(JmsTemplate.class);
    //从容器中获得Destination对象
    Destination destination = (Destination) applicationContext.getBean("test-queue");
    //发送消息
    jmsTemplate.send(destination, new MessageCreator() {
        @Override
        public Message createMessage(Session session) throws JMSException {
            return session.createTextMessage("spring activemq send queue message");
        }
    });
}

接收消息

由于项目中使用ActiveMQ实现索引库和数据库的同步，为了接收ActiveMQ的消息，需要创建一个MessageListener实现类。

/**
 * 接收ActiveMQ发送的消息
 * Created by cdx0312
 * 2018/3/9
 */
public class MyMesseageListener implements MessageListener{
    @Override
    public void onMessage(Message message) {
        //接收消息
        try {
            TextMessage textMessage = (TextMessage) message;
            String text = textMessage.getText();
            System.out.println(text);
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }
}

然后在Spring中整合ActiveMQ的消息监听器

<!--ConnectionFactory,JMS服务厂商提供的ConnectionFactory-->
<bean id="targetConnecctionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
    <constructor-arg name="brokerURL" value="tcp://192.168.25.128:61616"/>
</bean>
<!--spring对ConnectionFactory的封装-->
<bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
    <property name="targetConnectionFactory" ref="targetConnecctionFactory"/>
</bean>
<!--消息的目的地-->
<bean id="test-queue" class="org.apache.activemq.command.ActiveMQQueue">
    <constructor-arg name="name" value="test-queue"/>
</bean>
<bean id="item-add-topic" class="org.apache.activemq.command.ActiveMQTopic">
    <constructor-arg name="name" value="item-add-topic"/>
</bean>
<!--配置消息的接受者-->
<!--配置监听器-->
<bean id="messeageListener" class="com.market.search.listen.MyMesseageListener"/>
<!--消息监听容器-->
<bean class="org.springframework.jms.listener.DefaultMessageListenerContainer">
    <property name="connectionFactory" ref="connectionFactory"/>
    <property name="destination" ref="test-queue"/>
    <property name="messageListener" ref="messeageListener"/>
</bean>

测试代码：

@Test
public void testQueueConsumer() throws Exception {
    //初始化Spring容器
    ApplicationContext applicationContext = new ClassPathXmlApplicationContext("spring/applicationContext-activemq.xml");
    System.in.read();
}

查看全文

相关阅读:
apt-clone安装与使用
 利用异或求（整数数组中，有2K+1个数，其中有2k个相同，找出不相同的那个数）
运行程序，填写结果
 throw与throws的区别
 牛客网多线程程序执行结果选择题
 一个继承了抽象类的普通类的执行顺序
 int i=0;i=i++
HashMap浅入理解
 &&和&、||和|的区别
 System.out.println()

原文地址：https://www.cnblogs.com/lijianming180/p/12046818.html