RabbitMQ（不完整版）-Toy模板网

这篇具有很好参考价值的文章主要介绍了RabbitMQ（不完整版）。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

RabbitMQ

入门

1.什么是MQ

RabbitMQ就是一个消息中间件

2.为什么要用MQ

流量消峰

假设一个订单系统最多被访问1万次，当用户访问超过1万次就会导致订单系统宕机，此时在用户和订单系统之间添加MQ中间件，此时用户先访问MQ，然后再去访问订单系统，因此达到消峰的能力。MQ此时对访问的用户进行排队，缺点也很明显，排队导致浪费时间，访问的速度就会下降，好处就是订单系统不会宕机。

应用解耦

一电商应用为例，应用中有订单系统、支付系统、库存系统、物流系统。用户创建订单后，如果耦合调用支付系统、库存系统、物流系统，任何一个子系统出现故障，都会造成下单异常。此时转为消息队列的方式后（在原本基础上订单系统和其他子系统之间添加MQ），当支付系统、库存系统、物流系统，任何一个子系统出现故障，队列会监督某个出现故障的子系统，直至完成结束。所以不会出现任何一个子系统出现故障影响整个系统的运转。

异步处理

有些服务调用是异步的，假设A调用B，B需要很长时间执行，但是A需要知道B什么时候可以执行完。此时有两种方式，第一种：A每过一段时间去询问B是否已经执行完。第二种：A调用B（调用完成之后A可以做其他事情），B执行完之后将消息发送给MQ，MQ在将消息发送给A表示B已经执行完成，A就收到了B执行完成的结果。

3.MQ的分类

ActiveMQ

优点：可用性高，很难丢失消息，丢失消息概率低

缺点：官方效率在5.X版本后维护越来越少，高吞吐量场景很少使用

Kafka

大数据采用、大型公司建议使用，如果有日志采集功能，首选Kafka

优点：吞吐量高、分布式，一个数据有多个副本，少数机器宕机后不会导致数据丢失

缺点：kafka 单机超过64个队列/分区，Load（加载）会发生明显的CPU飙高，队列越多，load越高，发送消息响应时间边长、消息失败不支持重试、支持消息顺序，但是一台机器宕机后就护发生消息乱序，社区更新慢。

RocketMQ

Java语言实现、天生为金融互联网领域而生，对于可靠性要求很高的场景，尤其是订单扣款，业务消峰等

优点：可用性高、分布式架构、消息做到0丢失

缺点：支持客户端语言不多，目前是Java和C++、某些系统迁移需要改动大量代码

RabbitMQ

2007年发布是一个在AMQP（高级消息队列协议）基础上完成，当前最主流的消息中间件之一。

优点：性能号、吞吐量达到万级，MQ功能比较完备、支持多种语言、支持AJAX文档齐全

缺点：商业版需要收费，学习成本较高

4.RabbitMQ

RabbitMQ的概念

RabbitMQ 是一个消息中间件：它接收/转发消息（还能存储消息）。相当于一个快递站，当你要发送一个消息时，你把你的消息放到RabbitMQ里面，然后RabbitMQ再把消息发送出去。

四大核心概念

i.生产者 Producer

谁发送消息谁就是生产者

ii.交换机 Exchange

MQ中包含交换机在队列

一个交换机对应多个队列（1：n关系）

交换机—绑定关系—>队列1

交换机—绑定关系—>队列2

交换机—绑定关系—>队列3

iii.队列 Queue

一个队列对应一个消费者（1：1关系）

iv.消费者 Consumer

谁接收到消息谁就是消费者

核心部分

六大模式

1）简单模式 “Hello world！”

生产者—生产消息—>队列—接收消息—>消费者

生产者代码

/**
 * 生产者：发消息
 */
public class Producer {
    //队列名称
    public static final String QUEUE_NAME = "hello";

    //发消息
    public static void main(String[] args) throws Exception {
        //创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        //设置工厂Ip 连接RabbitMQ队列
        factory.setHost("192.168.40.128");
        //设置用户名
        factory.setUsername("admin");
        //设置密码
        factory.setPassword("1234");
        //创建连接
        Connection connection = factory.newConnection();




        //创建信道（发送消息）
        Channel channel = connection.createChannel();
        /**
         * 生成一个队列
         * 参数1：队列名称
         * 参数2：队列中的消息是否持久化
         * 默认情况:消息存储在内存中.即:默认不支持持久化（持久化：保存到磁盘上）
         * 参数3：该队列是否供一个消费者进行消费  是否进行消费的共享
         * true：多个消费者可以消费   false：只能一个消费者消费
         * 参数4：是否自动删除
         * 最后一个消费者断开连接之后，该队列是否自动删除
         * true：自动删除
         * 参数5：其他参数
         *
         */
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);




        //发消息
        String message = "hello world";
        /**
         * 发送一个消息
         * 参数1：发送到哪个交换机
         * 参数2：路由的key值是哪个   本次是队列名称
         * 参数3：其他参数信息
         * 参数4：发送消息的消息体
         */
        channel.basicPublish("",QUEUE_NAME,null,message.getBytes());
        System.out.println("消息发送完毕");

    }

}

消费者代码

/**
 * 消费者：接收消息（消费消息）
 */
public class Consumer {
    public static final String QUEUE_NAME = "hello";

    public static void main(String[] args) throws Exception {
        //创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        //设置工厂Ip
        factory.setHost("192.168.40.128");
        //设置用户名
        factory.setUsername("admin");
        //设置密码
        factory.setPassword("1234");
        //创建连接
        Connection connection = factory.newConnection();

        //创建信道（接收消息）
        Channel channel = connection.createChannel();

        //声明  接收消息
        DeliverCallback deliverCallback = (consumerTag, message) -> {
            System.out.println(new String(message.getBody()));
        };

        //取消消息的回调
        CancelCallback cancelCallback = consumerTag -> {
            System.out.println("消息接收中断");
        };


        /**
         * 消费者接收消息
         * 参数1：接收哪个队列
         * 参数2：接收成功后是否要自动应答  true：自动        false：手动
         * 参数3：消费者未成功接收的回调
         * 参数4：消费者取消接收的回调
         */
        channel.basicConsume(QUEUE_NAME, true, deliverCallback, cancelCallback);
    }
}

2）工作模式 Work queues

不能重复处理同一个消息。即：一个消息最多被处理一次

生产者发送大量消息到队列中，队列发送给多个工作线程来处理消息

每个工作线程之间是竞争的关系

轮训分发消息

按周期进行

创建通用工具类

public class RabbitMqUtils {
    public static Channel getChannel() throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("192.168.40.128");
        factory.setUsername("admin");
        factory.setPassword("1234");
        Connection connection = factory.newConnection();
        return connection.createChannel();
    }
}

生产者：

public class Producer01 {
    public static final String QUEUE_NAME = "hello";

    //发送大量消息
    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMqUtils.getChannel();
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        //从控制台输入
        Scanner sc = new Scanner(System.in);
        while (sc.hasNext()) {
            String message = sc.next();
            channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
            System.out.println("发送消息完成:" + message);
        }
    }
}

工作线程：多个工作线程代码一致只需更换C1即可文章来源地址https://www.toymoban.com/news/detail-819061.html

/**
 * 工作线程 == 消费者
 */
public class Worker01 {
    //队列名称
    public static final String QUEUE_NAME = "hello";

    //接收消息
    public static void main(String[] args) throws Exception {

        Channel channel = RabbitMqUtils.getChannel();
        //消息的接收
        DeliverCallback deliverCallback = (consumerTag, message) -> {
            System.out.println("接收到的消息" + new String(message.getBody()));
        };

        //消息接收被取消时，执行下面的内容
        CancelCallback cancelCallback = consumerTag -> {
            System.out.println(consumerTag + "消费者取消消费者接口回调逻辑");
        };

        System.out.println("C1等待接收消息......");
        //消息的接收
        channel.basicConsume(QUEUE_NAME, true, deliverCallback, cancelCallback);
    }
}