RabbitMQ入门

RabbItMQ核心概念：

​ RabbitMQ是一个开源的消息代理和队列服务器，用来通过普通协议在完全不同的应用之间共享数据，RabbitMQ是用Erlang语言来编写的，并且RabbitMQ是基于AMQP协议的。

1、RabbitMQ的高性能之道是如何做到的？

​ Erlang语言最初在于交换机领域的架构模式，这样使得RabbitMQ在Broker之间进行数据交互的性能是非常优秀的。

​ Erlang的优点：Erlang有着和原生Socket一样的延迟。

2、什么是AMQP高级消息队列协议？

​ AMQP：是具有现代特征的二进制协议。是一个提供统一消息服务的应用层标准高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。

AMQP核心概念：

  • Server：又称Broker，接受客户端的连接，实现AMQP实体服务

  • Connection:连接，应用程序与Broker的网络连接

  • Channel：网络信道，几乎所有的操作都在Channel中进行，Channel是进行消息读写的通道。客户端可建立多个Channel，每个Channel代表一个会话任务。

  • Message:消息，服务器和应用程序之间传送的数据，由Properties和Body组成。Properties可以对消息进行修饰，比如消息的优先级、延迟等高级特性；Body则就是消息体内容。

  • Virtual host：虚拟地址，用于进行逻辑隔离，最上层的消息路由。一个Virtual Host里面可以有若干个Exchange和Queue，同一个Virtual host里面不能有相同名称的Exchange和Queue。

  • Exchange:交换机，接收消息，根据路由键转发消息到绑定的队列

  • Binding:Exchange和Queue之间的虚拟连接，binding中可以包含Routing key

  • Routing Key ：一个路由规则，虚拟机可以用它来确定如何路由一个特定消息。

  • Queue：也称为Message Queue，消息队列，保存消息并将它们转发给消费者

3、RabbitMQ消息是如何流转的？

4、Exchange交换机详解

​ 接收消息，并根据路由键转发消息到所绑定的队列。

交换机属性：

  • Name:交换机名称

  • Type：交换机类型为direct、topic、fanout、headers

    • Direct Exchange：所有发送到Direct Exchange的消息被转发到RouteKey中指定的Queue（不需要进行任何绑定操作）

    • Topic Exchage:所有发送到Topic Exchange的消息被转发到所有关心RouteKey中指定Topic的Queue上,Exchange将RouteKey和某Topic进行模糊匹配，此时队列需要绑定一个Topic。

Fanout Exchange:

​ 1、不处理路由键，只需要简单的将队列绑定到交换机上

	2、发送到交换机的消息都会被转发到与该交换机绑定的所有队列上

​ 3、Fanout交换机转发消息是最快的。（不做任何匹配处理）

Queue-消息队列

  • 消息队列，实际存储消息数据
  • Durability：是否持久化，Durable：是，Transient：否
  • Auto delete：如选yes，代表当最后一个监听被移除后，该Queue会自动被删除

Virtual host-虚拟主机：

  • 虚拟地址，用于进行逻辑隔离，最上层的消息路由
  • 一个Virtual Host里面可以有若干个Exchange和Queue
  • 同一个Virtual Host里面不能有相同名称的Exchange和Queue

RabbitMQ高级特性：

1、消息如何保障100%的投递成功？

  • 消息落库，对消息状态进行打标

  • 消息的延迟投递，做二次确认，回调检查

2、消费端—幂等性保障

  • 唯一ID+指纹码机制

    好处：实现简单

    坏处：高并发下有数据库写入的性能瓶颈

    解决方案：根据id进行分库分表进行算法路由

  • 利用redis原子特性实现

3、Confirm确认消息

理解Comfirm消息确认机制：

  • 消息的确认，是指生产者投递消息后，如果Broker收到消息，则会给我们生产者一个应答。
  • 生产者进行接收应答，用来确定这条消息是否正常的发送到Broker，这种方式也是消息的可靠性投递的核心保障！

如何实现Confirm确认消息？

​ 第一步：在channel上开启确认模式：channel.confirmSelect()

​ 第二步：在channel上添加监听：addConfirmListener，监听成功和失败的 返回结果，根据具体的结果对消息进行重新发送、或记录日志等后续处理！

Return消息机制：Return Listener 用于处理一些不可路由的消息！

  • 被 broker 接收到只能表示 message 已到达服务器，并不能保证消息一定会被投递到目标 queue。所以还需 returnCallback 。
  • 我们的消息生产者，通过指定一个Exchange和RoutingKey，把消息送达到其一个队列中去，然后我们的消费者监听队列，进行消费处理操作！
  • 但是在某些情况下，如果我们在发送消息的时候，当前的Exchange不存在或者指定的路由key路由不到，这个时候如果我们需要监听这种不可达的消息，就要使用Return Listener！

4、TTL队列、消息

  • TTL是Time To Live的缩写，也就是生存时间
  • RabbitMQ支持消息的过期时间，在消息发送时可以进行指定
  • RabbitMQ支持队列的过期时间，从消息入队列开始计算，只要超过队列超时时间配置，那么消息会自动的清除

5、死信队列

利用DLX，当消息在一个队列中变成了死信之后，它能被重新publish到另一个Exchange，这个Exchange就是DLX。

消息变成死信有以下几种情况：

  • 消息被拒绝，并且requeue=false
  • 消息TTL过期
  • 队列达到最大长度

6、RabbitMQ的使用场景：

1、跨系统的异步通信

​ 2、多个应用之间的耦合

​ 3、应用内的同步变异步

​ 4、消息驱动的架构

​ 5、应用需要更灵活的耦合方式

​ 6、跨局域网

Springboot整合RabbitMQ

消息提供者：

先导入相对应的依赖

 <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>

然后编写消息提供者的配置文件

#对于rabbitMQ的支持
spring.rabbitmq.host=127.0.0.1
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
#项目的名称
spring.application.name=rabbitmq-provider

再编写消息提供者的配置类，写好消息的队列、交换机、绑定策略等

@Configuration
public class RabbitConfig {

    @Bean
    public Queue TestDirectQueue(){
        return new Queue("TestDirectQueue1111",true);
    }

    @Bean
    public DirectExchange TestDirectExchange(){
        return new DirectExchange("TestDirectExchange",true,false);
    }
    @Bean
    public Binding bindingDirect() {
        return BindingBuilder.bind(TestDirectQueue()).to(TestDirectExchange()).with("TestDirectRouting");
    }
    @Bean
    DirectExchange lonelyDirectExchange() {
        return new DirectExchange("lonelyDirectExchange");
    }
}

最后在编写消息提供者的接口层

@RestController
public class SendMessageController {
    @Autowired
    RabbitTemplate rabbitTemplate;

    @GetMapping("/sendDirectMessage")
    public String sendDirectMessage() {
        String messageId = String.valueOf(UUID.randomUUID());
        String messageData = "test message, hello!";
        String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
        Map<String,Object> map=new HashMap<>();
        map.put("messageId",messageId);
        map.put("messageData",messageData);
        map.put("createTime",createTime);
        //将消息携带绑定键值：TestDirectRouting 发送到交换机TestDirectExchange
        rabbitTemplate.convertAndSend("TestDirectExchange", "TestDirectRouting", map);
        return "ok";
    }

}

启动项目，访问该接口，就能在RabbitMQ的可视化界面可以看到该队列已经存在。

RabbitMQ消费者：

一样的先导入依赖：

<dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>

配置application.properties配置文件

server.port=8081
#对于rabbitMQ的支持
spring.rabbitmq.host=192.168.202.231
spring.rabbitmq.port=5672
spring.rabbitmq.username=oilchem
spring.rabbitmq.password=PoHF*fn%
#项目的名称
spring.application.name=rabbitmq-consumer

这时候其实消费者不需要再写配置类了，直接写相关业务

@Component
@RabbitListener(queues = "TestDirectQueue1111")
public class DirectReceiver {

    @RabbitHandler
    public void process(Map testMessage){
        System.out.println("DirectReceiver消费者接收的消息为"+testMessage);
    }
}

这时候启动消息消费者就能取到指定队列中的消息了

这样就实现了一个最简单的Springboot整合RabbitMQ的案例了。

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