【2】springCloud 2021 中间件基本使用方法

RabbitMQ

Broker

异步调用好处：

吞吐量提升：无需等待订阅者处理完成，响应更快速 故障隔离：服务没有直接调用，不存在级联失败问题 调用间没有阻塞，不会造成无效的资源占用 耦合度极低，每个服务都可以灵活插拔，可替换 流量削峰：不管发布事件的流量波动多大，都由 Broker 接收，订阅者可以按照自己的速度去处理事件 异步调用缺点：

架构复杂了，业务没有明显的流程线，不好管理 需要依赖于 Broker 的可靠、安全、性能 MQ消息队列 MQ，中文是消息队列（MessageQueue），字面来看就是存放消息的队列，也就是事件驱动架构中的 Broker

比较常见的 MQ 实现： ActiveMQ RabbitMQ RocketMQ Kafka 几种常见MQ的对比：

rabbitmq安装

docker pull rabbitmq:3-management

docker run \
 -e RABBITMQ_DEFAULT_USER=admin \
 -e RABBITMQ_DEFAULT_PASS=123456 \
 --name mq \
 --hostname mq1 \
 -p 15672:15672 \
 -p 5672:5672 \
 -d \
 rabbitmq:3-management

访问 ip:15672

RabbitMQ 中的一些角色

publisher：生产者 consumer：消费者 exchange：交换机，负责消息路由 queue：队列，存储消息 virtualHost：虚拟主机，隔离不同租户的 exchange、queue、消息的隔离

入门案例

Hello World 模型 官方的 HelloWorld 是基于最基础的消息队列模型来实现的，只包括三个角色： publisher：消息发布者，将消息发送到队列queue queue：消息队列，负责接受并缓存消息 consumer：订阅队列，处理队列中的消息 publisher实现

• 建立连接
• 创建 channel
• 声明队列
• 发送消息
• 关闭连接和 channel

public class PublisherTest {
    @Test
    public void testSendMessage() throws IOException, TimeoutException {
        // 1.建立连接
        ConnectionFactory factory = new ConnectionFactory();
        // 1.1.设置连接参数，分别是：主机名、端口号、vhost、用户名、密码
        factory.setHost("192.168.211.128");
        factory.setPort(5672);
        factory.setVirtualHost("/");
        factory.setUsername("admin");
        factory.setPassword("123456");
        // 1.2.建立连接
        Connection connection = factory.newConnection();
        // 2.创建通道Channel
        Channel channel = connection.createChannel();
        // 3.创建队列
        String queueName = "simple.queue";
        channel.queueDeclare(queueName, false, false, false, null);
        // 4.发送消息
        String message = "Hello RabbitMQ！";
        channel.basicPublish("", queueName, null, message.getBytes());
        System.out.println("发送消息成功：[" + message + "]");
        // 5.关闭通道和连接
        channel.close();
        connection.close();
    }
}

consumer实现

• 建立连接
• 创建 channel
• 声明队列
• 订阅消息

public class ConsumerTest {
    public static void main(String[] args) throws IOException, TimeoutException {
        // 1.建立连接
        ConnectionFactory factory = new ConnectionFactory();
        // 1.1.设置连接参数，分别是：主机名、端口号、vhost、用户名、密码
        factory.setHost("192.168.211.128");
        factory.setPort(5672);
        factory.setVirtualHost("/");
        factory.setUsername("admin");
        factory.setPassword("123456");
        // 1.2.建立连接
        Connection connection = factory.newConnection();
        // 2.创建通道Channel
        Channel channel = connection.createChannel();
        // 3.创建队列
        String queueName = "simple.queue";
        channel.queueDeclare(queueName, false, false, false, null);
        // 4.订阅消息
        channel.basicConsume(queueName, true, new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope,
                                       AMQP.BasicProperties properties, byte[] body) {
                // 5.处理消息
                String message = new String(body);
                System.out.println("接收到消息：[" + message + "]");
            }
        });
        System.out.println("等待接收消息中");
    }
}

SpringAMQP

SpringAMQP 是基于 RabbitMQ 封装的一套模板，并且还利用 SpringBoot 对其实现了自动装配，使用起来非常方便。

SpringAMQP 的官方地址：https://spring.io/projects/spring-amqp

SpringAMQP 提供了三个功能：

自动声明队列、交换机及其绑定关系 基于注解的监听器模式，异步接收消息 封装了 RabbitTemplate 工具，用于发送消息

<!--AMQP依赖，包含RabbitMQ-->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

BasicQueue

# 首先配置 MQ地址，在 publisher、consumer 服务中的 application.yml 中添加配置
spring:
  rabbitmq:
    host: 192.168.150.101 # 主机名
    port: 5672 # 端口
    virtual-host: / # 虚拟主机
    username: admin # 用户名
    password: 123456 # 密码

# Bean配置mq队列
@Configuration
public class MyRabbitMQConfig {
    @Bean
    public Queue stockReleaseStockQueue() {
        //String name, boolean durable, boolean exclusive, boolean autoDelete, Map arguments
        Queue queue = new Queue("queue1", true, false, false);
        return queue;
    }
}

# 在 consumer 服务中添加监听队列
@Component
public class RabbitMQListener {
    @RabbitListener(queues = "queue1")
    public void listenSimpleQueueMessage(String msg) throws InterruptedException {
        System.out.println("消费者接收到消息：【" + msg + "】");
    }
}

# 在 publisher 服务中添加发送消息的测试类
@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringAmqpTest {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Test
    public void testSimpleQueue() {
        // 队列名称
        String queueName = "queue1";
        // 消息
        String message = "你好啊";
        // 发送消息
        rabbitTemplate.convertAndSend(queueName, message);
    }
}

Work queues

Work queues，也被称为（Task queues），任务模型。简单来说就是让多个消费者绑定到一个队列，共同消费队列中的消息。 当消息处理比较耗时的时候，可能生产消息的速度会远远大于消息的消费速度。长此以往，消息就会堆积越来越多，无法及时处理。

此时就可以使用 work 模型，多个消费者共同处理消息处理，速度就能大大提高了。

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

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

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

发布/订阅

图中可以看到，在订阅模型中，多了一个 exchange 角色，而且过程略有变化 Publisher：生产者，也就是要发送消息的程序，但是不再发送到队列中，而是发给 exchange（交换机） Consumer：消费者，与以前一样，订阅队列，没有变化 Queue：消息队列也与以前一样，接收消息、缓存消息 Exchange：交换机，一方面，接收生产者发送的消息；另一方面，知道如何处理消息，例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作，取决于 Exchange 的类型。Exchange 有以下3种类型：

Fanout：广播，将消息交给所有绑定到交换机的队列 Direct：定向，把消息交给符合指定 routing key 的队列 Topic：通配符，把消息交给符合 routing pattern（路由模式） 的队列 Exchange（交换机）只负责转发消息，不具备存储消息的能力，因此如果没有任何队列与 Exchange 绑定，或者没有符合路由规则的队列，那么消息会丢失！

Fanout

Fanout，英文翻译是扇出，在 MQ 中我们也可以称为广播。

在广播模式下，消息发送流程是这样的：

可以有多个队列 每个队列都要绑定到 Exchange（交换机） 生产者发送的消息，只能发送到交换机，交换机来决定要发给哪个队列，生产者无法决定 交换机把消息发送给绑定过的所有队列 订阅队列的消费者都能拿到消息 接下里我们用 SpringAMQP 来简单实现 FanoutExchange

在 consumer 服务中，利用代码声明队列、交换机，并将两者绑定 在 consumer 服务中，编写两个消费者方法，分别监听 fanout.queue1 和 fanout.queue2 在 publisher 中编写测试方法，向 fanout发送消息

声明队列和交换机

Spring 提供了一个接口 Exchange，来表示所有不同类型的交换机。 在 consumer 中创建一个类，声明队列、交换机、绑定对象 Binding

@Configuration
public class FanoutConfig {

    /**
     * 声明交换机
     * @return Fanout类型交换机
     */
    @Bean
    public FanoutExchange fanoutExchange(){
        return new FanoutExchange("xn2001.fanout");
    }

    /**
     * 声明队列
     * @return Queue
     */
    @Bean
    public Queue fanoutQueue1(){
        return new Queue("fanout.queue1");
    }
    @Bean
    public Queue fanoutQueue2(){
        return new Queue("fanout.queue2");
    }

    /**
     * 绑定队列和交换机
     */
    @Bean
    public Binding bindingQueue1(FanoutExchange fanoutExchange,Queue fanoutQueue1){
        return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);
    }
    @Bean
    public Binding bindingQueue2(FanoutExchange fanoutExchange,Queue fanoutQueue2){
        return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange);
    }

}

通过这样 @Bean 的方式来申明确实比较麻烦，其实我们也是可以直接通过 @RabbitListener 注解来完成的，代码如下：

在 consumer 服务的 SpringRabbitListener 中添加三个方法，作为消费者

@RabbitListener(queues = "fanout.queue1")
public void listenFanoutQueue1(String msg) throws InterruptedException {
    System.out.println("接收到fanout.queue1的消息：【" + msg + "】" + LocalTime.now());
}

@RabbitListener(queues = "fanout.queue2")
public void listenFanoutQueue2(String msg) throws InterruptedException {
    System.err.println("接收到fanout.queue2的消息：【" + msg + "】" + LocalTime.now());
}

@RabbitListener(bindings = @QueueBinding(
    value = @Queue(value = "fanout.queue3"),
    exchange = @Exchange(value = "xn2001.fanout",type = "fanout")
))
public void listenFanoutQueue3(String msg) {
    System.out.println("接收到fanout.queue3的消息：【" + msg + "】" + LocalTime.now());
}

众所周知，JDK序列化存在下列问题：

数据体积过大 有安全漏洞 可读性差 我们推荐可以使用 JSON 来序列化

在 publisher 和 consumer 两个服务中都引入依赖

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

配置消息转换器。 在各自的启动类中添加一个 Bean 即可

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