1.1 導(dǎo)入依賴

<dependencies><!--RabbitMQ 依賴--><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency><dependency><groupId>com.alibaba</groupId><artifactId>fastjson</artifactId><version>1.2.47</version></dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId></dependency><!--swagger--><dependency><groupId>io.springfox</groupId><artifactId>springfox-swagger2</artifactId><version>2.9.2</version></dependency><dependency><groupId>io.springfox</groupId><artifactId>springfox-swagger-ui</artifactId><version>2.9.2</version></dependency><!--RabbitMQ 測(cè)試依賴--><dependency><groupId>org.springframework.amqp</groupId><artifactId>spring-rabbit-test</artifactId><scope>test</scope></dependency></dependencies>

1.2 修改配置文件

spring.rabbitmq.host=xxx.xxx.xxx.xxx spring.rabbitmq.port=5672 spring.rabbitmq.username=admin spring.rabbitmq.password=123

1.3 隊(duì)列ttl

1.3.1 隊(duì)列架構(gòu)圖

創(chuàng)建兩個(gè)隊(duì)列 QA 和 QB，兩者隊(duì)列 TTL 分別設(shè)置為 10S 和 40S，然后在創(chuàng)建一個(gè)交換機(jī) X 和死信交 換機(jī) Y，它們的類型都是 direct，創(chuàng)建一個(gè)死信隊(duì)列 QD，它們的綁定關(guān)系如下：

1.3.2 隊(duì)列配置類

@Configuration public class TtlQueueConfig {private static final String X_EXCHANGE = "X";private static final String QUEUE_A = "QA";private static final String QUEUE_B = "QB";private static final String Y_DEAD_LETTER_EXCHANGE = "Y";private static final String DEAD_LETTER_QUEUE = "QD"; ?// 聲明 xExchange@Bean("xExchange")public DirectExchange xExchange() {return new DirectExchange(X_EXCHANGE);} ?// 聲明 xExchange@Bean("yExchange")public DirectExchange yExchange() {return new DirectExchange(Y_DEAD_LETTER_EXCHANGE);} ?//聲明隊(duì)列 A ttl 為 10s 并綁定到對(duì)應(yīng)的死信交換機(jī)@Bean("queueA")public Queue queueA() {Map<String, Object> args = new HashMap<>(3);//聲明當(dāng)前隊(duì)列綁定的死信交換機(jī)args.put("x-dead-letter-exchange", Y_DEAD_LETTER_EXCHANGE);//聲明當(dāng)前隊(duì)列的死信路由 keyargs.put("x-dead-letter-routing-key", "YD");//聲明隊(duì)列的 TTLargs.put("x-message-ttl", 10000);return QueueBuilder.durable(QUEUE_A).withArguments(args).build();} ?// 聲明隊(duì)列 A 綁定 X 交換機(jī)@Beanpublic Binding queueaBindingX(@Qualifier("queueA") Queue queueA,@Qualifier("xExchange") DirectExchange xExchange) {return BindingBuilder.bind(queueA).to(xExchange).with("XA");} ?//聲明隊(duì)列 B ttl 為 40s 并綁定到對(duì)應(yīng)的死信交換機(jī)@Bean("queueB")public Queue queueB() {Map<String, Object> args = new HashMap<>(3);//聲明當(dāng)前隊(duì)列綁定的死信交換機(jī)args.put("x-dead-letter-exchange", Y_DEAD_LETTER_EXCHANGE);//聲明當(dāng)前隊(duì)列的死信路由 keyargs.put("x-dead-letter-routing-key", "YD");//聲明隊(duì)列的 TTLargs.put("x-message-ttl", 40000);return QueueBuilder.durable(QUEUE_B).withArguments(args).build();} ?//聲明隊(duì)列 B 綁定 X 交換機(jī)@Beanpublic Binding queuebBindingX(@Qualifier("queueB") Queue queue1B,@Qualifier("xExchange") DirectExchange xExchange) {return BindingBuilder.bind(queue1B).to(xExchange).with("XB");} ?//聲明死信隊(duì)列 QD@Bean("queueD")public Queue queueD() {return new Queue(DEAD_LETTER_QUEUE);} ?//聲明死信隊(duì)列 QD 綁定關(guān)系@Beanpublic Binding deadLetterBindingQAD(@Qualifier("queueD") Queue queueD,@Qualifier("yExchange") DirectExchange yExchange) {return BindingBuilder.bind(queueD).to(yExchange).with("YD");} }

1.3.3 生產(chǎn)者

@Slf4j @RequestMapping("ttl") @RestController public class SendMsgController {@Autowiredprivate RabbitTemplate rabbitTemplate; ?@GetMapping("sendMsg/{message}")public void sendMsg(@PathVariable String message) {log.info("當(dāng)前時(shí)間：{},發(fā)送一條信息給兩個(gè) TTL 隊(duì)列:{}", new Date(), message);rabbitTemplate.convertAndSend("X", "XA", "消息來(lái)自 ttl 為 10S 的隊(duì)列: " + message);rabbitTemplate.convertAndSend("X", "XB", "消息來(lái)自 ttl 為 40S 的隊(duì)列: " + message);} }

1.3.4 消費(fèi)者

@Slf4j @Component public class DeadLetterQueueConsumer {@RabbitListener(queues = "QD")public void receiveD(Message message, Channel channel) throws IOException {String msg = new String(message.getBody());log.info("當(dāng)前時(shí)間：{},收到死信隊(duì)列信息{}", new Date().toString(), msg);} }

1.3.5 消息消費(fèi)

發(fā)起一個(gè)請(qǐng)求 http://localhost:8080/ttl/sendMsg/hello

第一條消息在 10S 后變成了死信消息，然后被消費(fèi)者消費(fèi)掉，第二條消息在 40S 之后變成了死信消息，然后被消費(fèi)掉，這樣一個(gè)延時(shí)隊(duì)列就打造完成了。

1.3.6 消費(fèi)者延遲時(shí)間設(shè)置問(wèn)題

? ? /*** 可以發(fā)送指定過(guò)期時(shí)間的消息* 不同過(guò)期時(shí)間指定不同的隊(duì)列會(huì)造成隊(duì)列過(guò)多* 我們可以不指定隊(duì)列的過(guò)期時(shí)間，而是在生產(chǎn)者這邊指定消息的過(guò)期時(shí)間** @param message* @param ttlTime*/@GetMapping("sendExpirationMsg/{message}/{ttlTime}")public void sendMsg(@PathVariable String message, @PathVariable String ttlTime) {rabbitTemplate.convertAndSend("X", "XC", message, correlationData -> {correlationData.getMessageProperties().setExpiration(ttlTime);return correlationData;});log.info("當(dāng)前時(shí)間：{},發(fā)送一條時(shí)長(zhǎng){}毫秒 TTL 信息給隊(duì)列 C:{}", new Date(), ttlTime, message);}

RabbitMQ 只會(huì)檢查第一個(gè)消息是否過(guò)期，如果過(guò)期則丟到死信隊(duì)列， 如果第一個(gè)消息的延時(shí)時(shí)長(zhǎng)很長(zhǎng)，而第二個(gè)消息的延時(shí)時(shí)長(zhǎng)很短，第二個(gè)消息并不會(huì)優(yōu)先得到執(zhí)行。

1.3.7 延時(shí)隊(duì)列總結(jié)

延時(shí)隊(duì)列在需要延時(shí)處理的場(chǎng)景下非常有用，使用 RabbitMQ 來(lái)實(shí)現(xiàn)延時(shí)隊(duì)列可以很好的利用 RabbitMQ 的特性，如：消息可靠發(fā)送、消息可靠投遞、死信隊(duì)列來(lái)保障消息至少被消費(fèi)一次以及未被正確處理的消息不會(huì)被丟棄。另外，通過(guò) RabbitMQ 集群的特性，可以很好的解決單點(diǎn)故障問(wèn)題，不會(huì)因?yàn)閱蝹€(gè)節(jié)點(diǎn)掛掉導(dǎo)致延時(shí)隊(duì)列不可用或者消息丟失。

當(dāng)然，延時(shí)隊(duì)列還有很多其它選擇，比如利用 Java 的 DelayQueue，利用 Redis 的 zset，利用 Quartz 或者利用 kafka 的時(shí)間輪，這些方式各有特點(diǎn),看需要適用的場(chǎng)景。

總結(jié)

以上是生活随笔為你收集整理的RabbitMQ（六）整合SpringBoot的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。