1. 概述
老話說的好：做人要懂得變通，善于思考，有時稍微轉個彎，也許問題就解決了。
言歸正傳，之前我們聊了 RabbitMQ 3.9.7 鏡像模式集群的搭建，今天我們來聊聊 RabbitMQ 3.9.7 鏡像模式集群與Springboot 2.5.5 整合。
2. 場景說明
服務器A IP：192.168.1.22
服務器B IP：192.168.1.8
服務器C IP：192.168.1.144
此三臺服務器上已搭建好了 RabbitMQ鏡像模式集群，鏡像模式集群的搭建，可參見我的上一篇文章。
3. 與Springboot的整合
3.1 引入依賴
|
1
2
3
4
5
6
|
<parent> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-parent</artifactId> <version>2.5.5</version> <relativePath/> <!-- lookup parent from repository --> </parent> |
|
1
2
3
4
|
<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId> </dependency> |
3.2 生產服務配置
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
|
spring: rabbitmq: addresses: 192.168.1.22:5672,192.168.1.8:5672,192.168.1.144:5672 username: guest password: guest virtual-host: / connection-timeout: 16000 # 啟用消息確認模式 publisher-confirm-type: correlated # 啟用 return 消息模式 publisher-returns: true template: mandatory: true |
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
|
import org.springframework.amqp.AmqpException;import org.springframework.amqp.core.MessagePostProcessor;import org.springframework.amqp.rabbit.connection.CorrelationData;import org.springframework.amqp.rabbit.core.RabbitTemplate;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.messaging.Message;import org.springframework.messaging.MessageHeaders;import org.springframework.messaging.support.MessageBuilder;import org.springframework.stereotype.Component;import java.util.Map;@Componentpublic class Producer { @Autowired private RabbitTemplate rabbitTemplate; /** * 確認回調 */ final RabbitTemplate.ConfirmCallback confirmCallback = new RabbitTemplate.ConfirmCallback() { @Override public void confirm(CorrelationData correlationData, boolean ack, String cause) { // correlationData 唯一標識 // ack mq是否收到消息 // cause 失敗原因 System.out.println("correlationData：" + correlationData.getId()); System.out.println("ack：" + ack); System.out.println("cause：" + cause); } }; /** * 發送消息 * @param messageBody 消息體 * @param headers 附加屬性 * @throws Exception */ public void sendMessage(String messageBody, Map<String, Object> headers, String id) throws Exception { MessageHeaders messageHeaders = new MessageHeaders(headers); Message<String> message = MessageBuilder.createMessage(messageBody, messageHeaders); rabbitTemplate.setConfirmCallback(confirmCallback); String exchangeName = "exchange-hello"; String routingKey = "test.123"; CorrelationData correlationData = new CorrelationData(id); rabbitTemplate.convertAndSend(exchangeName, routingKey, message, new MessagePostProcessor() { /** * 發送消息后做的事情 * @param message * @return * @throws AmqpException */ @Override public org.springframework.amqp.core.Message postProcessMessage(org.springframework.amqp.core.Message message) throws AmqpException { return message; } }, correlationData); }} |
3.4 消費服務配置
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
|
spring: rabbitmq: addresses: 192.168.1.22:5672,192.168.1.8:5672,192.168.1.144:5672 username: guest password: guest virtual-host: / connection-timeout: 16000 listener: simple: # 設置為手工ACK acknowledge-mode: manual concurrency: 5 prefetch: 1 max-concurrency: 10 |
3.5 消費服務代碼
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
|
import com.rabbitmq.client.Channel;import org.springframework.amqp.rabbit.annotation.*;import org.springframework.amqp.support.AmqpHeaders;import org.springframework.messaging.Message;import org.springframework.stereotype.Component;@Componentpublic class Consumer { @RabbitListener(bindings = @QueueBinding( value = @Queue(value = "queue-hello", durable = "true"), exchange = @Exchange(value = "exchange-hello" , durable = "true", type = "topic"), key = "test.*" )) @RabbitHandler public void onMessage(Message message, Channel channel) throws Exception { System.out.println("收到消息：" + message.getPayload()); Long deliveryTag = (Long)message.getHeaders().get(AmqpHeaders.DELIVERY_TAG); channel.basicAck(deliveryTag, false); }} |
3.6 Rest 測試代碼
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
|
@RestController@RequestMapping("/mq")public class RabbitmqController { @Autowired private Producer producer; @GetMapping("/sendMessage") public String sendMessage(@RequestParam String messageBody, @RequestParam String id) throws Exception { Map<String, Object> headers = new HashMap<>(); producer.sendMessage(messageBody, headers, id); return "success"; }} |
4. 綜述
到此這篇關于RabbitMQ 3.9.7 鏡像模式集群與Springboot 2.5.5 整合的文章就介紹到這了,更多相關RabbitMQ鏡像模式集群內容請搜索服務器之家以前的文章或繼續瀏覽下面的相關文章希望大家以后多多支持服務器之家！
原文鏈接：https://www.cnblogs.com/w84422/p/15361662.html
