服务器之家：专注于服务器技术及软件下载分享
分类导航

PHP教程|ASP.NET教程|JAVA教程|ASP教程|编程技术|正则表达式|C/C++|IOS|C#|Swift|Android|JavaScript|易语言|

服务器之家 - 编程语言 - JAVA教程 - Spring Boot 入门之消息中间件的使用

Spring Boot 入门之消息中间件的使用

2021-04-02 15:46moonlightL JAVA教程

本篇文章主要介绍了Spring Boot 入门之消息中间件的使用，小编觉得挺不错的，现在分享给大家，也给大家做个参考。一起跟随小编过来看看吧

一、前言

在消息中间件中有 2 个重要的概念：消息代理和目的地。当消息发送者发送消息后，消息就被消息代理接管，消息代理保证消息传递到指定目的地。

我们常用的消息代理有 JMS 和 AMQP 规范。对应地，它们常见的实现分别是 ActiveMQ 和 RabbitMQ。

二、整合 ActiveMQ

2.1 添加依赖

?
1
2
3
4
5
6
7
8
9
<dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-activemq</artifactId>
</dependency>
<!-- 如果需要配置连接池，添加如下依赖 -->
<dependency>
  <groupId>org.apache.activemq</groupId>
  <artifactId>activemq-pool</artifactId>
</dependency>

2.2 添加配置

?
1
2
3
4
5
6
7
8
# activemq 配置
spring.activemq.broker-url=tcp://192.168.2.12:61616
spring.activemq.user=admin
spring.activemq.password=admin
spring.activemq.pool.enabled=false
spring.activemq.pool.max-connections=50
# 使用发布/订阅模式时，下边配置需要设置成 true
spring.jms.pub-sub-domain=false

此处 spring.activemq.pool.enabled=false，表示关闭连接池。

2.3 编码

配置类：

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
@Configuration
public class JmsConfirguration {
  public static final String QUEUE_NAME = "activemq_queue";
  
  public static final String TOPIC_NAME = "activemq_topic";
  
  @Bean
  public Queue queue() {
    return new ActiveMQQueue(QUEUE_NAME);
  }
  
  @Bean
  public Topic topic() {
    return new ActiveMQTopic(TOPIC_NAME);
  }
}

负责创建队列和主题。

消息生产者：

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
@Component
public class JmsSender {
  @Autowired
  private Queue queue;
  
  @Autowired
  private Topic topic;
  
  @Autowired
  private JmsMessagingTemplate jmsTemplate;
  
  public void sendByQueue(String message) {
    this.jmsTemplate.convertAndSend(queue, message);
  }
  
  public void sendByTopic(String message) {
    this.jmsTemplate.convertAndSend(topic, message);
  }
}

消息消费者：

?
1
2
3
4
5
6
7
8
9
10
11
12
13
@Component
public class JmsReceiver {
  
  @JmsListener(destination = JmsConfirguration.QUEUE_NAME)
  public void receiveByQueue(String message) {
    System.out.println("接收队列消息:" + message);
  }
  
  @JmsListener(destination = JmsConfirguration.TOPIC_NAME)
  public void receiveByTopic(String message) {
    System.out.println("接收主题消息:" + message);
  }
}

消息消费者使用 @JmsListener 注解监听消息。

2.4 测试

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
@RunWith(SpringRunner.class)
@SpringBootTest
public class JmsTest {
  @Autowired
  private JmsSender sender;
  @Test
  public void testSendByQueue() {
    for (int i = 1; i < 6; i++) {
      this.sender.sendByQueue("hello activemq queue " + i);
    }
  }
  
  @Test
  public void testSendByTopic() {
    for (int i = 1; i < 6; i++) {
      this.sender.sendByTopic("hello activemq topic " + i);
    }
  }
}

打印结果：

接收队列消息:hello activemq queue 1
接收队列消息:hello activemq queue 2
接收队列消息:hello activemq queue 3
接收队列消息:hello activemq queue 4
接收队列消息:hello activemq queue 5

测试发布/订阅模式时，设置 spring.jms.pub-sub-domain=true

接收主题消息:hello activemq topic 1
接收主题消息:hello activemq topic 2
接收主题消息:hello activemq topic 3
接收主题消息:hello activemq topic 4
接收主题消息:hello activemq topic 5

三、整合 RabbitMQ

3.1 添加依赖

?
1
2
3
4
<dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

3.2 添加配置

?
1
2
3
4
5
spring.rabbitmq.host=192.168.2.30
spring.rabbitmq.port=5672
spring.rabbitmq.username=light
spring.rabbitmq.password=light
spring.rabbitmq.virtual-host=/test

3.3 编码

配置类：

?
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
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
@Configuration
public class AmqpConfirguration {
  //=============简单、工作队列模式===============
  
  public static final String SIMPLE_QUEUE = "simple_queue";
  @Bean
  public Queue queue() {
    return new Queue(SIMPLE_QUEUE, true);
  }
  
  //===============发布/订阅模式============
  
  public static final String PS_QUEUE_1 = "ps_queue_1";
  public static final String PS_QUEUE_2 = "ps_queue_2";
  public static final String FANOUT_EXCHANGE = "fanout_exchange";
  
  @Bean
  public Queue psQueue1() {
    return new Queue(PS_QUEUE_1, true);
  }
  
  @Bean
  public Queue psQueue2() {
    return new Queue(PS_QUEUE_2, true);
  }
  
  @Bean
  public FanoutExchange fanoutExchange() {
    return new FanoutExchange(FANOUT_EXCHANGE);
  }
  
  @Bean
  public Binding fanoutBinding1() {
    return BindingBuilder.bind(psQueue1()).to(fanoutExchange());
  }
  
  @Bean
  public Binding fanoutBinding2() {
    return BindingBuilder.bind(psQueue2()).to(fanoutExchange());
  }
  //===============路由模式============
  
  public static final String ROUTING_QUEUE_1 = "routing_queue_1";
  public static final String ROUTING_QUEUE_2 = "routing_queue_2";
  public static final String DIRECT_EXCHANGE = "direct_exchange";
  
  @Bean
  public Queue routingQueue1() {
    return new Queue(ROUTING_QUEUE_1, true);
  }
  
  @Bean
  public Queue routingQueue2() {
    return new Queue(ROUTING_QUEUE_2, true);
  }
  
  @Bean
  public DirectExchange directExchange() {
    return new DirectExchange(DIRECT_EXCHANGE);
  }
  
  @Bean
  public Binding directBinding1() {
    return BindingBuilder.bind(routingQueue1()).to(directExchange()).with("user");
  }
  
  @Bean
  public Binding directBinding2() {
    return BindingBuilder.bind(routingQueue2()).to(directExchange()).with("order");
  }
  
  //===============主题模式============
  
  public static final String TOPIC_QUEUE_1 = "topic_queue_1";
  public static final String TOPIC_QUEUE_2 = "topic_queue_2";
  public static final String TOPIC_EXCHANGE = "topic_exchange";
  
  @Bean
  public Queue topicQueue1() {
    return new Queue(TOPIC_QUEUE_1, true);
  }
  
  @Bean
  public Queue topicQueue2() {
    return new Queue(TOPIC_QUEUE_2, true);
  }
  
  @Bean
  public TopicExchange topicExchange() {
    return new TopicExchange(TOPIC_EXCHANGE);
  }
  
  @Bean
  public Binding topicBinding1() {
    return BindingBuilder.bind(topicQueue1()).to(topicExchange()).with("user.add");
  }
  
  @Bean
  public Binding topicBinding2() {
    return BindingBuilder.bind(topicQueue2()).to(topicExchange()).with("user.#");
  } 
}

RabbitMQ 有多种工作模式，因此配置比较多。想了解相关内容的读者可以查看《RabbitMQ 工作模式介绍》或者自行百度相关资料。

消息生产者：

?
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
34
35
36
37
38
@Component
public class AmqpSender {
  @Autowired
  private AmqpTemplate amqpTemplate;
  /**
   * 简单模式发送
   *
   * @param message
   */
  public void simpleSend(String message) {
    this.amqpTemplate.convertAndSend(AmqpConfirguration.SIMPLE_QUEUE, message);
  }
  /**
   * 发布/订阅模式发送
   *
   * @param message
   */
  public void psSend(String message) {
    this.amqpTemplate.convertAndSend(AmqpConfirguration.FANOUT_EXCHANGE, "", message);
  }
  /**
   * 路由模式发送
   *
   * @param message
   */
  public void routingSend(String routingKey, String message) {
    this.amqpTemplate.convertAndSend(AmqpConfirguration.DIRECT_EXCHANGE, routingKey, message);
  }
  /**
   * 主题模式发送
   *
   * @param routingKey
   * @param message
   */
  public void topicSend(String routingKey, String message) {
    this.amqpTemplate.convertAndSend(AmqpConfirguration.TOPIC_EXCHANGE, routingKey, message);
  }
}

消息消费者：

?
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
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
@Component
public class AmqpReceiver {
  /**
   * 简单模式接收
   *
   * @param message
   */
  @RabbitListener(queues = AmqpConfirguration.SIMPLE_QUEUE)
  public void simpleReceive(String message) {
    System.out.println("接收消息:" + message);
  }
  /**
   * 发布/订阅模式接收
   *
   * @param message
   */
  @RabbitListener(queues = AmqpConfirguration.PS_QUEUE_1)
  public void psReceive1(String message) {
    System.out.println(AmqpConfirguration.PS_QUEUE_1 + "接收消息:" + message);
  }
  @RabbitListener(queues = AmqpConfirguration.PS_QUEUE_2)
  public void psReceive2(String message) {
    System.out.println(AmqpConfirguration.PS_QUEUE_2 + "接收消息:" + message);
  }
  /**
   * 路由模式接收
   *
   * @param message
   */
  @RabbitListener(queues = AmqpConfirguration.ROUTING_QUEUE_1)
  public void routingReceive1(String message) {
    System.out.println(AmqpConfirguration.ROUTING_QUEUE_1 + "接收消息:" + message);
  }
  @RabbitListener(queues = AmqpConfirguration.ROUTING_QUEUE_2)
  public void routingReceive2(String message) {
    System.out.println(AmqpConfirguration.ROUTING_QUEUE_2 + "接收消息:" + message);
  }
  /**
   * 主题模式接收
   *
   * @param message
   */
  @RabbitListener(queues = AmqpConfirguration.TOPIC_QUEUE_1)
  public void topicReceive1(String message) {
    System.out.println(AmqpConfirguration.TOPIC_QUEUE_1 + "接收消息:" + message);
  }
  
  @RabbitListener(queues = AmqpConfirguration.TOPIC_QUEUE_2)
  public void topicReceive2(String message) {
    System.out.println(AmqpConfirguration.TOPIC_QUEUE_2 + "接收消息:" + message);
  }
}

消息消费者使用 @RabbitListener 注解监听消息。

3.4 测试

?
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
@RunWith(SpringRunner.class)
@SpringBootTest
public class AmqpTest {
  @Autowired
  private AmqpSender sender;
  @Test
  public void testSimpleSend() {
    for (int i = 1; i < 6; i++) {
      this.sender.simpleSend("test simpleSend " + i);
    }
  }
  @Test
  public void testPsSend() {
    for (int i = 1; i < 6; i++) {
      this.sender.psSend("test psSend " + i);
    }
  }
  
  @Test
  public void testRoutingSend() {
    for (int i = 1; i < 6; i++) {
      this.sender.routingSend("order", "test routingSend " + i);
    }
  }
  
  @Test
  public void testTopicSend() {
    for (int i = 1; i < 6; i++) {
      this.sender.topicSend("user.add", "test topicSend " + i);
    }
  }
}

测试结果略过。。。

踩坑提醒1：ACCESS_REFUSED – Login was refused using authentication mechanism PLAIN

解决方案：

1) 请确保用户名和密码是否正确，需要注意的是用户名和密码的值是否包含空格或制表符（笔者测试时就是因为密码多了一个制表符导致认证失败）。

2) 如果测试账户使用的是 guest，需要修改 rabbitmq.conf 文件。在该文件中添加 “loopback_users = none” 配置。

踩坑提醒2：Cannot prepare queue for listener. Either the queue doesn't exist or the broker will not allow us to use it

解决方案：

我们可以登陆 RabbitMQ 的管理界面，在 Queue 选项中手动添加对应的队列。

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持服务器之家。

原文链接：https://blog.phpsoho.com/2018/02/08/spring-boot-入门之消息中间件

  • spring boot
  • 消息中间件

    • 延伸 · 阅读

    精彩推荐
    • JAVA教程java中的转义字符介绍

      java中的转义字符介绍

      普通的转义字符序列和八进制转义字符都比Unicode转义字符要好得多，因为与Unicode转义字符不同，转义字符序列是在程序被解析为各种符号之后被处理的 ...

      java之家4392019-10-14
    • JAVA教程详解Junit 测试之 Spring Test

      详解Junit 测试之 Spring Test

      本篇文章主要介绍了Junit 测试之 Spring Test，小编觉得挺不错的，现在分享给大家，也给大家做个参考。一起跟随小编过来看看吧...

      jijs6912021-01-23
    • JAVA教程Java编程中的一些常见问题汇总

      Java编程中的一些常见问题汇总

      这篇文章主要介绍了Java编程中的一些常见问题汇总,本文总结的都是一些Java代码中比较典型的错误,需要的朋友可以参考下 ...

      junjie3592019-12-01
    • JAVA教程Java单例模式简单示例

      Java单例模式简单示例

      这篇文章主要介绍了Java单例模式,结合实例形式简单分析了java单例模式的定义与使用技巧,需要的朋友可以参考下...

      qq73422724972020-11-24
    • JAVA教程老生常谈比较排序之堆排序

      老生常谈比较排序之堆排序

      下面小编就为大家带来一篇老生常谈比较排序之堆排序。小编觉得挺不错的，现在就分享给大家，也给大家做个参考。一起跟随小编过来看看吧...

      Java教程网4792020-11-23
    • JAVA教程java直接插入排序示例

      java直接插入排序示例

      这篇文章主要介绍了java直接插入排序示例,插入排序的比较次数仍然是n的平方，但在一般情况下，它要比冒泡排序快一倍，比选择排序还要快一点。它常常...

      Java教程网2922019-11-22
    • JAVA教程30分钟入门Java8之默认方法和静态接口方法学习

      30分钟入门Java8之默认方法和静态接口方法学习

      这篇文章主要介绍了30分钟入门Java8之默认方法和静态接口方法学习,详细介绍了默认方法和接口，有兴趣的可以了解一下。...

      JohnTsai5152020-09-07
    • JAVA教程浅谈Java自动装箱与拆箱及其陷阱

      浅谈Java自动装箱与拆箱及其陷阱

      下面小编就为大家带来一篇浅谈Java自动装箱与拆箱及其陷阱。小编觉得挺不错的，现在就分享给大家，也给大家做个参考。一起跟随小编过来看看吧 ...

      java教程网4542020-05-14
    最近更新
    编辑推荐
    阅读排行
    热门标签
    501 　 406 　 2000 　 1433 　 415 　 401 　 空心菱形 　 选择排序法 　 随机抽取 　 内存卡 　 下载器 　 十六进制字符 　 响应 　 静态工厂 　 构造函数 　 swt 　 Overview 　 对象池模式 　 子线程 　 短信猫 　 htmlparser 　 支票 　 金额大写转换 　 消息中间件 　 AspectJ 　 面向切面编程 　 方法名 　 网址信息 　 动态代理模式 　 线程安全 　
    © 2019-2021 服务器之家(www.zzvips.com) 版权所有关于我们联系我们版权申明网站地图