投稿
  1. 极客之音首页
  2. 技术随笔

RabbitMQ 的安装和使用

小半 • • 技术随笔 • 阅读 53

世上唯一不能复制的是时间，唯一不能重演的是人生，唯一不劳而获的是年龄。该怎么走，过什么样的生活，全凭自己的选择和努力。人生很贵，请别浪费！与智者为伍，与良善者同行。RabbitMQ 的安装和使用，希望对大家有帮助，欢迎收藏，转发！站点地址：www.bmabk.com，来源：原文

引言

RabbitMQ 作为一个开源的消息中间件广泛使用。

  • 支持分布式部署。
  • 异步消息传递，支持多种消息协议、消息队列、送达确认、灵活的队列路由、多种交换类型。
  • 提供多种监听和管理工具，HTTP-API, 命令行工具command line tool, UI界面。

安装

容器安装

目前最新版本安装和启动命令如下：

docker run -it --rm --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:3.11-management

Linux或Windows安装

通过官网提供的安装包安装即可。具体安装方法可查看 https://www.rabbitmq.com/install-windows.html 。

  • 先安装Erlang。
  • 安装对应的 RabbitMQ 安装包。

RabbitMQ访问

通过UI界面访问对应的系统。

  • 登录地址 127.0.0.1:15762 。
  • 账号密码默认 guest\guest 。

Springboot集成RabbitMQ

  1. 修改依赖加入RabbitMQ启动项目，此处以maven为例子说明。

        <!-- rabbitmq -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>
  1. 增加 rabbitmq 配置。
spring:
  application:
    name: boot-rabbitmq
  # rabbitmq 配置
  rabbitmq:
    # Redis 服务器地址
    host: 127.0.0.1
    # 连接端口号
    port: 5672
    username: guest
    password: guest
  1. 增加 rabbit相关配置。

/**
 * 自定义注入配置bean相关组件
 */
@Configuration
@Slf4j
public class RabbitMQConfig {

    // 自动装配RabbitMQ的链接工厂实例
    @Autowired
    private CachingConnectionFactory connectionFactory;
    // 自动装配消息监听器所在的容器工厂配置类实例
    @Autowired
    private SimpleRabbitListenerContainerFactoryConfigurer factoryConfigurer;

    /**
     * 单一消费者
     * @return
     */
    @Bean(name = "singleListenerContainer")
    public SimpleRabbitListenerContainerFactory listenerContainer(){
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setMessageConverter(new Jackson2JsonMessageConverter());
        factory.setConcurrentConsumers(1);
        factory.setMaxConcurrentConsumers(1);
        factory.setPrefetchCount(1);
        return factory;
    }

    /**
     * 多个消费者
     * @return
     */
    @Bean(name = "multiListenerContainer")
    public SimpleRabbitListenerContainerFactory multiListenerContainer(){
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factoryConfigurer.configure(factory,connectionFactory);
        factory.setMessageConverter(new Jackson2JsonMessageConverter());
        factory.setAcknowledgeMode(AcknowledgeMode.NONE);
        factory.setConcurrentConsumers(10);
        factory.setMaxConcurrentConsumers(15);
        factory.setPrefetchCount(10);
        return factory;
    }

    /**
     * RabbitMQ发送消息的操作组件实例
     * @return
     */
    @Bean
    public RabbitTemplate rabbitTemplate(){
        connectionFactory.setPublisherConfirms(true);
        connectionFactory.setPublisherReturns(true);
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate.setMandatory(true);
        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> log.info("消息发送成功:correlationData({}),ack({}),cause({})",correlationData,ack,cause));
        rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> log.info("消息丢失:exchange({}),route({}),replyCode({}),replyText({}),message:{}",exchange,routingKey,replyCode,replyText,message));
        return rabbitTemplate;
    }
}

Springboot使用RabbitMQ

  1. 启动的配置类注入对应的队列，包括队列名称、交换机、路由。
  2. 使用 RabbitTemplate 作为生产者发送消息。
  3. @RabbitListener 作为消费者监听对应队列消费消息。
## 环境变量配置
mq.env=local
mq.basic.info.queue.name=${mq.env}.middleware.mq.basic.info.queue
mq.basic.info.exchange.name=${mq.env}.middleware.mq.basic.info.exchange
mq.basic.info.routing.key.name=${mq.env}.middleware.mq.basic.info.routing.key

/**
 * springboot 启动时去5672 端口监听配置的所有监听队列，
 * 若这个队列不存在，监听不到则会报错，需要在程序启动时注入这个队列
 */

//定义读取配置文件的环境变量实例
    @Autowired
    private Environment env;

    /**创建简单消息模型：队列、交换机和路由 **/

    //创建队列
    @Bean(name = "basicQueue")
    public Queue basicQueue(){
        return new Queue(env.getProperty("mq.basic.info.queue.name"),true);
    }

    //创建交换机：在这里以DirectExchange为例，在后面章节中我们将继续详细介绍这种消息模型
    @Bean
    public DirectExchange basicExchange(){
        return new DirectExchange(env.getProperty("mq.basic.info.exchange.name"),true,false);
    }
    //创建绑定
    @Bean
    public Binding basicBinding(){
        return BindingBuilder.bind(basicQueue()).to(basicExchange()).with(env.getProperty("mq.basic.info.routing.key.name"));
    }


 @Autowired
    private RabbitTemplate rabbitTemplate;
    @Autowired
    private Environment env;


    /**
     * 发送消息
     * @param message
     */
    public void sendMsg(String message){
        if (!ObjectUtils.isEmpty(message)){
            try {
                rabbitTemplate.setExchange(env.getProperty("mq.basic.info.exchange.name"));
                rabbitTemplate.setRoutingKey(env.getProperty("mq.basic.info.routing.key.name"));

                Message msg=MessageBuilder.withBody(message.getBytes("utf-8"))
                        .setDeliveryMode(MessageDeliveryMode.PERSISTENT).build();

                rabbitTemplate.convertAndSend(msg);

                log.info("基本消息模型-生产者-发送消息：{} ",message);
            }catch (Exception e){
                log.error("基本消息模型-生产者-发送消息发生异常：{} ",message,e.fillInStackTrace());
            }
        }
    }

@RabbitListener(queues = "${mq.basic.info.queue.name}",containerFactory = "singleListenerContainer")
  public void consumeMsg(@Payload byte[] msg){
      try {
          String message=new String(msg,"utf-8");
          log.info("基本消息模型-消费者-监听消费到消息：{} ",message);
      }catch (Exception e){
          log.error("基本消息模型-消费者-发生异常：",e.fillInStackTrace());
      }
  }

RabbitMQ常见消息模式

Simple 模式

Simple 模式是最简单的一个模式，由一个生产者，一个队列，一个消费者组成，生产者将消息通过交换机（此时，图中并没有交换机的概念，如不定义交换机，会使用默认的交换机）把消息存储到队列，消费者从队列中取出消息进行处理。

Fanout 模式

Fanout——发布订阅模式，是一种广播机制。

此模式包括：一个生产者、一个交换机 (exchange)、多个队列、多个消费者。生产者将消息发送到交换机，交换机不存储消息，将消息存储到队列，消费者从队列中取消息。如果生产者将消息发送到没有绑定队列的交换机上，消息将丢失。

Direct 模式

Direct 模式是在 Fanout 模式基础上添加了 routing key，Fanout（发布/订阅）模式是交换机将消息存储到所有绑定的队列中，而 Direct 模式是在此基础上，添加了过滤条件，交换机只会将消息存储到满足 routing key 的队列中。

Topic 模式

Topic 模式是生产者通过交换机将消息存储到队列后，交换机根据绑定队列的 routing key 的值进行通配符匹配，如果匹配通过，消息将被存储到该队列，如果 routing key 的值匹配到了多个队列，消息将会被发送到多个队列；如果一个队列也没匹配上，该消息将丢失。

RabbitMQ常见使用场景

解耦、削峰、异步

解耦

在微服务架构体系中，微服务A需要与微服务B进行通信，传统的做法是A调用B的接口。但这样做如果系统B无法访问或连接超时，系统A需要等待，直到系统B做出响应，并且A与B存在严重的耦合现象。如果引入消息队列进行系统AB的通信，流程是这样的：

  • 系统A将消息存储到消息队列中，返回成功信息
  • 系统B从队列中获取消息，进行处理操作
  • 系统A将消息放到队列中，就不用关心系统B是否可以获取等其他事情了，实现了两个系统间的解耦。

使用场景：

短信、邮件通知

削峰

系统A每秒请求100个，系统可以稳定运行，但如果在秒杀活动中，每秒并发达到1w个，但系统最大处理能力只能每秒处理 1000 个，所以，在秒杀活动中，系统服务器会出现宕机的现象。如果引入 MQ ，可以解决这个问题。每秒 1w个请求会导致系统崩溃，那我们让用户发送的请求都存储到队列中，由于系统最大处理能力是每秒1000个请求，让系统A每秒只从队列中拉取1000个请求，保证系统能稳定运行，在秒杀期间，请求大量进入到队列，积压到MQ中，而系统每秒只从队列中取1000个请求处理。这种短暂的高峰期积压是没问题的，因为高峰期一旦过去，每秒请求数迅速递减，而系统每秒还是从队列中取1000个请求进行处理，系统会快速将积压的消息消费掉。

使用场景：

秒杀活动
团抢活动

异步

用户注册，需要发送注册邮件和注册短信，传统的做法有两种：串行、并行。

  • 串行方式：将注册信息写库后（50ms），发送邮件（50ms），再发送短信（50ms），任务完成后，返回客户端，共耗时（150ms）
  • 并行方式：将注册信息写库后（50ms），开启子线程让发送邮件和发送短信同时进行（50ms），返回客户端，共耗时（100ms）
  • 引入MQ，将注册信息写库（50ms），将发送邮件和短信的操作写入队列（5ms），返回客户端，而消费者什么时候从队列中取消息进行处理，不用关心，共耗时（55ms）

使用场景：

将不是必须等待响应结果的业务逻辑进行异步处理

参考

版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 举报，一经查实，本站将立刻删除。

文章由极客之音整理，本文链接：https://www.bmabk.com/index.php/post/221983.html

赞 (0)
小半的头像小半
0 0

相关推荐

极客之音——专业性很强的中文编程技术网站，欢迎收藏到浏览器，订阅我们！