SpringBoot JMS(ActiveMQ) 使用实践

ActiveMQ

1. 下载windows办的activeMQ后,在以下目录可以启动:

2. 启动后会有以下提示

3. 所以我们可以通过http://localhost:8161访问管理页面,通过tcp://localhost:61616来连接消息服务器,用到的用户名和密码都在以下文件中(默认为admin=admin)

springboot连接ActiveMQ

1. 加入依赖:

            spring-boot-starter-activemq

2. 配置连接属性:

spring.activemq.broker-url=tcp://localhost:61616
spring.activemq.user=admin
spring.activemq.password=admin
spring.activemq.pool.enabled=false

消息的发送和接收

生产者/消费者模式

    1. 创建生产者

package com.example.demo8activemq.jms;

import org.apache.activemq.command.ActiveMQQueue;
import org.springframework.jms.core.JmsMessagingTemplate;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import javax.jms.Destination;

/**
 * @author Created by yawn on 2017-10-26 16:15
 */
@Service
public class Producer {

    @Resource
    private JmsMessagingTemplate jmsMessagingTemplate;

    public void sendMsg(String destinationName, String message) {
        System.out.println("============>>>>> 发送queue消息 " + message);
        Destination destination = new ActiveMQQueue(destinationName);
        jmsMessagingTemplate.convertAndSend(destination, message);
    }
}

    2. 创建消费者

package com.example.demo8activemq.jms;

import org.springframework.jms.annotation.JmsListener;
import org.springframework.stereotype.Service;

/**
 * @author Created by yawn on 2017-10-26 16:15
 */
@Service
public class Consumer {

    @JmsListener(destination = "test.queue")
    public void receiveMsg(String text) {
        System.out.println("<<<<<<============ 收到消息: " + text);
    }
}

    注意: @JmsListener是一个可重复的注解,在java7及以下版本jdk中,可以使用@JmsListeners代替它。

    3. 测试类

@RunWith(SpringRunner.class)
@SpringBootTest
public class Demo8ActivemqApplicationTests {

	@Resource
	private Producer producer;

	@Test
	public void contextLoads() {
		for (int i = 0; i < 10; i++) {
			producer.sendMsg("test.queue", "Queue Message " + i);
		}
	}
}

    4. 运行测试

发布/订阅模式

    1. 发布话题

package com.example.demo8activemq.jms;

import org.apache.activemq.command.ActiveMQTopic;
import org.springframework.jms.core.JmsMessagingTemplate;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import javax.jms.Destination;

/**
 * @author Created by yawn on 2017-10-28 17:09
 */
@Service
public class Publisher {

    @Resource
    private JmsMessagingTemplate jmsMessagingTemplate;

    public void publish(String destinationName, String message) {
        Destination destination = new ActiveMQTopic(destinationName);
        System.out.println("============>>>>> 发布topic消息 " + message);
        jmsMessagingTemplate.convertAndSend(destination, message);
    }
}

    2. 订阅话题

package com.example.demo8activemq.jms;

import org.springframework.jms.annotation.JmsListener;
import org.springframework.stereotype.Service;

/**
 * @author Created by yawn on 2017-10-28 17:15
 */
@Service
public class Subscriber {

    @JmsListener(destination = "test.topic", containerFactory = "myJmsContainerFactory")
    public void subscribe(String text) {
        System.out.println("===========<<<<<<<<收到订阅的消息" + text);
    }
}

    注意: 在pub/sub模式中,对消息的监听需要对containerFactory进行以下配置

    @Bean
    JmsListenerContainerFactory<?> myJmsContainerFactory(ConnectionFactory connectionFactory){
        SimpleJmsListenerContainerFactory factory = new SimpleJmsListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setPubSubDomain(true);
        return factory;
    }

    3. 测试

    @Test
	public void test() {
		for (int i = 0; i < 10; i++) {
			publisher.publish("test.topic", "Topic Message " + i);
		}
	}

应用

    按照以上步骤,在springboot中很容易就实现类两种模式的消息发送和接收。但是jms具体的应用场景是在不同的应用程序之间,生产者和消费者往往是在不同的应用中的。此外,以上例子中的消息我们只发送字符串,其实还可以发送Object类型的消息,甚至可以使用messageCreator自定义消息的转换,而不使用convertAndSend方法默认转换。

多个应用程序之间发送消息

1. 先使用一个只有发送者,没有消费者或订阅者的应用发送两类消息各十条

2. 我们打开localhost:8161,可以看到

    两类都曾有十条消息入队,但只有queues中还存留10条消息。

3. 现在我们启动包含消费者和订阅者的应用程序

    果然,只有消费者收到了queues中的消息。

    这说明订阅者接收topic是需要在topic发布之前订阅;而生产/消费模式下,消息发出后会存放在队列中,等待消费者消费。

    4. 我们先启动两个包含订阅者和消费者的程序,再发布消息

    两个订阅者都收到 topic message 1~9, 而消费者中,一个收到消息 1、3、5、7、9,另一个收到0、2、4、6、8。

    这说明有多个消息接收者时,生产/消费模式下多个消费者会轮流消费队列中的消息,而pub/sub模式下所有订阅者都会得到所有的消息。

    以上就是在多个应用程序之间验证了发布/订阅模式和生产/消费模式的不同特点。

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏行者悟空

Spark的容错机制

1344
来自专栏芋道源码1024

【消息队列 MQ 专栏】消息队列之 RocketMQ

RocketMQ 是阿里巴巴在2012年开源的分布式消息中间件,目前已经捐赠给 Apache 软件基金会,并于2017年9月25日成为 Apache 的顶级项目...

1220
来自专栏Spark学习技巧

Kafka源码系列之源码分析zookeeper在kafka的作用

1023
来自专栏Java架构沉思录

Kafka基本原理

Apache Kafka是分布式发布-订阅消息系统。它最初由LinkedIn公司开发,之后成为Apache项目的一部分。Kafka是一种快速、可扩展的、分布式的...

601
来自专栏张善友的专栏

Enterprise Library 4 缓存应用程序块的设计

缓存应用程序为以下目的而设计: 提供一个大小可管理的 API 集合。 允许开发人员添加标准的缓存操作到他们的应用程序中,而不用学习应用程序块的内部工作。 ...

1776
来自专栏程序猿DD

分布式消息队列 RocketMQ 源码分析 —— Message 拉取与消费(下)

前传:分布式消息队列 RocketMQ 源码分析 —— Message 拉取与消费(上) 本文主要基于 RocketMQ 4.0.x 正式版 1、概述 2、Co...

43010
来自专栏aoho求索

Spring Cloud Stream应用与自定义RocketMQ Binder:编程模型

前言: 本文作者张天,节选自笔者与其合著的《Spring Cloud微服务架构进阶》,即将在八月出版问世。将其中Spring Cloud Stream应用与自定...

1042
来自专栏匠心独运的博客

消息中间件—RocketMQ消息消费(一)

文章摘要:在发送消息给RocketMQ后,消费者需要消费。消息的消费比发送要复杂一些,那么RocketMQ是如何来做的呢? 在RocketMQ系列文章的前面几...

723
来自专栏Spark学习技巧

Kafka源码系列之源码分析zookeeper在kafka的作用

浪尖的kafka源码系列以kafka0.8.2.2源码为例给大家进行讲解的。纯属个人爱好,希望大家对不足之处批评指正。 一,zookeeper在分布式集群的作...

27510
来自专栏Jaycekon

Java消息队列--JMS概述

1、什么是JMS  JMS即Java消息服务(Java Message Service)应用程序接口,是一个Java平台中关于面向消息中间件(MOM)的API...

3145

扫描关注云+社区