高效的消息中间件-ActiveMQ

概述

  JMS(Java Messaging Service)是 Java 平台上有关面向消息中间件的技术规范,它便于消息系统中的 Java 应用程序进行消息交换,并且通过提供标准的产生、发送、接收消息的接口简化企业应用的开发。

  JMS 本身只定义了一系列的接口规范,是一种与厂商无关的 API,用来访问消息收发系统。它类似于 JDBC(java Database Connectivity):这里,JDBC 是可以用来访问许多不同关系数据库的 API,而 JMS 则提供同样与厂商无关的访问方法,以访问消息收发服务。

JMS的优势

  • 间接通信。客户机不直接对话,而是通过消息中间件。
  • 持久性。 JMS的提供者能够持久化存储消息,直到客户机接受到消息。
  • 异步通信。JMS通信是异步的,当客户端获取消息时,不用主动发送请求,消息会自动发送给可用客户端,实现消息的高效传递。
  • 通用性。使用JMS,可以使用相同API访问ActiveMQ、RabbitMQ等消息服务。

JMS消息传递类型

点对点(PTP)

  • 一个生产者和一个消费者一一对应

发布订阅(pub/sub)

  • 一个生产者产生消息并进行发送后,可以由多个消费者进行接收。

ActiveMQ快速入门

点对点模式

  • 创建生产者QueueProducer
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
public class QueueProducer {
public static void main(String[] args) throws JMSException {
//1.创建连接工厂
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.25.128:61616");
//2.创建连接
Connection connection = connectionFactory.createConnection();
//3.启动连接
connection.start();
//4.获取session(会话对象) 参数1:是否启用事务(false为自动commit) 参数2:消息确认模式
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
//5.创建队列对象
Queue queue = session.createQueue("test-queue");
//6.创建消息生产者对象
MessageProducer producer = session.createProducer(queue);
//7.创建消息对象(文本消息)
TextMessage textMessage = session.createTextMessage("消息对象创建了...");
//8.发送消息
producer.send(textMessage);
//9.关闭资源
producer.close();
session.close();
connection.close();
}
}
  • 先执行QueueProducer,创建完成后,可以在Queue中看到该消息已暂存在队列中:
  • 创建消费者QueueConsumer

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    public class QueueConsumer {
    public static void main(String[] args) throws JMSException, IOException {
    //1.创建连接工厂
    ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.25.128:61616");
    //2.创建连接
    Connection connection = connectionFactory.createConnection();
    //3.启动连接
    connection.start();
    //4.获取session(会话对象) 参数1:是否启用事务(false为自动commit) 参数2:消息确认模式
    Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    //5.创建队列对象
    Queue queue = session.createQueue("test-queue");//注意这里队列名称必须和生产者一样
    //6.创建消息消费者对象
    MessageConsumer consumer = session.createConsumer(queue);
    //7.设置监听
    consumer.setMessageListener(new MessageListener() {
    public void onMessage(Message message) {
    TextMessage textMessage = (TextMessage) message;
    try {
    System.out.println("提取消息中的文本:"+textMessage.getText());
    } catch (JMSException e) {
    e.printStackTrace();
    }
    }
    });
    //8.等待键盘输入
    System.in.read();
    //9.关闭资源
    consumer.close();
    session.close();
    connection.close();
    }
    }
  • 再执行QueueConsumer,成功取出了队列中的消息:

  • 打开网页客户端,消息被取出:
  • 第二次执行QueueConsumer,无法获取消息:

发布订阅模式

  • 生产者TopicProducer:只需把点对点模式中的Queue对象改为创建Topic对象即可

    1
    2
    3
    4
    //5.创建主题对象
    Topic topic = session.createTopic("test-topic");
    //6.创建消息生产者对象
    MessageProducer producer = session.createProducer(topic);
  • 先执行生产者,再执行消费者,则无法收到消息

  • 先执行多个消费者,再执行生产者:两个消费者都收到了消息(类似广播)

  • 结论:
    • 发布/订阅模式消费者必须在生产者发出消息的同时接收消息,才能接收到
    • 发布/订阅模式可以有多个执行方法的消费者同时获得消息
    • 点对点模式生产者发出消息后可以暂存在客户端,等待消费者执行方法获取消息
    • 点对点模式只能有一个最早执行方法的消费者获得消息

Spring整合JMS

点对点模式

  1. 建立springjms_producer工程

  2. 导入依赖

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
     <properties>
    <spring.version>4.2.4.RELEASE</spring.version>
    </properties>

    <dependencies>
    <dependency>
    <groupId>org.springframework</groupId>
    <artifactId>spring-jms</artifactId>
    <version>${spring.version}</version>
    </dependency>
    <dependency>
    <groupId>org.springframework</groupId>
    <artifactId>spring-test</artifactId>
    <version>${spring.version}</version>
    </dependency>
    <dependency>
    <groupId>junit</groupId>
    <artifactId>junit</artifactId>
    <version>4.9</version>
    </dependency>
    <dependency>
    <groupId>org.apache.activemq</groupId>
    <artifactId>activemq-client</artifactId>
    <version>5.13.4</version>
    </dependency>
    </dependencies>
  1. applicationContext-jms-producer.xml配置文件

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    <context:component-scan base-package="com.wzc.demo"></context:component-scan>     

    <!-- 真正可以产生Connection的ConnectionFactory,由对应的 JMS服务厂商提供-->
    <bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
    <property name="brokerURL" value="tcp://192.168.25.128:61616"/>
    </bean>

    <!-- Spring用于管理真正的ConnectionFactory的ConnectionFactory -->
    <bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
    <!-- 目标ConnectionFactory对应真实的可以产生JMS Connection的ConnectionFactory -->
    <property name="targetConnectionFactory" ref="targetConnectionFactory"/>
    </bean>

    <!-- Spring提供的JMS工具类,它可以进行消息发送、接收等 -->
    <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
    <!-- 这个connectionFactory对应的是我们定义的Spring提供的那个ConnectionFactory对象 -->
    <property name="connectionFactory" ref="connectionFactory"/>
    </bean>
    <!--这个是队列目的地,点对点的 文本信息-->
    <bean id="queueTextDestination" class="org.apache.activemq.command.ActiveMQQueue">
    <constructor-arg value="queue_text"/>
    </bean>
  1. 生产者QueueProducer

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    @Component
    public class QueueProducer {

    @Autowired
    private JmsTemplate jmsTemplate;

    @Autowired
    private Destination queueTextDestination;

    /**
    * 发送文本消息
    * @param text
    */
    public void sendTextMessage(final String text) {
    jmsTemplate.send(queueTextDestination, new MessageCreator() {

    public Message createMessage(Session session) throws JMSException {

    return session.createTextMessage(text);
    }
    });
    }
    }
  1. 测试类TestQueue

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    @RunWith(SpringJUnit4ClassRunner.class)
    @ContextConfiguration(locations="classpath:applicationContext-jms-producer.xml")
    public class TestQueue {

    @Autowired
    private QueueProducer queueProducer;

    @Test
    public void testSend() {
    queueProducer.sendTextMessage("SpringJMS点对点测试");
    }
    }

运行测试类,可以看到消息队列成功存储:

  1. 建立springjms_consumer工程

  2. 导入依赖(同生产者)

  3. applicationContext-jms-consumer-queue.xml配置文件

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    <!-- 真正可以产生Connection的ConnectionFactory,由对应的 JMS服务厂商提供-->  
    <bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
    <property name="brokerURL" value="tcp://192.168.25.128:61616"/>
    </bean>

    <!-- Spring用于管理真正的ConnectionFactory的ConnectionFactory -->
    <bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
    <!-- 目标ConnectionFactory对应真实的可以产生JMS Connection的ConnectionFactory -->
    <property name="targetConnectionFactory" ref="targetConnectionFactory"/>
    </bean>

    <!--这个是队列目的地,点对点的 文本信息-->
    <bean id="queueTextDestination" class="org.apache.activemq.command.ActiveMQQueue">
    <constructor-arg value="queue_text"/>
    </bean>

    <!-- 自定义监听类 -->
    <bean id="myMessageListener" class="com.wzc.demo.MyMessageListener"></bean>
    <!-- 消息监听容器 -->
    <bean class="org.springframework.jms.listener.DefaultMessageListenerContainer">
    <property name="connectionFactory" ref="connectionFactory" />
    <property name="destination" ref="queueTextDestination" />
    <property name="messageListener" ref="myMessageListener" />
    </bean>
  1. 自定义监听类MyMessageListener,该类要实现MessageListener接口

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    public class MyMessageListener implements MessageListener{
    public void onMessage(Message message) {
    TextMessage textMessage = (TextMessage) message;
    try {
    System.out.println("收到消息:"+textMessage.getText());
    } catch (JMSException e) {
    e.printStackTrace();
    }
    }
    }
  1. 测试类TestQueue

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    @RunWith(SpringJUnit4ClassRunner.class)
    @ContextConfiguration(locations="classpath:applicationContext-jms-consumer-queue.xml")
    public class TestQueue {
    @Test
    public void testQueue() {
    try {
    System.in.read();
    } catch (IOException e) {
    // TODO Auto-generated catch block
    e.printStackTrace();
    }
    }
    }

    运行测试类,可以看到消费者成功接收消息:

发布/订阅模式

  1. applicationContext-jms-producer.xml配置文件增加发布订阅模式文本信息

    1
    2
    3
    4
    <!-- 这个是订阅模式文本信息 -->
    <bean id="topicTextDestination" class="org.apache.activemq.command.ActiveMQTopic">
    <constructor-arg value="topic_text"/>
    </bean>
  2. 生产者TopicProducer(同QueueProducer,把Destination对象名中的queue替换为topic)

  3. 测试类TestTopic(同TestQueue,把QueueProducer替换为TopicProducer)

运行测试类,成功发送消息

  1. applicationContext-jms-consumer-topic.xml配置文件修改点对点模式为发布订阅模式

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    <!--这个是队列目的地,发布订阅的  文本信息-->  
    <bean id="topicTextDestination" class="org.apache.activemq.command.ActiveMQTopic">
    <constructor-arg value="topic_text"/>
    </bean>
    <!-- 消息监听容器 -->
    <bean class="org.springframework.jms.listener.DefaultMessageListenerContainer">
    <property name="connectionFactory" ref="connectionFactory" />
    <property name="destination" ref="topicTextDestination" />
    <property name="messageListener" ref="myMessageListener" />
    </bean>
  2. 测试类TestTopic(修改对应配置文件路径和方法名即可)

启动3个消费者测试类,再执行生产者中的测试类TestTopic,三个消费者测试类同时接收到消息

0%