目錄一、環(huán)境準備二、簡單模式三、工作隊列模式四、廣播模式（Fanout）五、直連模式（Direct）六、通配符模式（Topic）一、環(huán)境準備

1、pom依賴

<!-- 父工程依賴 --> <parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>2.3.6.RELEASE</version> </parent> <dependencies><dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-web</artifactId></dependency><dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId></dependency><dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-test</artifactId></dependency><dependency> <groupId>io.springfox</groupId> <artifactId>springfox-swagger2</artifactId> <version>2.6.0</version></dependency><dependency> <groupId>io.springfox</groupId> <artifactId>springfox-swagger-ui</artifactId> <version>2.6.0</version></dependency> </dependencies>

2、配置文件

server: port: 8080spring: rabbitmq: host: 192.168.131.171 port: 5672 username: jihu password: jihu virtual-host: /jihu

3、啟動類

@SpringBootApplicationpublic class RabbitMQApplication { public static void main(String[] args) { SpringApplication.run(RabbitMQApplication.class); }}

5、Swagger2類

@Configuration@EnableSwagger2public class Swagger2 { // http://127.0.0.1:8080/swagger-ui.html @Bean public Docket createRestApi() {return new Docket(DocumentationType.SWAGGER_2).apiInfo(apiInfo()).select().apis(RequestHandlerSelectors.basePackage('com.jihu')).paths(PathSelectors.any()).build(); } private ApiInfo apiInfo() {return new ApiInfoBuilder().title('極狐-Spring Boot中使用spring-boot-starter-amqp集成rabbitmq').description('測試SpringBoot整合進行各種工作模式信息的發(fā)送')/*.termsOfServiceUrl('https://www.jianshu.com/p/c79f6a14f6c9')*/.contact('roykingw').version('1.0').build(); }}

6、ProducerController

@RestControllerpublic class ProducerController { @Autowired private RabbitTemplate rabbitTemplate; //helloWorld 直連模式 @ApiOperation(value = 'helloWorld發(fā)送接口', notes = '直接發(fā)送到隊列') @GetMapping(value = '/helloWorldSend') public Object helloWorldSend(String message) throws AmqpException, UnsupportedEncodingException {//設置部分請求參數(shù)MessageProperties messageProperties = new MessageProperties();messageProperties.setContentType(MessageProperties.CONTENT_TYPE_TEXT_PLAIN);//發(fā)消息rabbitTemplate.send('helloWorldqueue', new Message(message.getBytes('UTF-8'), messageProperties));return 'message sended : ' + message; } //工作隊列模式 @ApiOperation(value = 'workqueue發(fā)送接口', notes = '發(fā)送到所有監(jiān)聽該隊列的消費') @GetMapping(value = '/workqueueSend') public Object workqueueSend(String message) throws AmqpException, UnsupportedEncodingException {MessageProperties messageProperties = new MessageProperties();messageProperties.setContentType(MessageProperties.CONTENT_TYPE_TEXT_PLAIN);//制造多個消息進行發(fā)送操作for (int i = 0; i < 10; i++) { rabbitTemplate.send('work_sb_mq_q', new Message(message.getBytes('UTF-8'), messageProperties));}return 'message sended : ' + message; } // pub/sub 發(fā)布訂閱模式 交換機類型 fanout @ApiOperation(value = 'fanout發(fā)送接口', notes = '發(fā)送到fanoutExchange。消息將往該exchange下的所有queue轉(zhuǎn)發(fā)') @GetMapping(value = '/fanoutSend') public Object fanoutSend(String message) throws AmqpException, UnsupportedEncodingException {MessageProperties messageProperties = new MessageProperties();messageProperties.setContentType(MessageProperties.CONTENT_TYPE_TEXT_PLAIN);//fanout模式只往exchange里發(fā)送消息。分發(fā)到exchange下的所有queuerabbitTemplate.send('fanoutExchange', '', new Message(message.getBytes('UTF-8'), messageProperties));return 'message sended : ' + message; } //routing路由工作模式 交換機類型 direct @ApiOperation(value = 'direct發(fā)送接口', notes = '發(fā)送到directExchange。exchange轉(zhuǎn)發(fā)消息時，會往routingKey匹配的queue發(fā)送') @GetMapping(value = '/directSend') public Object routingSend(String routingKey, String message) throws AmqpException, UnsupportedEncodingException {if (null == routingKey) { routingKey = 'china.changsha';}MessageProperties messageProperties = new MessageProperties();messageProperties.setContentType(MessageProperties.CONTENT_TYPE_TEXT_PLAIN);//fanout模式只往exchange里發(fā)送消息。分發(fā)到exchange下的所有queuerabbitTemplate.send('directExchange', routingKey, new Message(message.getBytes('UTF-8'), messageProperties));return 'message sended : routingKey >' + routingKey + ';message > ' + message; } //topic 工作模式 交換機類型 topic @ApiOperation(value = 'topic發(fā)送接口', notes = '發(fā)送到topicExchange。exchange轉(zhuǎn)發(fā)消息時，會往routingKey匹配的queue發(fā)送，*代表一個單詞，#代表0個或多個單詞。') @GetMapping(value = '/topicSend') public Object topicSend(String routingKey, String message) throws AmqpException, UnsupportedEncodingException {if (null == routingKey) { routingKey = 'changsha.kf';}MessageProperties messageProperties = new MessageProperties();messageProperties.setContentType(MessageProperties.CONTENT_TYPE_TEXT_PLAIN);//fanout模式只往exchange里發(fā)送消息。分發(fā)到exchange下的所有queuerabbitTemplate.send('topicExchange', routingKey, new Message(message.getBytes('UTF-8'), messageProperties));return 'message sended : routingKey >' + routingKey + ';message > ' + message; }}

7、ConcumerReceiver

@Componentpublic class ConcumerReceiver { //直連模式的多個消費者，會分到其中一個消費者進行消費。類似task模式 //通過注入RabbitContainerFactory對象，來設置一些屬性，相當于task里的channel.basicQos @RabbitListener(queues = 'helloWorldqueue') public void helloWorldReceive(String message) {System.out.println('helloWorld模式 received message : ' + message); } //工作隊列模式 @RabbitListener(queues = 'work_sb_mq_q') public void wordQueueReceiveq1(String message) {System.out.println('工作隊列模式1 received message : ' + message); } @RabbitListener(queues = 'work_sb_mq_q') public void wordQueueReceiveq2(String message) {System.out.println('工作隊列模式2 received message : ' + message); } //pub/sub模式進行消息監(jiān)聽 @RabbitListener(queues = 'fanout.q1') public void fanoutReceiveq1(String message) {System.out.println('發(fā)布訂閱模式1received message : ' + message); } @RabbitListener(queues = 'fanout.q2') public void fanoutReceiveq2(String message) {System.out.println('發(fā)布訂閱模式2 received message : ' + message); } //Routing路由模式 @RabbitListener(queues = 'direct_sb_mq_q1') public void routingReceiveq1(String message) {System.out.println('Routing路由模式routingReceiveq11111 received message : ' + message); } @RabbitListener(queues = 'direct_sb_mq_q2') public void routingReceiveq2(String message) {System.out.println('Routing路由模式routingReceiveq22222 received message : ' + message); } //topic 模式 //注意這個模式會有優(yōu)先匹配原則。例如發(fā)送routingKey=hunan.IT,那匹配到hunan.*(hunan.IT,hunan.eco),之后就不會再去匹配*.ITd @RabbitListener(queues = 'topic_sb_mq_q1') public void topicReceiveq1(String message) {System.out.println('Topic模式 topic_sb_mq_q1 received message : ' + message); } @RabbitListener(queues = 'topic_sb_mq_q2') public void topicReceiveq2(String message) {System.out.println('Topic模式 topic_sb_mq_q2 received message : ' + message); }}二、簡單模式

隊列配置：

/** * HelloWorld rabbitmq第一個工作模式 * 直連模式只需要聲明隊列，所有消息都通過隊列轉(zhuǎn)發(fā)。 * 無需設置交換機 */@Configurationpublic class HelloWorldConfig {@Beanpublic Queue setQueue() {return new Queue('helloWorldqueue');}}三、工作隊列模式

@Configurationpublic class WorkConfig { //聲明隊列 @Bean public Queue workQ1() {return new Queue('work_sb_mq_q'); }}四、廣播模式（Fanout）

/** * Fanout模式需要聲明exchange，并綁定queue，由exchange負責轉(zhuǎn)發(fā)到queue上。 * 廣播模式 交換機類型設置為：fanout */@Configurationpublic class FanoutConfig {//聲明隊列@Beanpublic Queue fanoutQ1() {return new Queue('fanout.q1');}@Beanpublic Queue fanoutQ2() {return new Queue('fanout.q2');}//聲明exchange@Beanpublic FanoutExchange setFanoutExchange() {return new FanoutExchange('fanoutExchange');}//聲明Binding,exchange與queue的綁定關系@Beanpublic Binding bindQ1() {return BindingBuilder.bind(fanoutQ1()).to(setFanoutExchange());}@Beanpublic Binding bindQ2() {return BindingBuilder.bind(fanoutQ2()).to(setFanoutExchange());}}五、直連模式（Direct）

/* 路由模式|Routing模式 交換機類型：direct*/@Configurationpublic class DirectConfig {//聲明隊列@Beanpublic Queue directQ1() {return new Queue('direct_sb_mq_q1');}@Beanpublic Queue directQ2() {return new Queue('direct_sb_mq_q2');}//聲明exchange@Beanpublic DirectExchange setDirectExchange() {return new DirectExchange('directExchange');}//聲明binding，需要聲明一個routingKey@Beanpublic Binding bindDirectBind1() {return BindingBuilder.bind(directQ1()).to(setDirectExchange()).with('china.changsha');}@Beanpublic Binding bindDirectBind2() {return BindingBuilder.bind(directQ2()).to(setDirectExchange()).with('china.beijing');}}六、通配符模式（Topic）

/*Topics模式 交換機類型 topic* */@Configurationpublic class TopicConfig {//聲明隊列@Beanpublic Queue topicQ1() {return new Queue('topic_sb_mq_q1');}@Beanpublic Queue topicQ2() {return new Queue('topic_sb_mq_q2');}//聲明exchange@Beanpublic TopicExchange setTopicExchange() {return new TopicExchange('topicExchange');}//聲明binding，需要聲明一個roytingKey@Beanpublic Binding bindTopicHebei1() {return BindingBuilder.bind(topicQ1()).to(setTopicExchange()).with('changsha.*');}@Beanpublic Binding bindTopicHebei2() {return BindingBuilder.bind(topicQ2()).to(setTopicExchange()).with('#.beijing');}}

測試

我們啟動上面的SpringBoot項目。

然后我們訪問swagger地址：http://127.0.0.1:8080/swagger-ui.html

然后我們就可以使用swagger測試接口了。

或者可以使用postman進行測試。

到此這篇關于SpringBoot整合RabbitMQ的5種模式實戰(zhàn)的文章就介紹到這了,更多相關SpringBoot整合RabbitMQ模式內(nèi)容請搜索好吧啦網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持好吧啦網(wǎng)！