Springboot 配置RabbitMQ文档的方法步骤

 更新时间:2019年03月18日 14:08:27   作者:houshiqun689  
这篇文章主要介绍了Springboot 配置RabbitMQ文档的方法步骤,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧

简介

RabbitMQ是实现AMQP(高级消息队列协议)的消息中间件的一种,用于在分布式系统中存储转发消息,在易用性、扩展性、高可用性等方面表现不俗

概念:

  • 生产者 消息的产生方,负责将消息推送到消息队列
  • 消费者 消息的最终接受方,负责监听队列中的对应消息,消费消息
  • 队列 消息的寄存器,负责存放生产者发送的消息
  • 交换机 负责根据一定规则分发生产者产生的消息
  • 绑定 完成交换机和队列之间的绑定

模式:

  • direct:直连模式,用于实例间的任务分发
  • topic:话题模式,通过可配置的规则分发给绑定在该exchange上的队列
  • headers:适用规则复杂的分发,用headers里的参数表达规则
  • fanout:分发给所有绑定到该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() {
  // 参数1:队列
  // 参数2:是否持久化
  // 参数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);
 }
}

三、编写生产者

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) {
    
}

以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持脚本之家。

相关文章

  • 一文带你彻底剖析Java中Synchronized原理

    一文带你彻底剖析Java中Synchronized原理

    Synchronized是Java中的隐式锁,它的获取锁和释放锁都是隐式的,完全交由JVM帮助我们操作,在了解Synchronized关键字之前,首先要学习的知识点就是Java的对象结构,本文介绍的非常详细,需要的朋友可以参考下
    2023-05-05
  • Java实现扫雷游戏的代码分享

    Java实现扫雷游戏的代码分享

    windows自带的游戏《扫雷》是陪伴了无数人的经典游戏,本文将利用Java语言实现这一经典的游戏,文中的示例代码讲解详细,感兴趣的可以学习一下
    2022-05-05
  • SpringBoot中实现登录拦截器的代码实例

    SpringBoot中实现登录拦截器的代码实例

    这篇文章主要介绍了SpringBoot中实现登录拦截器的代码实例,对于管理系统或其他需要用户登录的系统,登录验证都是必不可少的环节,在SpringBoot开发的项目中,通过实现拦截器来实现用户登录拦截并验证,需要的朋友可以参考下
    2023-10-10
  • java通过AES生成公钥加密数据ECC加密公钥

    java通过AES生成公钥加密数据ECC加密公钥

    这篇文章主要为大家介绍了java通过AES生成公钥加密数据ECC加密公钥实现案例详解,有需要的朋友可以借鉴参考下,希望能够有所帮助,祝大家多多进步,早日升职加薪
    2022-12-12
  • Spring框架实现AOP添加日志记录功能过程详解

    Spring框架实现AOP添加日志记录功能过程详解

    这篇文章主要介绍了Spring框架实现AOP添加日志记录功能过程详解,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下
    2019-12-12
  • 详解三种java实现多线程的方式

    详解三种java实现多线程的方式

    数据时代的到来,多线程一直都是比较关心的问题之一,这篇文章介绍了JAVA实现多线程的三种方法,有需要的朋友可以参考一下
    2015-08-08
  • JVM的GC日志及运行参数解读

    JVM的GC日志及运行参数解读

    这篇文章主要为大家介绍了JVM的GC日志及运行参数解读,有需要的朋友可以借鉴参考下,希望能够有所帮助,祝大家多多进步,早日升职加薪
    2023-09-09
  • springcloud结合bytetcc实现数据强一致性原理解析

    springcloud结合bytetcc实现数据强一致性原理解析

    这篇文章主要介绍了springcloud结合bytetcc实现数据强一致性原理解析,本文给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要的朋友可以参考下
    2021-03-03
  • JAVA 统计字符串中中文,英文,数字,空格,特殊字符的个数

    JAVA 统计字符串中中文,英文,数字,空格,特殊字符的个数

    这篇文章主要介绍了JAVA 统计字符串中中文,英文,数字,空格,特殊字符的个数 ,本文通过一段代码给大家介绍的非常详细,具有一定的参考借鉴价值,需要的朋友可以参考下
    2019-06-06
  • Java内存模型知识汇总

    Java内存模型知识汇总

    本文中,有很多定义和说法,都是笔者自己理解后定义出来的。希望能够让读者可以对Java内存模型有更加清晰的认识。当然,如有偏颇,欢迎指正。
    2018-09-09

最新评论