一、消息队列
消息队列（英语：Message queue）是一种进程间通信或同一进程的不同线程间的通信方式，软件的贮列用来处理一系列的输入，通常是来自用户。

消息队列提供了异步的通信协议，每一个贮列中的纪录包含详细说明的数据，包含发生的时间，输入设备的种类，以及特定的输入参数，也就是说：消息的发送者和接收者不需要同时与消息队列交互。消息会保存在队列中，直到接收者取回它。

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

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

RabbitMQ是由erlang语言开发，基于AMQP（Advanced Message Queue 高级消息队列协议）协议实现的消息队列，它是一种应用程序之间的通信方法，消息队列在分布式系统开发中应用非常广泛。

三、代码示例

在写代码之前，需先具备rabbitmq环境，我这里是在虚拟机上通过docker安装的，启动docker容器，打开浏览器输入虚拟机IP+15672端口即可访问rabbitmq管理界面，用户名和密码默认是guest。

然后创建项目，父工程pom.xml引入依赖如下：


        
            org.projectlombok
            lombok
        
        
        
            org.springframework.boot
            spring-boot-starter-amqp
        
        
        
            org.springframework.boot
            spring-boot-starter-test
        
        
            com.fasterxml.jackson.core
            jackson-databind

接着创建一个生产者和一个消费者子项目，目录结构如图：

配置文件配好rabbitmq的信息

spring:
  rabbitmq:
    host: 192.168.211.128 # rabbitMQ的ip地址
    port: 5672 # 端口
    username: guest
    password: guest
    virtual-host: / # 虚拟机地址

1.基础消息队列

编写测试类SpringAmqpTest，队列命名为simple.queue

@RunWith(SpringRunner.class)将测试类添加到Spring容器中，自动注入才能生效

@SpringBootTest注解会自动检索程序的配置文件，检索顺序是从当前包开始，逐级向上查找被@SpringBootApplication或@SpringBootConfiguration注解的类。

package com.demo.mq;

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

import java.util.HashMap;
import java.util.Map;


@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringAmqpTest {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Test
    public void testSendMessage2SimpleQueue() {
        String queueName = "simple.queue";
        String message = "Hello,world";
        rabbitTemplate.convertAndSend(queueName, message);
    }
}

编写消费者监听类：

package cn.itcast.mq.listener;

import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.time.LocalTime;
import java.util.Map;

@Component
public class SpringRabbitListener {

     @RabbitListener(queues = "simple.queue")
     public void listenSimpleQueue(String msg) {
         System.out.println("消费者接收到simple.queue的消息：【" + msg + "】");
     }

}

先启动消费者的启动类，然后运行生产者的测试类方法：

2.WorkQueue模型

上面的例子中只有一个消费者，所以采用了一个简单的基础模型，WorkQueque模型则是应对多个消费者的情况。

消费者监听类的中定义2个消费者，并且让每个消费者收到一条消息后都睡眠一段时间，其中消费者1睡眠20毫秒，消费者2睡眠200毫秒，那么理论上消费者1在1秒内可以接收50条左右的消息，消费者2在1秒内可以接收5条左右的消息。

package cn.itcast.mq.listener;

import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.time.LocalTime;
import java.util.Map;

@Component
public class SpringRabbitListener {

    @RabbitListener(queues = "simple.queue")
    public void listenWorkQueue1(String msg) throws InterruptedException {
        System.out.println("消费者1接收到消息：【" + msg + "】" + LocalTime.now());
        Thread.sleep(20);
    }

    @RabbitListener(queues = "simple.queue")
    public void listenWorkQueue2(String msg) throws InterruptedException {
        System.err.println("消费者2........接收到消息：【" + msg + "】" + LocalTime.now());
        Thread.sleep(200);
    }
}

编写生产者的测试方法，循环发送50条消息：

package com.demo.mq;

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

import java.util.HashMap;
import java.util.Map;


@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringAmqpTest {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Test
    public void testSendMessage2WorkQueue() {
        String queueName = "simple.queue";
        String message = "Hello,message--";
        for (int i = 1; i <= 50; i++) {
            rabbitTemplate.convertAndSend(queueName, message + i);
        }
    }
}

先启动消费者的启动类，然后运行生产者的测试类方法：

可以很明显的看到消费者1接收的都是奇数位的消息，消费者2接收的都是偶数位的消息，并且消费者1几乎1秒内就消费掉了25条的消息，而消费者2因为睡眠时间更长，导致剩下的25条消息处理得很慢。

这里的睡眠时间可以看做是实际项目中2个服务或者2台服务器处理的速度，消费者1处理更快，所以早早的结束了，剩下的消息则由消费者2慢慢的消费。对于对效率有要求的业务，这样肯定是不行的，所以Springamqp支持在配置文件中设置每次只能获取一条消息，处理完成才能获取下一个消息，这样就能保证效率更高的消费者“夺取”更多的消息，使得消息在更短的时间内消费掉。

在消费者项目中配置文件加上上述配置:

spring:
  rabbitmq:
    host: 192.168.211.128 # rabbitMQ的ip地址
    port: 5672 # 端口
    username: guest
    password: guest
    virtual-host: /
    listener:
      simple:
        prefetch: 1 # 每次只能获取一条消息，处理完成才能获取下一个消息

重新执行，消费者1消费了比消费者2更多的消息：

3.发布订阅模型-FanoutExchange

发布订阅模型示意图

此模型在前面2个例子的基础上，增加了一个exchange（交换机），从示意图中可以看出，生产者不再直接将消息发送给队列，而是先发送给exchange，再由exchange发送到队列，最后由消费者接收队列里的消息。

此处将换机的作用主要为：

⑴接收publisher发送的消息

⑵将消息按照规则路由到与之绑定的队列

需要注意的是：交换机不能缓存消息，路由失败的话，消息就会丢失

FanoutExchange会将消息路由到每个绑定的队列

编写配置类，定义交换机和队列，并且绑定起来，此处定义2个队列，方便演示：

package cn.itcast.mq.config;

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class FanoutConfig {
    // itcast.fanout
    @Bean
    public FanoutExchange fanoutExchange(){
        return new FanoutExchange("itcast.fanout");
    }

    // fanout.queue1
    @Bean
    public Queue fanoutQueue1(){
        return new Queue("fanout.queue1");
    }

    // 绑定队列1到交换机
    @Bean
    public Binding fanoutBinding1(Queue fanoutQueue1, FanoutExchange fanoutExchange){
        return BindingBuilder
                .bind(fanoutQueue1)
                .to(fanoutExchange);
    }

    // fanout.queue2
    @Bean
    public Queue fanoutQueue2(){
        return new Queue("fanout.queue2");
    }

    // 绑定队列2到交换机
    @Bean
    public Binding fanoutBinding2(Queue fanoutQueue2, FanoutExchange fanoutExchange){
        return BindingBuilder
                .bind(fanoutQueue2)
                .to(fanoutExchange);
    }
}

编写监听这2个队列的消费者：

package cn.itcast.mq.listener;

import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.time.LocalTime;
import java.util.Map;

@Component
public class SpringRabbitListener {

    @RabbitListener(queues = "fanout.queue1")
    public void listenFanoutQueue1(String msg) {
        System.out.println("消费者接收到fanout.queue1的消息：【" + msg + "】");
    }
    @RabbitListener(queues = "fanout.queue2")
    public void listenFanoutQueue2(String msg) {
        System.out.println("消费者接收到fanout.queue2的消息：【" + msg + "】");
    }
}

编写生产者测试方法,此处convertAndSend方法与上面的不同，这里有3个参数，第一个参数是交换机名称，第二个参数是路由key（这里为空），第三个参数是消息：

package com.demo.mq;

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

import java.util.HashMap;
import java.util.Map;


@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringAmqpTest {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Test
    public void testSendMessage2FanoutExchange() {
        String exchange = "itcast.fanout";
        String message = "Hello, everyone";
        rabbitTemplate.convertAndSend(exchange, "", message);
    }
}

测试结果，2个消费者都接收到了消息：

4.发布订阅模型-DirectExchange

本例中则用到上面例子中convertAndSend方法中没有定义的第二个参数。

编写消费者监听方法，定义directexchange和2个队列，bindingKey分别为blue，red和yellow，red

package cn.itcast.mq.listener;

import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.time.LocalTime;
import java.util.Map;

@Component
public class SpringRabbitListener {

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "direct.queue1"),
            exchange = @Exchange(name = "itcast.direct", type = ExchangeTypes.DIRECT),
            key = {"red", "blue"}
    ))
    public void listenDirectQueue1(String msg){
        System.out.println("消费者接收到direct.queue1的消息：【" + msg + "】");
    }

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "direct.queue2"),
            exchange = @Exchange(name = "itcast.direct", type = ExchangeTypes.DIRECT),
            key = {"red", "yellow"}
    ))
    public void listenDirectQueue2(String msg){
        System.out.println("消费者接收到direct.queue2的消息：【" + msg + "】");
    }
}

编写测试方法，发送消息给bindingKey为blue的队列：

package com.demo.mq;

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

import java.util.HashMap;
import java.util.Map;


@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringAmqpTest {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Test
    public void testSendMessage2DirectExchange() {
        String exchange = "itcast.direct";
        String message = "Hello, blue";
        rabbitTemplate.convertAndSend(exchange, "blue", message);
    }
}

测试结果：

再将bindingKey改为red，消息也改为red

@Test
    public void testSendMessage2DirectExchange() {
        String exchange = "itcast.direct";
        String message = "Hello, red";
        rabbitTemplate.convertAndSend(exchange, "red", message);
    }

测试结果：

5.发布订阅模型-TopicExchage

topic和direct比较类似，区别在于可以使用通配符来进行模糊匹配。

编写消费者代码：

package cn.itcast.mq.listener;

import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.time.LocalTime;
import java.util.Map;

@Component
public class SpringRabbitListener {

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "topic.queue1"),
            exchange = @Exchange(name = "itcast.topic", type = ExchangeTypes.TOPIC),
            key = "china.#"
    ))
    public void listenTopicQueue1(String msg){
        System.out.println("消费者接收到topic.queue1的消息：【" + msg + "】");
    }

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "topic.queue2"),
            exchange = @Exchange(name = "itcast.topic", type = ExchangeTypes.TOPIC),
            key = "#.news"
    ))
    public void listenTopicQueue2(String msg){
        System.out.println("消费者接收到topic.queue2的消息：【" + msg + "】");
    }
}

编写测试代码：

package com.demo.mq;

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

import java.util.HashMap;
import java.util.Map;


@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringAmqpTest {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Test
    public void testSendMessage2TopicExchange() {
        String exchange = "itcast.topic";
        String message = "今天天气不错";
        rabbitTemplate.convertAndSend(exchange, "china.weather", message);
    }
}

测试结果：

更改代码，将bindingKey更改为china.news

@Test
    public void testSendMessage2TopicExchange() {
        String exchange = "itcast.topic";
        String message = "Test news";
        rabbitTemplate.convertAndSend(exchange, "china.news", message);
    }

测试结果：

SpringAmqp之常用模型Demo

大数据系统相关栏目本月热门文章