国产片侵犯亲女视频播放_亚洲精品二区_在线免费国产视频_欧美精品一区二区三区在线_少妇久久久_在线观看av不卡

服務(wù)器之家：專注于服務(wù)器技術(shù)及軟件下載分享
分類導(dǎo)航

PHP教程|ASP.NET教程|Java教程|ASP教程|編程技術(shù)|正則表達(dá)式|C/C++|IOS|C#|Swift|Android|VB|R語(yǔ)言|JavaScript|易語(yǔ)言|vb.net|

服務(wù)器之家 - 編程語(yǔ)言 - Java教程 - Spring Boot 入門之消息中間件的使用

Spring Boot 入門之消息中間件的使用

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

本篇文章主要介紹了Spring Boot 入門之消息中間件的使用，小編覺(jué)得挺不錯(cuò)的，現(xiàn)在分享給大家，也給大家做個(gè)參考。一起跟隨小編過(guò)來(lái)看看吧

一、前言

在消息中間件中有 2 個(gè)重要的概念：消息代理和目的地。當(dāng)消息發(fā)送者發(fā)送消息后，消息就被消息代理接管，消息代理保證消息傳遞到指定目的地。

我們常用的消息代理有 JMS 和 AMQP 規(guī)范。對(duì)應(yīng)地，它們常見(jiàn)的實(shí)現(xiàn)分別是 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
# 使用發(fā)布/訂閱模式時(shí)，下邊配置需要設(shè)置成 true
spring.jms.pub-sub-domain=false

此處 spring.activemq.pool.enabled=false，表示關(guān)閉連接池。

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);
  }
}

負(fù)責(zé)創(chuàng)建隊(duì)列和主題。

消息生產(chǎn)者：

?
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);
  }
}

消息消費(fèi)者：

?
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("接收隊(duì)列消息:" + message);
  }
  
  @JmsListener(destination = JmsConfirguration.TOPIC_NAME)
  public void receiveByTopic(String message) {
    System.out.println("接收主題消息:" + message);
  }
}

消息消費(fèi)者使用 @JmsListener 注解監(jiān)聽(tīng)消息。

2.4 測(cè)試

?
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);
    }
  }
}

打印結(jié)果：

接收隊(duì)列消息:hello activemq queue 1
接收隊(duì)列消息:hello activemq queue 2
接收隊(duì)列消息:hello activemq queue 3
接收隊(duì)列消息:hello activemq queue 4
接收隊(duì)列消息:hello activemq queue 5

測(cè)試發(fā)布/訂閱模式時(shí)，設(shè)置 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 {
  //=============簡(jiǎn)單、工作隊(duì)列模式===============
  
  public static final String SIMPLE_QUEUE = "simple_queue";
  @Bean
  public Queue queue() {
    return new Queue(SIMPLE_QUEUE, true);
  }
  
  //===============發(fā)布/訂閱模式============
  
  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 有多種工作模式，因此配置比較多。想了解相關(guān)內(nèi)容的讀者可以查看《RabbitMQ 工作模式介紹》或者自行百度相關(guān)資料。

消息生產(chǎn)者：

?
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;
  /**
   * 簡(jiǎn)單模式發(fā)送
   *
   * @param message
   */
  public void simpleSend(String message) {
    this.amqpTemplate.convertAndSend(AmqpConfirguration.SIMPLE_QUEUE, message);
  }
  /**
   * 發(fā)布/訂閱模式發(fā)送
   *
   * @param message
   */
  public void psSend(String message) {
    this.amqpTemplate.convertAndSend(AmqpConfirguration.FANOUT_EXCHANGE, "", message);
  }
  /**
   * 路由模式發(fā)送
   *
   * @param message
   */
  public void routingSend(String routingKey, String message) {
    this.amqpTemplate.convertAndSend(AmqpConfirguration.DIRECT_EXCHANGE, routingKey, message);
  }
  /**
   * 主題模式發(fā)送
   *
   * @param routingKey
   * @param message
   */
  public void topicSend(String routingKey, String message) {
    this.amqpTemplate.convertAndSend(AmqpConfirguration.TOPIC_EXCHANGE, routingKey, message);
  }
}

消息消費(fèi)者：

?
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 {
  /**
   * 簡(jiǎn)單模式接收
   *
   * @param message
   */
  @RabbitListener(queues = AmqpConfirguration.SIMPLE_QUEUE)
  public void simpleReceive(String message) {
    System.out.println("接收消息:" + message);
  }
  /**
   * 發(fā)布/訂閱模式接收
   *
   * @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);
  }
}

消息消費(fèi)者使用 @RabbitListener 注解監(jiān)聽(tīng)消息。

3.4 測(cè)試

?
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);
    }
  }
}

測(cè)試結(jié)果略過(guò)。。。

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

解決方案：

1) 請(qǐng)確保用戶名和密碼是否正確，需要注意的是用戶名和密碼的值是否包含空格或制表符（筆者測(cè)試時(shí)就是因?yàn)槊艽a多了一個(gè)制表符導(dǎo)致認(rèn)證失敗）。

2) 如果測(cè)試賬戶使用的是 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 選項(xiàng)中手動(dòng)添加對(duì)應(yīng)的隊(duì)列。

以上就是本文的全部?jī)?nèi)容，希望對(duì)大家的學(xué)習(xí)有所幫助，也希望大家多多支持服務(wù)器之家。

原文鏈接：https://blog.phpsoho.com/2018/02/08/spring-boot-入門之消息中間件

延伸 · 閱讀

精彩推薦
主站蜘蛛池模板： 亚洲第一视频网站 | 亚洲在线电影 | aaa在线免费观看 | 岛国搬运工av在线播放 | 午夜剧场免费在线观看 | 精品久久一区二区三区 | 国产视频1区2区 | 欧美三区 | 日本99精品 | 97国产资源| 午夜精品福利电影 | 欧美九九九 | 亚洲久久 | 99热在线播放 | 日韩中文字幕在线观看 | 91偷拍精品一区二区三区 | 日韩国产欧美视频 | 黄色国产在线视频 | 欧美一区免费 | 男人的天堂视频网站 | 中文字幕视频播放 | www.一区二区三区 | 自拍偷拍色 | 北条麻妃在线一区二区免费播放 | 成人影院www在线观看 | 日本久久精品 | 久久精品国产亚洲一区二区三区 | 久久国产精品一区二区 | 亚洲免费观看在线视频 | 91cn在线观看 | 欧美日韩一区二区在线 | 亚洲一本 | jav成人av免费播放 | 久久久高清 | 久久久久久久国产精品 | 国产91视频在线观看 | 亚洲三级在线免费观看 | 伊人激情综合 | 国产综合精品一区二区三区 | 久草成人 | 免费污片网站 |