AMQP与RabbitMQ

转载请注明原文地址:https://www.cnblogs.com/ygj0930/p/10922433.html

一:AMQP是什么

  AMQP(Advanced Message Queuing Protocol)高级消息队列协议,是基于JMS进行进一步扩展和优化的异步消息处理协议。

  其在JMS的基础上,提供了更多的方法。

  AMQP引入了消息交换机Exchange的概念,实现了消息生产者与消息队列之间的解耦。消息不再直接发送到队列或者主题,而是统一发送给Exchange,由交换机根据路由规则,将消息分发到不同队列中。

  AMQP还引入了Channel概念,将一个connection细分为不同channel,适用于多线程场景下,消息消费者与AMQP服务器只需建立一个TCP连接即可,各个线程对应不同channel,通过channel实现消息的提取。

  

二:AMQP模型

  • Broker: 接收和分发消息的应用,RabbitMQ Server就是一个消息处理实体。
  • Virtual host: 出于多租户和安全因素设计的,把AMQP的基本组件划分到一个虚拟的分组中,类似于网络中的namespace概念。当多个不同的用户使用同一个RabbitMQ server提供的服务时,可以划分出多个vhost,每个用户在自己的vhost创建exchange/queue等。
  • Connection: publisher/consumer和broker之间的TCP连接。断开连接的操作只会在client端进行,Broker不会断开连接,除非出现网络故障或broker服务出现问题。
  • Channel: 如果每一次访问RabbitMQ都建立一个Connection,在消息量大的时候建立TCP Connection的开销将是巨大的,效率也较低。Channel是在connection内部建立的逻辑连接,如果应用程序支持多线程,通常每个thread创建单独的channel进行通讯,AMQP method包含了channel id帮助客户端和message broker识别channel,所以channel之间是完全隔离的。Channel作为轻量级的Connection极大减少了操作系统建立TCP connection的开销。
  • Exchange: message到达broker的第一站,根据分发规则,匹配查询表中的routing key,分发消息到queue中去。常用的类型有:direct (point-to-point), topic (publish-subscribe) and fanout (multicast)。
  • Queue: 消息最终被送到这里等待consumer取走。一个message可以被同时拷贝到多个queue中。
  • Binding: exchange和queue之间的虚拟连接,binding中可以包含routing key。Binding信息被保存到exchange中的查询表中,用于message的分发依据。

三:AMQP消息传送模型

  

  发布者(Publisher)发布消息(Message),传送到broker;

  在broker中,消息被交换机(Exchange)根据路由规则,通过binding传送到不同的队列;

  最后 AMQP 代理会将消息投递给订阅了此队列的消费者(push API),或者消费者按照需求自行获取(pull API)。

四:Exchange分发策略

  Direct:当消息的routing key 与 binding 的 routing key 直接匹配,消息路由到该队列

  Topic:   当消息routing key 与 binding 的 routing key 符合通配符匹配,消息路由到该队列(请百度通配符匹配)

  Headers:  当消息参数表中的头信息和值都与 binding参数表中匹配的话,消息路由到该队列

    Fanout: 任何消息直接匹配到所有队列上

五:RabbitMQ的使用

  1、SpringBoot集成RabbitMQ

  1)配置pom包:spring-boot-starter-amqp

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

  2)配置RabbitMQ服务器信息

  首先,需要一台主机安装并启动RabbitMQ。

  然后在项目中配置:

spring.application.name=应用名

spring.rabbitmq.host=RabbitMQ服务器的host
spring.rabbitmq.port=5672
spring.rabbitmq.username=RabbitMQ服务器的登录帐号
spring.rabbitmq.password=密码

  3)配置消息队列

@Configuration
public class RabbitConfig {

    @Bean
    public Queue Queue() {
        return new Queue("hello");
    }

}

  4)使用模版,实现消息生产者

@component
public class HelloSender {

    @Autowired
    private AmqpTemplate rabbitTemplate;

    public void send() {
        String context = "hello " + new Date();
        System.out.println("Sender : " + context);
        this.rabbitTemplate.convertAndSend("hello", context);//自动转换为消息对象并要求发送到hello队列
    }

}

  5)实现消息接收者——创建监听器,监听hello队列,一旦有消息则调用process函数进行处理

@Component
@RabbitListener(queues = "hello")
public class HelloReceiver {

    @RabbitHandler
    public void process(String hello) {
        System.out.println("Receiver  : " + hello);
    }

}

  2、Python使用RabbitMQ

  1)下载RabbitMQ并解压

  下载网址: http://www.rabbitmq.com/install-generic-unix.html 

  解压后,进入 sbin 目录, 运行 server。

  默认端口为5672。

  2)pip安装AMQP协议实现模块——pika

  3)消息生产者:

# -*- coding: utf-8 -*-

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.exchange_declare(exchange='first', type='fanout')

channel.queue_declare(queue='hello')
channel.queue_bind(exchange='first', queue='hello')

channel.basic_publish(exchange='first', routing_key='', body='Hello World!')
  • 获取连接.
  • 从连接上获取一个 channel.
  • 声明一个 exchange . (只会创建一次)
  • 声明一个 queue . (只会创建一次)
  • 把 queue 绑定到 exchange 上.
  • 向指定的 exchange 发送一条消息.

  4)消息消费者

# -*- coding: utf-8 -*-

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.queue_declare(queue='hello')

def callback(ch, method, properties, body):#消息处理函数
    print body

channel.basic_consume(callback, queue='hello', no_ack=True)
channel.start_consuming()
  • 获取连接.
  • 从连接上拿到 channel .
  • 声明需要的 queue .
  • 定义一个从指定 queue 获取消息的回调处理.
  • 开始接收消息.

  5)查看rabbitMQ的服务状态

  使用 rabbitmqctl 这个工具。例如:查看当前的队列情况

./rabbitmqctl list_queues

  

六:深入

  • 持久化
  • 调度策略
  • 分配策略
  • 状态反馈
原文地址:https://www.cnblogs.com/ygj0930/p/10922433.html