RabbitMQ基础简介

简介：

    rabbitMQ是一个消息服务的具体实现，那么什么是消息服务？

     消息服务就是两个程序时间进行通讯的标准。这里我介绍两种消息服务。一种是JMS（Java Message Service）java消息服务，通过统一的java API层面的标准，使得多个消息客户端通过JMS进行交互。另一种是AMQP（Advanced Message Queueing Protocol）高级消息队列，是协议层面的规范，因此是可以跨平台的。

    ActiveMQ是基于JMS标准实现的（两种消息发布模型：点对点和发布者订阅者模式），RabbitMQ是基于AMQP实现的，他们也称消息队列，是进程或线程之间的一种异步通信的方式。如果使用消息队列的话，消息生产者会将消息保存到消息队列里面，直到消费者消费。生产者和消费者不需要同时和消息队列交互。实现了解耦，提高系统的可靠性和可扩展性。

应用场景：

  1. 异步处理：假设用户注册信息到数据库后，需要发送短信和邮箱通知，这是三个操作当这三个操作都执行完，在响应给用户。而当引用了消息队列的时候，最重要的是用户注册。而发送邮箱和短信只是一个通知，不影响数据的改变，因此将这两个不是必须业务逻辑异步处理。也就是放到消息队列中去。由消息队列统一分发。
  2. 应用解耦：假设用户下单后，订单通知库存系统，传统的做法就是订单系统调用库存系统的接口，如果库存系统挂了。那么订单就会失效。所以订单和库存是高度耦合的。而使用消息队列，用户下单后，完成订单系统持久化处理，将消息写入消息队列，返回用户下单成功。库存系统订阅下单消息，获取下单消息，进行库操作。就算库存出现故障，消息队列也能保证消息的可靠性。
  3. 流量削峰：假设一个秒杀活动，如果流量上亿，就会导致系统崩溃，使用消息队列，将用户请求写入消息队列中，加入消息队列长度超过最大值，控制订阅人数，当人数达到一定程度，就将请求直接丢弃或返回错误信息。减缓了短时间的高流量压垮服务器。

基础概念：

  • Queue（队列）：它的作用是用来存储消息。特征是先进先出。生产者最终将消息送到RabbitMQ的内部对象消息队列中。而消费者从消息队列中去取消息。生产者发送的消息被传送到消息队列中，消费者发现消息队列中有订阅的消息。就会将消息取出进行业务操作。这是一个消费者对应一个队列。也可以有多个消费者同时订阅一个队列，但是如果消费者对消息的处理时间不同，就会导致某些消费者一直在忙碌，有些则一直处于空闲，因此可以限制每次发送消息的个数。
  • Exchange(消息交换机):它指定消息按什么规则,路由到哪个队列。假设生产者a和b，当它发消息的时候不知道发的消息是给队列a还是给队列b，所以需要一个规则来告诉它。这里有用到了Exchange。生产者发送消息到队列要经过Exchange，由Exchange将消息路由到一个或多个队列。如果不符合路由规则的将直接丢弃。Exchange有四种类型，不同类型有不同的策略，不同的策略决定绑定的队列不同，例如生产者发送了一条消息，Routing Key的规则是A，那么生产者会将这个key为A的消息推送到Exchange中，Exchange会根据对应的规则删选生产者发的消息。如果对应上了，就推送到对应的队列中去。那么Exchange的规则有哪些呢？
  • 规则一：fanout类型，它会将所有消息路由到所有与他绑定的Queue中。
  • 规则二： direct类型的Exchange路由规则是把消息路由到那些binding key与routing key完全匹配的Queue中。
  • 规则三：topic这个规则就是模糊匹配，可以通过通配符满足一部分规则就可以传送。direct规则是严格意义上的匹配，Routing Key必须与Binding Key相匹配的时候才将消息传送给Queue。
  • 规则四：headers类型的Exchange不依赖于routing key与binding key的匹配规则来路由消息，而是根据发送的消息内容中的headers属性进行匹配。在绑定Queue与Exchange时指定一组键值对；当消息发送到Exchange时，RabbitMQ会取到该消息的headers（也是一个键值对的形式），对比其中的键值对是否完全匹配Queue与Exchange绑定时指定的键值对；如果完全匹配则消息会路由到该Queue，否则不会路由到该Queue。
  • Connection，是RabbitMQ的Socket链接，封装了socket协议和部分逻辑，它会建立一个TCP连接，生产者和消费者同过tcp连接到RabbitMQ服务上。
  • ConnectionFactory：是Connection的制造工厂，用于生产Connection
  • Channel是RabbitMQ的重要一个接口，大部分业务操作在它里面完成，包括定义队列，消息交换机，绑定Queue和Exchange，发布消息等。它是虚拟连接，建立在Connection，tcp连接的基础上，来进行数据流动，为什么不直接建立在tcp上进行数据流动，是因为，频繁的创建和关闭tcp，对系统的性能有很大的影响，而且tcp连接是有限制的，这也限制了处理高并发的能力。

代码实现direct类型的Exchange路由规则：

一：创建Spring Boot项目，勾选rabbitMQ依赖，配置Rabbitmq的连接信息

spring.rabbitmq.host=localhost
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
spring.rabbitmq.port=5672

二：创建dierct配置类：

package com.zl.rabbitmq.config;

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.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;




@Configuration
public class rabbitDirectConfig {
    //ctrl+shift+u=小写替换成大写
    private final static String DIRECTNAME="zl";

//队列
    @Bean
    Queue queue() {

      return new Queue("hello ,zl");
    }

    @Bean
    DirectExchange directExchange(){
        //重启后是否有效
        //长期未使用是否删除
       return new DirectExchange(DIRECTNAME,true,false);
    }
    //将队列和DirectExchange绑定到一起
    Binding binding(){
        return BindingBuilder.bind(queue()).to(directExchange()).with("direct");
        }
}

      三：创建消费者：

package com.zl.rabbitmq.receiver;

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

//消费者
@Component
public class DirecReceiver {
    @RabbitListener(queues = "hello ,zl")
    public void handler(String msg){

        System.out.println("handler"+msg);
    }
}

四：发送消息：

package com.zl.rabbitmq;

import org.junit.jupiter.api.Test;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;

@SpringBootTest
class RabbitmqApplicationTests {
@Autowired
RabbitTemplate rabbitTemplate;
    @Test
    void contextLoads() {
        rabbitTemplate.convertAndSend("hello ,zl","hello zll");
    }

}

五：测试：

2020-04-11 12:14:05.442  INFO 32 --- [           main] o.s.a.r.c.CachingConnectionFactory       : Attempting to connect to: [localhost:5672]
2020-04-11 12:14:05.474  INFO 32 --- [           main] o.s.a.r.c.CachingConnectionFactory       : Created new connection: rabbitConnectionFactory#4e558728:0/SimpleConnection@43effd89 [delegate=amqp://guest@127.0.0.1:5672/, localPort= 50208]
2020-04-11 12:14:05.567  INFO 32 --- [           main] o.s.b.w.embedded.tomcat.TomcatWebServer  : Tomcat started on port(s): 8080 (http) with context path ''
2020-04-11 12:14:05.621  INFO 32 --- [           main] com.zl.rabbitmq.RabbitmqApplication      : Started RabbitmqApplication in 3.968 seconds (JVM running for 12.158)
handlerhello zll

流程：当消息到达DirectExchange的时候，会被转发到和该条消息的routing key相同的队列中去。

消息发送者发送一条"hello zll"的消息，并指定rountingkey为"hello ,zl"。然后消息到达DirectExchange中，由binding将Exchange和队列绑定在一起，将这个消息转发到和该条消息的routing key相同的队列中。然后消费者监听这个routing key的队列。当有消息时，就打印出来。

六：在实际应用场景中的使用：

本篇参考我的下一篇博客 RabbitMQ在java中的使用

【推广】 免费学中医，健康全家人
原文地址：https://www.cnblogs.com/javazl/p/12678030.html