第二章 RabbitMQ相关操作 v1.0

RabbitMQ的基础架构可以概括为：消息发布者（Publish）产生消息并发送到队列，队列（Queue）负责接收和缓存消息，消费者（Consumer）负责处理队列中的消息。

第一部分 RabbitMQ基础

一、基本架构

二、简单消息模型

官方文档：https://www.rabbitmq.com/getstarted.html

1.基本消息队列(BasicQueue)

基本消息队列模型中只包含三个角色：

• Publish：消息发布者，产生消息并发送到队列
• Queue：消息队列，负责接收并缓存消息
• Consumer：消息消费者，处理队列中的消息

2.发送流程

1. 建立connection
2. 创建channel
3. 利用channel声明队列
4. 利用channel向队列发送消息

3.接收流程

1. 建立connection
2. 创建channel
3. 利用channel声明队列
4. 定义consumer的消费行为handleDelivery()
5. 利用channel将消费者与队列绑定

4.工作消息队列(WorkQueue)

工作消息队列模型中也只包含三个角色：

• Publish：消息发布者，产生消息并发送到队列
• Queue：消息队列，负责接收并缓存消息
• Consumer：消息消费者，处理队列中的消息

特点：

可以提高消息处理速度，避免消息堆积

三、发布订阅(Publish/Subscribe)模型

1.发布订阅模型

发布订阅模型在模型中引入了交换机(exchange)，可以实现将同一消息发送给多个消费者

常见的exchange类型有：Fanout、Direct、Topic

2.广播(Fanout Exchange)

将接收到的消息路由到每一个跟其绑定的queue

3.路由(Direct Exchange)

将接收到的消息路由到指定的跟其绑定的queue

每一个Queue都会与Exchange设置一个BindingKey，发布者发送消息时，指定消息的RoutingKey

4.主题(Topic Exchange)

将接收到的消息路由到指定的跟其绑定的queue(类似路由模式)

在路由模式基础上，Queue与Exchange指定BindingKey是可以使用通配符：#(若干字符);*(一个字符)

第二部分 SpringAMQP

一、AMQP和SpringAMQP

AMQP(Advanced Message Queuing Protocols)是用于在应用程序或之间传递业务消息的开放标准。该协议与语言和平台无关，更符合微服务中独立性的要求。

Spring AMQP是基于AMQP协议定义的一套API规范， 提供了模板来发送和接收消息。Spring AMQP包含两部分，其中spring- amqp是基础抽象，spring-rabbit是底层的默认实现。

二、消息预取机制

在RabbitMQ中，存在消息预取机制，当Consumer处理能力不等时能力差的消费者会出现消息堆积

可以通过在配置文件设置prefetch值来限制预取消息的上限解决该问题

spring:
  rabbitmq:
    listener:
      simple:
      	prefetch: 1 #每次只能获取一条消息，处理完成才能获取下一个消息

三、功能实现

1.基础消息队列模型

1. 在父工程中引入spring-amqp依赖

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

1. 消息生产者利用RabbitTemplate发送消息到队列

spring:
  rabbitmq:
	 host: localhost
	 port: 5672  # 端口
	 virtual-host: /  # 虚拟主机名
	 username: username
	 password: <password>

@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringAmqpTest {
    @Autowired
    private RabbitTempLate rabbitTemplate;  //RabbitTempLate自动装配
    @Test
    public void testSimpLeQueue() {
	    String queueName = "name.queue";
        String message = "hello, world!" ;
        rabbitTemplate.convertAndSend(queueName, message);  //转换并发送消息
    }
}

1. 消息消费者绑定队列进行消费

spring:
  rabbitmq:
	 host: localhost
	 port: 5672  # 端口
	 virtual-host: /  # 虚拟主机名
	 username: username
	 password: <password>

@Component
public class SpringRabbitListener {
    @RabbitListener(queues = "name.queue")
    public void listenSimpleQueueMessage(String msg) throws Exception {
        System.out.println("spring消费者接收到消息: [" + msg + "] ");
    }
}

2.发布订阅模型

a.广播模式

1. 在consumer声明Exchange、Queue、Binding

@Configuration
public class FanoutConfig {
    @Bean  //声明FanoutExchange交换机
    public FanoutExchange fanoutExchange(){
        return new FanoutExchange ("name.fanout");
    }
    @Bean  //声明第1个队列
    public Queue fanoutQueue1(){
        return new Queue ("fanout.queue1");
    }
    @Bean  //綁定队列1和交换机
    public Binding bindingQueue1(Queue fanoutQueue1, FanoutExchange fanoutExchange){
	    return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);
    }
    //...以相同方式声明第2,3,...个队列，并完成綁定
}

1. 在consumer声明多个消费者

@RabbitListener(queues = "fanout.queue1")
public void listenFanoutQueue1(String msg) {  //消费逻辑
    System.out.println("消费者1接收到Fanout消息: [" + msg + "] ");
}
@RabbitListener(queues = "fanout.queue2" )
public void listenFanoutQueue2(String msg) {  //消费逻辑
    System.out.println("消费者2接收到Fanout消息: [" + msg + "] ");
}

b.路由模式

@RabbitListener(bindings = @QueueBinding(
    value =@Queue(name="direct.queue1"),  //指定队列名称
    exchange = @Exchange(name="name.direct", type = ExchangeTypes.DIRECT),  //指定交换机
    key = {"one", "two"}
))
public void listenFanoutQueue1(String msg) {  //消费逻辑
    System.out.println("消费者1接收到direct消息: [" + msg + "] ");
}

@RabbitListener(bindings = @QueueBinding(
    value =@Queue(name="direct.queue2"),  //指定队列名称
    exchange = @Exchange(name="name.direct", type = ExchangeTypes.DIRECT),  //指定交换机
    key = {"one", "three"}
))
public void listenFanoutQueue2(String msg) {  //消费逻辑
    System.out.println("消费者2接收到direct消息: [" + msg + "] ");
}

c.主题模式

@RabbitListener(bindings = @QueueBinding(
    value =@Queue(name="topic.queue1"),  //指定队列名称
    exchange = @Exchange(name="name.topic", type = ExchangeTypes.TOPIC),  //指定交换机
    key = "A.#"
))
public void listenFanoutQueue1(String msg) {  //消费逻辑
    System.out.println("消费者1接收到topic消息: [" + msg + "] ");
}

@RabbitListener(bindings = @QueueBinding(
    value =@Queue(name="topic.queue2"),  //指定队列名称
    exchange = @Exchange(name="name.topic", type = ExchangeTypes.TOPIC),  //指定交换机
    key = "B.#"
))
public void listenFanoutQueue2(String msg) {  //消费逻辑
    System.out.println("消费者2接收到topic消息: [" + msg + "] ");
}

d.生产者发送消息到交换机

@Test
public void testFanoutExchange() {
    //指定队列名称
    String exchangeName = "itcast.fanout";
    //指定消息
    String message = "hello, everyone!";
    //发送消息，参数分别是:交换机名称、RoutingKey(暂时为空)、消息
    rabbitTemplate.convertAndSend(exchangeName, "", message);
}

四、消息转换器

在SpringAMQP的发送方法中，接收消息的类型是0bject，也就是说我们可以发送任意对象类型的消息，默认SpringAMQP的消息转换器会帮我们序列化为字节后发送。

要修改消息转换器的转换操作，需要定义一个MessageConverter类型的Bean，如使用JSON方式序列化：

• Publisher(生产者端)&Consumer(消费者端)操作相同

<dependency>
  <groupId>com.fasterxml.jackson.dataformat</groupId>
  <artifactId>jackson-dataformat-xml</artifactId>
  <version>2.9.10</version>
</dependency>

@Bean
public MessageConverter jsonMessageConverter() {
    return new Jackson2JsonMessageConverter();
}