RabbitMQ

RabbitMQ学习笔记：消息队列-RabbitMQ篇章- 专栏 -KuangStudy

RabbitMQ使用基于TCP/IP协议的AMQP协议 RabbitMQ还支持MQTT协议，MQTT协议轻量，结构简单，传输快，但是默认关闭，需要自己打开。

安装RabbitMQ

RabbitMQ需要使用erlang语言环境，首先需要安装erlang。

安装文章：https://www.cnblogs.com/fengyumeng/p/11133924.html

在HIK虚拟机cnetOS7中安装的账户 liuyulin10/admin

RabbitMQ角色

1. none：无法访问management plugin

2. management：只能查看自己的信息

3. policymaker：能查看和创建和删除自己的virtual hosts的统计信息，包括其他用户在这个节点virtual hosts的活动信息

4. monitoring：可以看到所有的节点，但是不能操作

5. administrator：最高权限

AMQP

生产过程：建立连接，开启通道，发送消息，释放资源

消费过程：建立连接，开启通道，准备接收消息，broker推送消息，发送确认，释放资源

RabbitMQ的核心组成

Exchange：交换机，接受消息，根据路由键发送消息到绑定的队列。(==不具备消息存储的能力==)

Bindings：Exchange和Queue之间的虚拟连接，binding中可以保护多个routing key.

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

RabbitMQ工作模式

简单模式-Simple模式

生产者

package com.liuyulin10.rabbitmq.simple;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.MessageProperties;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class Producer {

    public static void main(String[] args) {

        //所有的中间件技术都是基于tcp/ip协议的 rabbitmq遵循amqp协议
        //创建连接工程
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("10.13.99.150");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("liuyulin10");
        connectionFactory.setPassword("admin");
        connectionFactory.setVirtualHost("/");

        Connection connection = null;
        Channel channel = null;
        try {
            //创建连接connection
            connection = connectionFactory.newConnection("producer");
            //通过连接获取通道channel
            channel = connection.createChannel();
            //声明队列
            String queueName = "queue1";
            //参数1：队列的名字 2：是否要持久化 3：排他性（是否独占） 4：是否自动删除 5：附加参数
            channel.queueDeclare(queueName, true, false, false, null);
            //准备消息内容
            String msg = "hello,liuyulin10";
            channel.basicPublish("", queueName, null, msg.getBytes());

        } catch (IOException e) {
            throw new RuntimeException(e);
        } catch (TimeoutException e) {
            throw new RuntimeException(e);
        } finally {
            //关闭通道
            if(channel!=null&& channel.isOpen()){
                try {
                    channel.close();
                } catch (IOException e) {
                    throw new RuntimeException(e);
                } catch (TimeoutException e) {
                    throw new RuntimeException(e);
                }
            }
            //关闭连接
            if (connection != null && connection.isOpen()){
                try {
                    connection.close();
                } catch (IOException e) {
                    throw new RuntimeException(e);
                }
            }
        }

    }
}

消费者

package com.liuyulin10.rabbitmq.simple;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class Consumer {

    public static void main(String[] args) {

        //创建连接工程
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("10.13.99.150");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("liuyulin10");
        connectionFactory.setPassword("admin");
        connectionFactory.setVirtualHost("/");

        Connection connection = null;
        Channel channel = null;

        try {
            connection = connectionFactory.newConnection("consumer");
            channel = connection.createChannel();

            channel.basicConsume("queue1", true, new DeliverCallback() {
                @Override
                public void handle(String s, Delivery delivery) throws IOException {
                    System.out.println("收到消息" + new String(delivery.getBody(), "UTF-8"));
                }
            }, new CancelCallback() {
                @Override
                public void handle(String s) throws IOException {
                    System.out.println("接收消息失败");
                }
            });

        } catch (IOException e) {
            throw new RuntimeException(e);
        } catch (TimeoutException e) {
            throw new RuntimeException(e);
        } finally {

            //关闭通道
            if (channel!=null && channel.isOpen()){
                try {
                    channel.close();
                } catch (IOException e) {
                    throw new RuntimeException(e);
                } catch (TimeoutException e) {
                    throw new RuntimeException(e);
                }
            }
            //关闭连接
            if(connection != null && connection.isOpen()){
                try {
                    connection.close();
                } catch (IOException e) {
                    throw new RuntimeException(e);
                }
            }
        }
    }
}

发布与订阅模式-fanout模式

订阅了之后都会收到消息

spring-boot绑定生产者和队列：

@Configuration
public class RabbitMQConfiguration {

    public static final String exchangerName = "faout_order_exchanger";
    public static final String orderQueueName = "order.fanout.queue";
    public static final String durableQueueName = "durable.fanout.queue";
    public static final String driverQueueName = "driver.fanout.queue";

    //声明交换机
    @Bean
    public FanoutExchange fanoutExchanger(){
        return new FanoutExchange(exchangerName,true,false);
    }

    //订单处理队列
    @Bean
    public Queue orderQueue(){
        return new Queue(orderQueueName,true);
    }

    //持久化处理队列
    @Bean
    public Queue durableQueue(){
        return new Queue(durableQueueName,true);
    }

    //driver处理队列
    @Bean
    public Queue driverQueue(){
        return new Queue(driverQueueName,true);
    }

    //绑定关系
    @Bean
    public Binding orderBinding(){
        return BindingBuilder.bind(orderQueue()).to(fanoutExchanger());
    }

    @Bean
    public Binding durableBinding(){
        return BindingBuilder.bind(durableQueue()).to(fanoutExchanger());
    }

    @Bean
    public Binding driverBinding(){
        return BindingBuilder.bind(driverQueue()).to(fanoutExchanger());
    }
}

spring-boot生产者：

    public void makeOrder(String userId, String skuId, int num){
        String orderNumer = UUID.randomUUID().toString();
        System.out.println("用户 " + userId + ",订单编号是：" + orderNumer);
        // 发送订单信息给RabbitMQ fanout
        rabbitTemplate.convertAndSend(exchangerName, routeKey, orderNumer);
    }

生产者测试：

    @Test
    void test1() {
        orderService.makeOrder("1", "1", 12);
    }

spring-boot消费者

@Service
@RabbitListener(queues = {"order.fanout.queue","order.direct.queue"})
public class OrderConsumer {

    @RabbitHandler
    public void receiveMessage(String message){
        System.out.println("order queue ---> 接收到信息：" + message);
    }
}

路由模式-Routing模式

在发布与订阅模式之上增加了路由key

主题模式-Topic模式

支持模糊匹配路由key的路由模式

*代表至少一个字母，#代表可有可无字母

参数模式-Header模式

可以添加参数，根据参数发布消息

工作队列模式-Work模式

轮询模式

一个消费者一条

公平分发

根据消费者的能力进行分发

消息过期时间TTL

可以给队列设置消息过期时间，如果超过时间，消息会被移除或者转移到死信队列，一般情况下，过期的消息会用死信队列接收。

队列配置：

@Configuration
public class TTLConfig {
    public static String ttlExchangerName = "ttl_fanout_exchange";

    //订单处理队列
    @Bean
    public Queue ttlFanoutQueue() {
        //设置队列过期时间
        Map<String,Object> args = new HashMap<>();
        args.put("x-message-ttl",60000);
        return new Queue("order.topic.queue", true,false,false,args);
    }

    @Bean
    public FanoutExchange ttlFanoutExchange() {
        return new FanoutExchange(ttlExchangerName, true, false);
    }

    //绑定queue
    @Bean
    public Binding orderBinding() {
        return BindingBuilder.bind(ttlFanoutQueue()).to(ttlFanoutExchange());
    }
}

也可以对单独某一条消息设置过期时间。

生产者：

public void sendTTLMsg(String msg){
        System.out.println("发送ttl消息:" + msg);
        //给消息设置过期时间5s以及编码格式UTF-8
        MessagePostProcessor messagePostProcessor = new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                message.getMessageProperties().setExpiration("5000");
                message.getMessageProperties().setContentEncoding("UTF-8");
                return message;
            }
        };
        rabbitTemplate.convertAndSend(TTLConfig.ttlExchangerName,"",msg,messagePostProcessor);
    }

过期队列中过期的消息，会被移除到死信队列中；非过期队列中的过期消息，不会被写到死信队列中。

死信队列

也叫死信交换机，也有人叫死信邮箱，DLX，绑定DLX的队列就是死信队列，进入死信队列