您好,登錄后才能下訂單哦!
這篇文章將為大家詳細(xì)講解有關(guān)Spring Boot如何集成RabbitMQ,小編覺得挺實(shí)用的,因此分享給大家做個(gè)參考,希望大家閱讀完這篇文章后可以有所收獲。
集成RabbitMQ
集成RabbitMQ只需要如下幾步即可
1、添加maven依賴
<!--rabbitmq--> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId> </dependency>
2、添加配置文件application.yaml
在application.yaml添加配置內(nèi)容如下
spring: rabbitmq: host: 192.168.1.161 port: 5672 username: guest password: guest cache: channel: size: 10 listener: type: simple simple: acknowledge-mode: auto concurrency: 5 default-requeue-rejected: true max-concurrency: 100 retry: enabled: true # initial-interval: 1000ms max-attempts: 3 # max-interval: 1000ms multiplier: 1 stateless: true # publisher-confirms: true</pre>
注意:
這里最基本的配置只需要配置host,port,username和password四個(gè)屬性即可
其他屬性都有各自的含義,比如retry是用于配置重試策略的,acknowledge-mode是配置消息接收確認(rèn)機(jī)制的。
3、編寫配置類
編寫RabbitConfig配置類,采用Java Configuration的方式配置RabbitTemplate、Exchange和Queue等信息,具體如下所示
package com.jackie.springbootdemo.config; import org.springframework.amqp.core.*; import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory; import org.springframework.amqp.rabbit.connection.ConnectionFactory; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter; import org.springframework.beans.factory.InitializingBean; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.beans.factory.annotation.Value; 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; import java.util.HashMap; import java.util.Map; @Configuration public class RabbitMQConfig implements InitializingBean { @Autowired SimpleRabbitListenerContainerFactory simpleRabbitListenerContainerFactory; @Override public void afterPropertiesSet() throws Exception { simpleRabbitListenerContainerFactory.setMessageConverter(new Jackson2JsonMessageConverter()); } @Bean("jackson2JsonMessageConverter") public Jackson2JsonMessageConverter jackson2JsonMessageConverter(ConnectionFactory connectionFactory) { return new Jackson2JsonMessageConverter(); } @Bean("rabbitTemplate") @Scope(ConfigurableBeanFactory.SCOPE_SINGLETON) public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory, @Qualifier("jackson2JsonMessageConverter") Jackson2JsonMessageConverter jackson2JsonMessageConverter) { RabbitTemplate template = new RabbitTemplate(connectionFactory); template.setMessageConverter(new Jackson2JsonMessageConverter()); return template; } // --------------------- 聲明隊(duì)列 ------------------------ @Bean public Queue demoQueue() { return new Queue("demo_queue"); } // --------------------- 聲明exchange ------------------------ @Bean public DirectExchange demoExchange() { return new DirectExchange("demo_exchange"); } // --------------------- 隊(duì)列綁定 ------------------------ @Bean public Binding bindingAlbumItemCreatedQueue(DirectExchange demoExchange, Queue demoQueue) { return BindingBuilder.bind(demoQueue).to(demoExchange).with("100"); } }
注意
這里聲明了Direct模式的Exchange,聲明一個(gè)Queue,并通過routing-key為100將demo_queue綁定到demo_exchange,這樣demo_queue就可以接收到demo_exchange發(fā)送的消息了。
4、編寫消息發(fā)送類
package com.jackie.springbootdemo.message; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.amqp.rabbit.support.CorrelationData; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; @Component public class Sender implements RabbitTemplate.ConfirmCallback { private RabbitTemplate rabbitTemplate; /** * 構(gòu)造方法注入 */ @Autowired public Sender(RabbitTemplate rabbitTemplate) { this.rabbitTemplate = rabbitTemplate; rabbitTemplate.setConfirmCallback(this); //rabbitTemplate如果為單例的話,那回調(diào)就是最后設(shè)置的內(nèi)容 } public void sendMsg(String content) { rabbitTemplate.convertAndSend("demo_exchange", "100", content); } /** * 回調(diào) */ @Override public void confirm(CorrelationData correlationData, boolean ack, String cause) { System.out.println(" 回調(diào)id:" + correlationData); if (ack) { System.out.println("消息成功消費(fèi)"); } else { System.out.println("消息消費(fèi)失敗:" + cause); } } }
注意
發(fā)送內(nèi)容content,路由到routing-key為100上,則我們就可以在demo_queue隊(duì)列中看到發(fā)送的消息內(nèi)容了
confirm函數(shù)是回調(diào)函數(shù),這里因?yàn)闆]有消費(fèi)者,且acknoledge-mode是auto(其他兩種值分別是none和manual),所以ack是false。
5、編寫發(fā)送消息測試類
package com.jackie.springbootdemo; import com.jackie.springbootdemo.message.Sender; import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.test.context.SpringBootTest; import org.springframework.test.context.junit4.SpringRunner; import org.springframework.test.context.web.WebAppConfiguration; @RunWith(SpringRunner.class) @SpringBootTest(classes = SpringbootDemoApplication.class) @WebAppConfiguration public class RabbitApplicationTests { @Autowired Sender sender; @Test public void contextLoads() throws Exception { sender.sendMsg("test"); } }
運(yùn)行該測試類,我們可以看到如下結(jié)果
6、編寫消息消費(fèi)類
package com.jackie.springbootdemo.message; import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.stereotype.Component; @Component public class Receiver { @RabbitListener(queues = "demo_queue") public void created(String message) { System.out.println("orignal message: " + message); } }
注意
消息消費(fèi)類也非常簡單,添加注解@RabbitListener,指定要監(jiān)聽的隊(duì)列名稱即可
除了注解@RabbitListener,我們經(jīng)常還能看到@RabbitHandler,這兩個(gè)注解可以配合起來使用。
@RabbitListener 標(biāo)注在類上面表示當(dāng)有收到消息的時(shí)候,就交給 @RabbitHandler 的方法處理,具體使用哪個(gè)方法處理,根據(jù) MessageConverter 轉(zhuǎn)換后的參數(shù)類型,形如
@RabbitListener(queues = "demo_queue") public class Receiver { @RabbitHandler public void processMessage1(String message) { System.out.println(message); } @RabbitHandler public void processMessage2(byte[] message) { System.out.println(new String(message)); } }
7、運(yùn)行消息發(fā)送測試類
從執(zhí)行結(jié)果可以看到,因?yàn)橛辛讼M(fèi)者,所以這次打印的結(jié)果是"消息消費(fèi)成功"
而且,我們看到Receiver類將消息消費(fèi)并打印出消息的內(nèi)容為"test"。
關(guān)于“Spring Boot如何集成RabbitMQ”這篇文章就分享到這里了,希望以上內(nèi)容可以對大家有一定的幫助,使各位可以學(xué)到更多知識,如果覺得文章不錯(cuò),請把它分享出去讓更多的人看到。
免責(zé)聲明:本站發(fā)布的內(nèi)容(圖片、視頻和文字)以原創(chuàng)、轉(zhuǎn)載和分享為主,文章觀點(diǎn)不代表本網(wǎng)站立場,如果涉及侵權(quán)請聯(lián)系站長郵箱:is@yisu.com進(jìn)行舉報(bào),并提供相關(guān)證據(jù),一經(jīng)查實(shí),將立刻刪除涉嫌侵權(quán)內(nèi)容。