1.application.yml
|
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
|
server: port: 8184spring: application: name: rabbitmq-demo rabbitmq: host: 127.0.0.1 # ip地址 port: 5672 username: admin # 連接賬號 password: 123456 # 連接密碼 template: retry: enabled: true # 開啟失敗重試 initial-interval: 10000ms # 第一次重試的間隔時長 max-interval: 300000ms # 最長重試間隔，超過這個間隔將不再重試 multiplier: 2 # 下次重試間隔的倍數(shù)，此處是2即下次重試間隔是上次的2倍 exchange: topic.exchange # 缺省的交換機名稱，此處配置后，發(fā)送消息如果不指定交換機就會使用這個 publisher-confirm-type: correlated # 生產(chǎn)者確認(rèn)機制，確保消息會正確發(fā)送，如果發(fā)送失敗會有錯誤回執(zhí)，從而觸發(fā)重試 publisher-returns: true listener: type: simple simple: acknowledge-mode: manual prefetch: 1 # 限制每次發(fā)送一條數(shù)據(jù)。 concurrency: 3 # 同一個隊列啟動幾個消費者 max-concurrency: 3 # 啟動消費者最大數(shù)量 # 重試策略相關(guān)配置 retry: enabled: true # 是否支持重試 max-attempts: 5 stateless: false multiplier: 1.0 # 時間策略乘數(shù)因子 initial-interval: 1000ms max-interval: 10000ms default-requeue-rejected: true |
2.pom.xml引入依賴
|
1
2
3
4
5
|
<!-- rabbitmq --> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId> </dependency> |
3.常量類創(chuàng)建
|
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
|
/** * @author kkp * @ClassName RabbitMqConstants * @date 2021/11/3 14:16 * @Description */public class RabbitMqConstants { public final static String TEST1_QUEUE = "test1-queue"; public final static String TEST2_QUEUE = "test2-queue"; public final static String EXCHANGE_NAME = "test.topic.exchange"; /** * routingKey1 */ public final static String TOPIC_TEST1_ROUTINGKEY = "topic.test1.*"; public final static String TOPIC_TEST1_ROUTINGKEY_TEST = "topic.test1.test"; /** * routingKey1 */ public final static String TOPIC_TEST2_ROUTINGKEY = "topic.test2.*"; public final static String TOPIC_TEST2_ROUTINGKEY_TEST = "topic.test2.test";} |
4.配置Configuration
|
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
|
import com.example.demo.common.RabbitMqConstants;import lombok.extern.slf4j.Slf4j;import org.springframework.amqp.core.*;import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;import org.springframework.amqp.rabbit.core.RabbitTemplate;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.beans.factory.annotation.Qualifier;import org.springframework.beans.factory.config.ConfigurableBeanFactory;import org.springframework.context.annotation.Bean;import org.springframework.context.annotation.Configuration;import org.springframework.context.annotation.Scope;/** * @author kkp * @ClassName RabbitMqConfig * @date 2021/11/3 14:16 * @Description */@Slf4j@Configurationpublic class RabbitMqConfig { @Autowired private CachingConnectionFactory connectionFactory; /** * 聲明交換機 */ @Bean(RabbitMqConstants.EXCHANGE_NAME) public Exchange exchange(){ //durable(true) 持久化，mq重啟之后交換機還在 // Topic模式 //return ExchangeBuilder.topicExchange(RabbitMqConstants.EXCHANGE_NAME).durable(true).build(); //發(fā)布訂閱模式 return ExchangeBuilder.fanoutExchange(RabbitMqConstants.EXCHANGE_NAME).durable(true).build(); } /** * 聲明隊列 * new Queue(QUEUE_EMAIL,true,false,false) * durable="true" 持久化 rabbitmq重啟的時候不需要創(chuàng)建新的隊列 * auto-delete 表示消息隊列沒有在使用時將被自動刪除 默認(rèn)是false * exclusive 表示該消息隊列是否只在當(dāng)前connection生效,默認(rèn)是false */ @Bean(RabbitMqConstants.TEST1_QUEUE) public Queue esQueue() { return new Queue(RabbitMqConstants.TEST1_QUEUE); } /** * 聲明隊列 */ @Bean(RabbitMqConstants.TEST2_QUEUE) public Queue gitalkQueue() { return new Queue(RabbitMqConstants.TEST2_QUEUE); } /** * TEST1_QUEUE隊列綁定交換機，指定routingKey */ @Bean public Binding bindingEs(@Qualifier(RabbitMqConstants.TEST1_QUEUE) Queue queue, @Qualifier(RabbitMqConstants.EXCHANGE_NAME) Exchange exchange) { return BindingBuilder.bind(queue).to(exchange).with(RabbitMqConstants.TOPIC_TEST1_ROUTINGKEY).noargs(); } /** * TEST2_QUEUE隊列綁定交換機，指定routingKey */ @Bean public Binding bindingGitalk(@Qualifier(RabbitMqConstants.TEST2_QUEUE) Queue queue, @Qualifier(RabbitMqConstants.EXCHANGE_NAME) Exchange exchange) { return BindingBuilder.bind(queue).to(exchange).with(RabbitMqConstants.TOPIC_TEST2_ROUTINGKEY).noargs(); } /** * 如果需要在生產(chǎn)者需要消息發(fā)送后的回調(diào)， * 需要對rabbitTemplate設(shè)置ConfirmCallback對象， * 由于不同的生產(chǎn)者需要對應(yīng)不同的ConfirmCallback， * 如果rabbitTemplate設(shè)置為單例bean， * 則所有的rabbitTemplate實際的ConfirmCallback為最后一次申明的ConfirmCallback。 * @return */ @Bean @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE) public RabbitTemplate rabbitTemplate() { RabbitTemplate template = new RabbitTemplate(connectionFactory); return template; }} |
5.Rabbit工具類創(chuàng)建
|
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
|
import lombok.extern.slf4j.Slf4j;import org.springframework.amqp.core.Message;import org.springframework.amqp.rabbit.connection.CorrelationData;import org.springframework.amqp.rabbit.core.RabbitTemplate;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.stereotype.Component;import java.util.UUID;/** * @author kkp * @ClassName RabbitMqUtils * @date 2021/11/3 14:21 * @Description */@Slf4j@Componentpublic class RabbitMqUtils implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback{ private RabbitTemplate rabbitTemplate; /** * 構(gòu)造方法注入 */ @Autowired public RabbitMqUtils(RabbitTemplate rabbitTemplate) { this.rabbitTemplate = rabbitTemplate; //這是是設(shè)置回調(diào)能收到發(fā)送到響應(yīng) rabbitTemplate.setConfirmCallback(this); //如果設(shè)置備份隊列則不起作用 rabbitTemplate.setMandatory(true); rabbitTemplate.setReturnCallback(this); } /** * 回調(diào)確認(rèn) */ @Override public void confirm(CorrelationData correlationData, boolean ack, String cause) { if(ack){ log.info("消息發(fā)送成功：correlationData({}),ack({}),cause({})",correlationData,ack,cause); }else{ log.info("消息發(fā)送失敗：correlationData({}),ack({}),cause({})",correlationData,ack,cause); } } /** * 消息發(fā)送到轉(zhuǎn)換器的時候沒有對列,配置了備份對列該回調(diào)則不生效 * @param message * @param replyCode * @param replyText * @param exchange * @param routingKey */ @Override public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) { log.info("消息丟失：exchange({}),route({}),replyCode({}),replyText({}),message:{}",exchange,routingKey,replyCode,replyText,message); } /** * 發(fā)送到指定Queue * @param queueName * @param obj */ public void send(String queueName, Object obj){ CorrelationData correlationId = new CorrelationData(UUID.randomUUID().toString()); this.rabbitTemplate.convertAndSend(queueName, obj, correlationId); } /** * 1、交換機名稱 * 2、routingKey * 3、消息內(nèi)容 */ public void sendByRoutingKey(String exChange, String routingKey, Object obj){ CorrelationData correlationId = new CorrelationData(UUID.randomUUID().toString()); this.rabbitTemplate.convertAndSend(exChange, routingKey, obj, correlationId); }} |
6.service創(chuàng)建
|
1
2
3
4
5
6
|
public interface TestService { String sendTest1(String content); String sendTest2(String content);} |
7.impl實現(xià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
|
import com.example.demo.common.RabbitMqConstants;import com.example.demo.util.RabbitMqUtils;import lombok.extern.slf4j.Slf4j;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.stereotype.Service;/** * @author kkp * @ClassName TestServiceImpl * @date 2021/11/3 14:24 * @Description */@Service@Slf4jpublic class TestServiceImpl implements TestService { @Autowired private RabbitMqUtils rabbitMqUtils; @Override public String sendTest1(String content) { rabbitMqUtils.sendByRoutingKey(RabbitMqConstants.EXCHANGE_NAME, RabbitMqConstants.TOPIC_TEST1_ROUTINGKEY_TEST, content); log.info(RabbitMqConstants.TOPIC_TEST1_ROUTINGKEY_TEST+"***************發(fā)送成功*****************"); return "發(fā)送成功！"; } @Override public String sendTest2(String content) { rabbitMqUtils.sendByRoutingKey(RabbitMqConstants.EXCHANGE_NAME, RabbitMqConstants.TOPIC_TEST2_ROUTINGKEY_TEST, content); log.info(RabbitMqConstants.TOPIC_TEST2_ROUTINGKEY_TEST+"***************發(fā)送成功*****************"); return "發(fā)送成功！"; }} |
8.監(jiā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
39
40
41
42
43
44
45
46
47
48
|
import com.example.demo.common.RabbitMqConstants;import lombok.extern.slf4j.Slf4j;import org.springframework.amqp.core.ExchangeTypes;import org.springframework.amqp.core.Message;import org.springframework.amqp.rabbit.annotation.Exchange;import org.springframework.amqp.rabbit.annotation.Queue;import org.springframework.amqp.rabbit.annotation.QueueBinding;import org.springframework.amqp.rabbit.annotation.RabbitListener;import org.springframework.stereotype.Component;import com.rabbitmq.client.Channel;/** * @author kkp * @ClassName RabbitMqListener * @date 2021/11/3 14:22 * @Description */@Slf4j@Componentpublic class RabbitMqListener { @RabbitListener(queues = RabbitMqConstants.TEST1_QUEUE) public void test1Consumer(Message message, Channel channel) { try { //手動確認(rèn)消息已經(jīng)被消費 channel.basicAck(message.getMessageProperties().getDeliveryTag(), false); log.info("Counsoum1消費消息：" + message.toString() + "。成功！"); } catch (Exception e) { e.printStackTrace(); log.info("Counsoum1消費消息：" + message.toString() + "。失敗！"); } } @RabbitListener(queues = RabbitMqConstants.TEST2_QUEUE) public void test2Consumer(Message message, Channel channel) { try { //手動確認(rèn)消息已經(jīng)被消費 channel.basicAck(message.getMessageProperties().getDeliveryTag(), false); log.info("Counsoum2消費消息：" + message.toString() + "。成功！"); } catch (Exception e) { e.printStackTrace(); log.info("Counsoum2消費消息：" + message.toString() + "。失敗！"); } }} |
9.Controller測試
|
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
|
import com.example.demo.server.TestService;import jdk.nashorn.internal.objects.annotations.Getter;import lombok.extern.slf4j.Slf4j;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.web.bind.annotation.*;import java.util.Map;/** * @author kkp * @ClassName TestController * @date 2021/11/3 14:25 * @Description */@Slf4j@RestController@RequestMapping("/enterprise")public class TestController { @Autowired private TestService testService; @GetMapping("/finance") public String hello3(@RequestParam(required = false) Map<String, Object> params) { return testService.sendTest2(params.get("entId").toString()); } /** * 發(fā)送消息test2 * @param content * @return */ @PostMapping(value = "/finance2") public String sendTest2(@RequestBody String content) { return testService.sendTest2(content); }} |
到此這篇關(guān)于springboot2.5.6集成RabbitMq實現(xiàn)Topic主題模式的文章就介紹到這了,更多相關(guān)springboot集成RabbitMq內(nèi)容請搜索服務(wù)器之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持服務(wù)器之家！
原文鏈接：https://blog.csdn.net/weixin_44907173/article/details/121124048
