Springboot 配置RabbitMQ文檔的方法步驟

更新時間：2019年03月18日 14:08:27 作者：houshiqun689

這篇文章主要介紹了Springboot 配置RabbitMQ文檔的方法步驟,文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧

簡介

RabbitMQ是實現(xiàn)AMQP（高級消息隊列協(xié)議）的消息中間件的一種，用于在分布式系統(tǒng)中存儲轉(zhuǎn)發(fā)消息，在易用性、擴展性、高可用性等方面表現(xiàn)不俗

概念：

生產(chǎn)者消息的產(chǎn)生方，負責將消息推送到消息隊列
消費者消息的最終接受方，負責監(jiān)聽隊列中的對應消息，消費消息
隊列消息的寄存器，負責存放生產(chǎn)者發(fā)送的消息
交換機負責根據(jù)一定規(guī)則分發(fā)生產(chǎn)者產(chǎn)生的消息
綁定完成交換機和隊列之間的綁定

模式：

direct：直連模式，用于實例間的任務分發(fā)
topic：話題模式，通過可配置的規(guī)則分發(fā)給綁定在該exchange上的隊列
headers：適用規(guī)則復雜的分發(fā)，用headers里的參數(shù)表達規(guī)則
fanout：分發(fā)給所有綁定到該exchange上的隊列，忽略routing key

SpringBoot集成RabbitMQ

一、引入maven依賴

<dependency>
 <groupId>org.springframework.boot</groupId>
 <artifactId>spring-boot-starter-amqp</artifactId>
 <version>1.5.2.RELEASE</version>
</dependency>

二、配置application.properties

# rabbitmq
spring.rabbitmq.host = dev-mq.a.pa.com
spring.rabbitmq.port = 5672
spring.rabbitmq.username = admin
spring.rabbitmq.password = admin
spring.rabbitmq.virtualHost = /message-test/

三、編寫AmqpConfiguration配置文件

package message.test.configuration;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.amqp.RabbitProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class AmqpConfiguration {
/**
 * 消息編碼
 */
 public static final String MESSAGE_ENCODING = "UTF-8";
 public static final String EXCHANGE_ISSUE = "exchange_message_issue";
 public static final String QUEUE_ISSUE_USER = "queue_message_issue_user";
 public static final String QUEUE_ISSUE_ALL_USER = "queue_message_issue_all_user";
 public static final String QUEUE_ISSUE_ALL_DEVICE = "queue_message_issue_all_device";
 public static final String QUEUE_ISSUE_CITY = "queue_message_issue_city";
 public static final String ROUTING_KEY_ISSUE_USER = "routing_key_message_issue_user";
 public static final String ROUTING_KEY_ISSUE_ALL_USER = "routing_key_message_issue_all_user";
 public static final String ROUTING_KEY_ISSUE_ALL_DEVICE = "routing_key_message_issue_all_device";
 public static final String ROUTING_KEY_ISSUE_CITY = "routing_key_message_issue_city";
 public static final String EXCHANGE_PUSH = "exchange_message_push";
 public static final String QUEUE_PUSH_RESULT = "queue_message_push_result";

 @Autowired
 private RabbitProperties rabbitProperties;

 @Bean
 public Queue issueUserQueue() {
  return new Queue(QUEUE_ISSUE_USER);
 }

 @Bean
 public Queue issueAllUserQueue() {
  return new Queue(QUEUE_ISSUE_ALL_USER);
 }

 @Bean
 public Queue issueAllDeviceQueue() {
  return new Queue(QUEUE_ISSUE_ALL_DEVICE);
 }

 @Bean
 public Queue issueCityQueue() {
  return new Queue(QUEUE_ISSUE_CITY);
 }

 @Bean
 public Queue pushResultQueue() {
  return new Queue(QUEUE_PUSH_RESULT);
 }

 @Bean
 public DirectExchange issueExchange() {
  return new DirectExchange(EXCHANGE_ISSUE);
 }

 @Bean
 public DirectExchange pushExchange() {
  // 參數(shù)1：隊列
  // 參數(shù)2：是否持久化
  // 參數(shù)3：是否自動刪除
  return new DirectExchange(EXCHANGE_PUSH, true, true);
 }

 @Bean
 public Binding issueUserQueueBinding(@Qualifier("issueUserQueue") Queue queue,
    @Qualifier("issueExchange") DirectExchange exchange) {
   return BindingBuilder.bind(queue).to(exchange).with(ROUTING_KEY_ISSUE_USER);
 }

 @Bean
 public Binding issueAllUserQueueBinding(@Qualifier("issueAllUserQueue") Queue queue,
    @Qualifier("issueExchange") DirectExchange exchange) {
  return BindingBuilder.bind(queue).to(exchange).with(ROUTING_KEY_ISSUE_ALL_USER);
 }

 @Bean
 public Binding issueAllDeviceQueueBinding(@Qualifier("issueAllDeviceQueue") Queue queue,
    @Qualifier("issueExchange") DirectExchange exchange) {
  return BindingBuilder.bind(queue).to(exchange).with(ROUTING_KEY_ISSUE_ALL_DEVICE);
 }

 @Bean
 public Binding issueCityQueueBinding(@Qualifier("issueCityQueue") Queue queue,
    @Qualifier("issueExchange") DirectExchange exchange) {
  return BindingBuilder.bind(queue).to(exchange).with(ROUTING_KEY_ISSUE_CITY);
 }

 @Bean
 public Binding pushResultQueueBinding(@Qualifier("pushResultQueue") Queue queue,
    @Qualifier("pushExchange") DirectExchange exchange) {
  return BindingBuilder.bind(queue).to(exchange).withQueueName();
 }

 @Bean
 public ConnectionFactory defaultConnectionFactory() {
  CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
  connectionFactory.setHost(rabbitProperties.getHost());
  connectionFactory.setPort(rabbitProperties.getPort());
  connectionFactory.setUsername(rabbitProperties.getUsername());
  connectionFactory.setPassword(rabbitProperties.getPassword());
  connectionFactory.setVirtualHost(rabbitProperties.getVirtualHost());
  return connectionFactory;
 }

 @Bean
 public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(
    @Qualifier("defaultConnectionFactory") ConnectionFactory connectionFactory) {
  SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
  factory.setConnectionFactory(connectionFactory);
  factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
  return factory;
 }

 @Bean
 public AmqpTemplate rabbitTemplate(@Qualifier("defaultConnectionFactory") ConnectionFactory connectionFactory) 
 {
  return new RabbitTemplate(connectionFactory);
 }
}

三、編寫生產(chǎn)者

body = JSON.toJSONString(issueMessage).getBytes(AmqpConfiguration.MESSAGE_ENCODING);
 rabbitTemplate.convertAndSend(AmqpConfiguration.EXCHANGE_ISSUE,
            AmqpConfiguration.ROUTING_KEY_ISSUE_USER, body);

四、編寫消費者

@RabbitListener(queues = AmqpConfiguration.QUEUE_PUSH_RESULT)
public void handlePushResult(@Payload byte[] data, Channel channel,
    @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) {
    
}

以上就是本文的全部內(nèi)容，希望對大家的學習有所幫助，也希望大家多多支持腳本之家。

您可能感興趣的文章: