当前位置：移动技术网 > IT编程>开发语言>Java > RabbitMQ简介

RabbitMQ简介

2020年04月21日 | 移动技术网IT编程 | 我要评论

rabbitmq简介

　　官网也讲述的清楚而且还有例子，只不过是英文的，很多人看到英文就不明白说什么了吧，即使有翻译成中文，总觉得哪里怪怪的，有些翻译并不流畅。我还是支持多看官网，官网：。下面是自己做的一点下笔记，有参考其他文档。如有什么不对的地方，希望大家能够告诉我，通过留言板，像消息队列一样，你发送消息，我接收消息。

一、消息队列（mq）

什么是消息队列：即mq，message queue。是一种应用程序对应用对应用程序的通信方式。应用程序通过读写出入队列的消息（针对应用程序的数据）来通信，而无需专用连接来连接他们。消息队列是典型的：生产者、消费者模型。生产者不断向消息队列中生产消息，消费者不断从队列中获取消息。【因为消息的生产和消费都是异步的，而且只关心消息的发送和接收，没有业务逻辑的侵入，这样就实现了生产者和消费者的解耦】
amqp和jms
- amqp：即advanced message queuing protocol，一个提供统一消息服务的应用层标准（高级消息队列协议）。【限定了数据传输的格式和方式，跨语言跨平台，和http协议类似】。是应用层协议的一个开放标准，为面向消息的中间件设计。
- jms：java messageservice，实际上指jms api。jms是sun公司早期提出的消息标准，旨在为java应用提供统一的消息操作，包括create、send、recieve等。jms已经成为java enterprise edition的一部分。
- 两者间的区别和联系：
  - jms是定义了统一的接口，来对消息操作进行统一；amqp是通过规定协议来统一数据交互的格式。
  - jms限定了必须使用java语言；amqp只是协议，不规定实现方式，因此是跨语言的。
  - jms规定了两种消息模型；而amqp的消息模型更加丰富。
常见mq产品：
- activemq：基于jms， apache旗下的
- rabbitmq：基于amqp协议，erlang（一种通用的面向并发的编程语言）语言开发，稳定性好
- rocketmq：基于jms，阿里巴巴产品，目前交由apache基金会
- kafka：分布式消息系统，高吞吐量

二、rabbitmq：

下载安装：先安装erlang，在安装rabbitmq。（这里不详细说明，如果有不是很清楚的，可以去搜索，相信很多伙伴已经分享过很多这样的博客啦）
启动步骤【window系统下的cmd命令行】：cd \rabbitmq server\rabbitmq_server-3.7.15\sbin（进入到你安装的rabbitmq目录的sbin目录） --> rabbitmq-plugins enable rabbitmq_management --> rabbitmq-server【浏览器访问窗口：】

五种消息模型：rabbitmq提供了6中消息模型，但是第6种rpc，并不是mq。其中3/4/5这三种属于订阅模式，只不过进行路由的方式不同。

基本消息模型：rabbitmq是一个消息的代理者（message broker）：它接收消息并且传递消息。你可以认为它是一个邮局，当你投递邮件到一个邮箱，你很肯定邮递员终究会将邮件递交给你的收件人。与此类似，rabbitmq可以是一个邮箱、邮局、同时还是邮递员。不同之处在于：rabbitmq不是传递纸质邮件，而是二进制的数据。

　　问题：那么rabbitmq怎么知道消息被接收了呢？（转换思维，即如何避免消息的丢失？）【答：消费者的消息确认机制acknowlege，当消费者获取消息后，会向rabbitmq发送回执ack，告知消息已经被接收。不过这种回执ack分两种情况：① 自动ack：消息一旦被接收，消费者自动发送ack；② 手动ack：消息接收后，不会发送ack，需要手动调用。很好理解，望名知意嘛，嘿嘿嘿！！！】。

public class connectionutil {

    /**
     * 建立与rabbitmq的链接
     */
    public static connection getconnection() throws ioexception, timeoutexception {
        // 定义连接工厂
        connectionfactory factory = new connectionfactory();
        // 设置服务地址
        factory.sethost("127.0.0.1");
        // 端口
        factory.setport(5672);
        // 设置账号信息，用户名、密码、vhost
        factory.setvirtualhost("/demo");
        factory.setusername("guest");
        factory.setpassword("guest");
        // 通过工厂获取连接
        connection connection = factory.newconnection();
        return connection;
    }
}

public class send {

    private final static string queue_name = "simple_queue";

    public static void main(string[] args) throws ioexception, timeoutexception {
        // 获取到连接
        connection connection = connectionutil.getconnection();
        // 从连接中创建通道，使用通道才能完成消息相关的操作
        channel channel = connection.createchannel();
        // 声明（创建）队列
        channel.queuedeclare(queue_name, false, false, false, null);
        // 消息内容
        string message = "hello word!";
        // 向指定的队列中发送消息
        channel.basicpublish("", queue_name, null, message.getbytes());

        system.out.println(" [x] sent '" + message + "'");

        // 关闭通道和连接
        channel.close();
        connection.close();
    }
}

public class recv {

    private final static string queue_name = "simple_queue";

    public static void main(string[] args) throws exception {
        // 获取连接
        connection connection = connectionutil.getconnection();
        // 创建通道
        channel channel = connection.createchannel();
        // 声明队列
        channel.queuedeclare(queue_name, false, false, false, null);
        // 定义队列的消费者
        defaultconsumer consumer = new defaultconsumer(channel) {
            // 获取消息，并且处理，这个方法类似事件监听，如果有消息时，会自动被调用
            @override
            public void handledelivery(string consumertag, envelope envelope, amqp.basicproperties properties,
                                       byte[] body) throws ioexception {
                // body 即消息体
                string msg = new string(body);
                system.out.println(" [x] received: " + msg + "!");
            }
        };
        // 监听队列，第二个参数：是否自动进行消息确认
        channel.basicconsume(queue_name, true, consumer);
    }
}

public class recv2 {

    private final static string queue_name = "simple_queue";

    public static void main(string[] args) throws exception {
        // 获取连接
        connection connection = connectionutil.getconnection();
        // 创建通道
        channel channel = connection.createchannel();
        // 声明队列
        channel.queuedeclare(queue_name, false, false, false, null);
        // 定义队列的消费者
        defaultconsumer consumer = new defaultconsumer(channel) {
            // 获取消息，并且处理，这个方法类似事件监听，如果有消息时，会自动被调用
            @override
            public void handledelivery(string consumertag, envelope envelope, amqp.basicproperties properties,
                                       byte[] body) throws ioexception {
                // body 即消息体
                string msg = new string(body);
                system.out.println(" [x] received: " + msg + "!");
                // 手动进行ack
                channel.basicack(envelope.getdeliverytag(), false);
            }
        };
        // 监听队列，第二个参数false,手动进行ack
        channel.basicconsume(queue_name, false, consumer);
    }
}

work消息模型：也称为task queue任务模型。当消息处理比较耗时的时候，可能产生消息的速度会远远大于消息的消费速度。长此以往，消息就会堆积越来越多，无法及时处理。此时就可以使用该模型：让多个消费者绑定到一个队列，共同消费队列中的消息。队列中的消息一旦消费，就会消失，因此任务是不会被重复执行的。

　　问题：又会存在疑惑，任务是如何分配的，可以和显示生活中任务的分配联想起来呦？【答案：① 默认是任务平分，一次分配完全（即公平分配）；② channel.basicqos(int num); 设置每个消费者同时只能处理num条数据（即能者多劳，耗时小的多处理些，你懂滴）】。

public class send {

    private final static string queue_name = "task_work_queue";

    public static void main(string[] args) throws exception {
        // 获取到连接
        connection connection = connectionutil.getconnection();
        // 从连接中创建通道，使用通道才能完成消息相关的操作
        final channel channel = connection.createchannel();
        // 声明（创建）队列
        channel.queuedeclare(queue_name, false, false, false, null);
        // 循环发布任务
        for (int i=0; i<50; i++) {
            // 消息内容
            string message = "task ... " + i;
            channel.basicpublish("", queue_name, null, message.getbytes());
            system.out.println(" [x] sent '" + message + "'");
            thread.sleep(i * 2);
        }
        // 关闭通道和连接
        channel.close();
        connection.close();
    }
}

public class recv {

    private final static string queue_name = "task_work_queue";

    public static void main(string[] args) throws exception {
        // 获取连接
        connection connection = connectionutil.getconnection();
        // 创建通道
        channel channel = connection.createchannel();
        // 声明队列
        channel.queuedeclare(queue_name, false, false, false, null);
        // 设置每个消费同时只能处理一条消息
        channel.basicqos(1);
        // 定义队列的消费者
        defaultconsumer consumer = new defaultconsumer(channel) {
            // 获取消息，并且处理，这个方法类似事件监听，如果有消息时，会自动被调用
            @override
            public void handledelivery(string consumertag, envelope envelope, amqp.basicproperties properties,
                                       byte[] body) throws ioexception {
                // body 即消息体
                string msg = new string(body);
                system.out.println(" [x] received: " + msg + "!");

                try {
                    // 模拟完成任务的耗时：1000ms
                    thread.sleep(1000);
                } catch (interruptedexception e) {
                    e.printstacktrace();
                }
                channel.basicack(envelope.getdeliverytag(), false);
            }
        };
        // 监听队列，第二个参数：是否自动进行消息确认
        channel.basicconsume(queue_name, false, consumer);
    }
}

/*
* 对比上个消费者：耗时小，完成任务多些
*/
public class recv2 {

    private final static string queue_name = "task_work_queue";

    public static void main(string[] args) throws exception {
        // 获取连接
        connection connection = connectionutil.getconnection();
        // 创建通道
        final channel channel = connection.createchannel();
        // 声明队列
        channel.queuedeclare(queue_name, false, false, false, null);
        // 设置每个消费同时只能处理一条消息
        channel.basicqos(1);
        // 定义队列的消费者
        defaultconsumer consumer = new defaultconsumer(channel) {
            // 获取消息，并且处理，这个方法类似事件监听，如果有消息时，会自动被调用
            @override
            public void handledelivery(string consumertag, envelope envelope, amqp.basicproperties properties,
                                       byte[] body) throws ioexception {
                // body 即消息体
                string msg = new string(body);
                system.out.println(" [x] received: " + msg + "!");
                // 手动进行ack
                channel.basicack(envelope.getdeliverytag(), false);
            }
        };
        // 监听队列，第二个参数false,手动进行ack
        channel.basicconsume(queue_name, false, consumer);
    }
}

订阅模型分类：

订阅模型 - fanout：广播。一条消息，会被所有订阅的队列消费。

public class send {

    private final static string exchange_name = "fanout_exchange_test";

    public static void main(string[] args) throws exception {
        // 获取连接
        connection connection = connectionutil.getconnection();
        // 获取通道
        channel channel = connection.createchannel();
        // 声明exchange，指定类型为fanout
        channel.exchangedeclare(exchange_name, builtinexchangetype.fanout);
        // 消息内容
        string message = "hello everyone";
        // 发布消息到exchange
        channel.basicpublish(exchange_name, "", null, message.getbytes());
        system.out.println(" [生产者] sent '" + message + "'");
        channel.close();
        connection.close();
    }
}

public class recv {

    private final static string queue_name = "fanout_exchange_queue_1";
    private final static string exchange_name = "fanout_exchange_test";

    public static void main(string[] args) throws ioexception, timeoutexception {
        // 获取到链接
        connection connection = connectionutil.getconnection();
        // 获取通道
        channel channel = connection.createchannel();
        // 声明队列
        channel.queuedeclare(queue_name, false, false, false, null);
        // 绑定队列到交换机
        channel.queuebind(queue_name, exchange_name, "");
        // 定义队列的消费者
        defaultconsumer consumer = new defaultconsumer(channel) {
            // 获取消息，并且处理，这个方法类似事件监听，如果有消息时，会自动调用
            @override
            public void handledelivery(string consumertag, envelope envelope, amqp.basicproperties properties, byte[] body) throws ioexception {
                string msg = new string(body);
                system.out.println(" [消费者1] received: " + msg + "!");
            }
        };
        // 监听队列，自动返回完成
        channel.basicconsume(queue_name, true, consumer);
    }
}

public class recv2 {

    private final static string queue_name = "fanout_exchange_queue_2";
    private final static string exchange_name = "fanout_exchange_test";

    public static void main(string[] args) throws ioexception, timeoutexception {
        // 获取到链接
        connection connection = connectionutil.getconnection();
        // 获取通道
        channel channel = connection.createchannel();
        // 声明队列
        channel.queuedeclare(queue_name, false, false, false, null);
        // 绑定队列到交换机
        channel.queuebind(queue_name, exchange_name, "");
        // 定义队列的消费者
        defaultconsumer consumer = new defaultconsumer(channel) {
            @override
            public void handledelivery(string consumertag, envelope envelope, amqp.basicproperties properties, byte[] body) throws ioexception {
                string msg = new string(body);
                system.out.println(" [消费者2] received: " + msg + "!");
            }
        };
        // 监听队列，自动返回完成
        channel.basicconsume(queue_name, true, consumer);
    }
}

订阅模型 - direct：不同的消息被不同的队列消费。在direct模型下：

　　队列与交换机的绑定，不能是任意绑定，而是要指定至少一个 routingkey（路由key）；
　　消息的发送方向 exchange 发送消息时，也必须指定消息的 routingkey；
　　exchange 不再把消息交给每一个绑定的队列，而是根据消息的 routingkey 进行判断，只有队列的routingkey与消息的routingkey完全一致，才会接收到消息。

public class send {

    private final static string exchange_name = "direct_exchange_test";

    public static void main(string[] args) throws exception {
        // 获取连接
        connection connection = connectionutil.getconnection();
        // 获取通道
        channel channel = connection.createchannel();
        // 声明exchange，指定类型为direct
        channel.exchangedeclare(exchange_name, builtinexchangetype.direct);
        // 消息内容
        string message = "商品增加了，id = 1002";
        // 发布消息到exchange,并且指定routing key为：delete，代表删除商品
        channel.basicpublish(exchange_name, "insert", null, message.getbytes());
        system.out.println(" [商品服务] sent '" + message + "'");
        channel.close();
        connection.close();
    }
}

public class recv {

    private final static string queue_name = "direct_exchange_queue_1";
    private final static string exchange_name = "direct_exchange_test";

    public static void main(string[] args) throws ioexception, timeoutexception {
        // 获取到链接
        connection connection = connectionutil.getconnection();
        // 获取通道
        channel channel = connection.createchannel();
        // 声明队列
        channel.queuedeclare(queue_name, false, false, false, null);
        // 绑定队列到交换机,同时指定需要订阅的routing key。假设此处需要update和delete消息
        channel.queuebind(queue_name, exchange_name, "update");
        channel.queuebind(queue_name, exchange_name, "delete");
        // 定义队列的消费者
        defaultconsumer consumer = new defaultconsumer(channel) {
            // 获取消息，并且处理，这个方法类似事件监听，如果有消息时，会自动调用
            @override
            public void handledelivery(string consumertag, envelope envelope, amqp.basicproperties properties, byte[] body) throws ioexception {
                string msg = new string(body);
                system.out.println(" [消费者1] received: " + msg + "!");
            }
        };
        // 监听队列，自动返回完成
        channel.basicconsume(queue_name, true, consumer);
    }
}

public class recv2 {

    private final static string queue_name = "direct_exchange_queue_2";
    private final static string exchange_name = "direct_exchange_test";

    public static void main(string[] args) throws ioexception, timeoutexception {
        // 获取到链接
        connection connection = connectionutil.getconnection();
        // 获取通道
        channel channel = connection.createchannel();
        // 声明队列
        channel.queuedeclare(queue_name, false, false, false, null);
        // 绑定队列到交换机,同时指定需要订阅的routing key。假设此处需要update和delete消息
        channel.queuebind(queue_name, exchange_name, "update");
        channel.queuebind(queue_name, exchange_name, "delete");
        channel.queuebind(queue_name, exchange_name, "insert");
        // 定义队列的消费者
        defaultconsumer consumer = new defaultconsumer(channel) {
            // 获取消息，并且处理，这个方法类似事件监听，如果有消息时，会自动调用
            @override
            public void handledelivery(string consumertag, envelope envelope, amqp.basicproperties properties, byte[] body) throws ioexception {
                string msg = new string(body);
                system.out.println(" [消费者2] received: " + msg + "!");
            }
        };
        // 监听队列，自动返回完成
        channel.basicconsume(queue_name, true, consumer);
    }
}

订阅模型 - topic：可以根据routingkey把消息路由到不同的队列，topic类型exchange可以让队列在绑定routingkey的时候使用通配符。

routingkey 一般都是有一个或多个单词组成，多个单词之间以“ . ”分割，例：item.insert。
通配符规则：
- #：匹配一个或多个词
- *：匹配不多不少恰好一个词

public class send {

    private final static string exchange_name = "topic_durable_exchange_test";

    public static void main(string[] args) throws exception {
        // 获取连接
        connection connection = connectionutil.getconnection();
        // 获取通道
        channel channel = connection.createchannel();

        // 开启生产者确认
//        channel.confirmselect();

        // 声明exchange，指定类型为topic, 并且设置durable为true，持久化
        channel.exchangedeclare(exchange_name, builtinexchangetype.topic, true);
        // 消息内容
        string message = "商品新增了，id = 1002";
        // 发布消息到exchange,并且指定routing key，消息持久化
        channel.basicpublish(exchange_name, "item.insert", messageproperties.persistent_text_plain, message.getbytes());
        system.out.println(" [商品服务] sent '" + message + "'");

        // 等待rabbitmq的确认消息,true为确认收到，false为发出有误
//        channel.waitforconfirms();

        channel.close();
        connection.close();
    }
}

public class recv {

    private final static string queue_name = "topic_durable_exchange_queue_1";
    private final static string exchange_name = "topic_durable_exchange_test";

    public static void main(string[] args) throws ioexception, timeoutexception {
        // 获取到链接
        connection connection = connectionutil.getconnection();
        // 获取通道
        channel channel = connection.createchannel();
        // 声明队列, 第二个参数：true代表声明为持久化
        channel.queuedeclare(queue_name, true, false, false, null);
        // 绑定队列到交换机,同时指定需要订阅的routing key。假设此处需要update和delete消息
        channel.queuebind(queue_name, exchange_name, "item.update");
        channel.queuebind(queue_name, exchange_name, "item.delete");
        // 定义队列的消费者
        defaultconsumer consumer = new defaultconsumer(channel) {
            // 获取消息，并且处理，这个方法类似事件监听，如果有消息时，会自动调用
            @override
            public void handledelivery(string consumertag, envelope envelope, amqp.basicproperties properties, byte[] body) throws ioexception {
                string msg = new string(body);
                system.out.println(" [消费者1] received: " + msg + "!");
            }
        };
        // 监听队列，自动返回完成
        channel.basicconsume(queue_name, true, consumer);
    }
}

public class recv2 {

    private final static string queue_name = "topic_durable_exchange_queue_2";
    private final static string exchange_name = "topic_durable_exchange_test";

    public static void main(string[] args) throws ioexception, timeoutexception {
        // 获取到链接
        connection connection = connectionutil.getconnection();
        // 获取通道
        channel channel = connection.createchannel();
        // 声明队列
        channel.queuedeclare(queue_name, true, false, false, null);
        // 绑定队列到交换机,同时指定需要订阅的routing key。假设此处需要update和delete消息
        channel.queuebind(queue_name, exchange_name, "item.*");
        // 定义队列的消费者
        defaultconsumer consumer = new defaultconsumer(channel) {
            // 获取消息，并且处理，这个方法类似事件监听，如果有消息时，会自动调用
            @override
            public void handledelivery(string consumertag, envelope envelope, amqp.basicproperties properties, byte[] body) throws ioexception {
                string msg = new string(body);
                system.out.println(" [消费者2] received: " + msg + "!");
            }
        };
        // 监听队列，自动返回完成
        channel.basicconsume(queue_name, true, consumer);
    }
}

当然啦，国内这么hot的rabbitmq自然也是有集成到了springboot中滴，开心。

maven坐标依赖：

<parent>
        <groupid>org.springframework.boot</groupid>
        <artifactid>spring-boot-starter-parent</artifactid>
        <version>2.0.4.release</version>
</parent>

<dependencies>
        <dependency>
            <groupid>org.springframework.boot</groupid>
            <artifactid>spring-boot-starter-amqp</artifactid>
        </dependency>
</dependencies>

application.yml：

spring:
  rabbitmq:
    host: 127.0.0.1
    username: guest
    password: guest
    virtual-host: /demo

@component
public class listener {

    @rabbitlistener(bindings = @queuebinding(
            value = @queue(value = "spring.test.queue", durable = "true"),
            exchange = @exchange(
                    value = "spring.test.exchange",
                    type = exchangetypes.topic),
            key = {"#.#"}))
    public void listen(string msg) {
        system.out.println("接收到的消息: " + msg);
    }
}

持久化：要将消息持久化，前提是：队列、exchange都持久化。
- 交换机持久化：channel.exchangedeclare(exchange_name, builtinexchangetype.topic, true);　　// 参数三：设置durable为true。
- 队列持久化：channel.queuedeclare(queue_name, true, false, false, null);　　// 参数二：设置为true，表示设置队列持久化。
- 消息持久化：channel.basicpublish(exchange_name, "item.insert", messageproperties.persistent_text_plain, message.getbytes());
解决消息丢失？
- ack（消费者确认，由消费者向mq发送，防止消息丢失于消费者）
- 持久化（防止rabbitmq把消息丢失）
- 生产者确认机制publisher confirm（由mq向生产者发送，有些mq包含，有些不包含，比如：activemq不包含该机制，rabbitmq包含该机制）
- 发送消息前，将消息持久化到数据库，并记录消息状态（可靠消息服务）
思考问题（这个问题就留给你们思考啦？冲啊）：如何保证消息发送的重复性，如何保证接口的幂等性（同一接口被重复执行，其结果一致）？【提示：加标识消息的重发要谨慎】

您可能感兴趣的文章:

如对本文有疑问，点击进行留言回复！！

关于msyql事务隔离你要知道

什么是事务？事务是数据库管理系统执行过程中的一个逻辑单位，由一个有限的数据库操作序列构成。数据库事务通常包含了一个序列的对数据库的读/写操作。包含有以下两个目的... [阅读全文]
JavaFX实现UI美观效果代码实例

相对于swing来说，javafx在ui上改善了很多，不仅可以通过fxml来排版布局界面，同时也可以通过css样式表来美化ui。其实在开发javafx应用的时候... [阅读全文]
使用java实现网络爬虫

接着上面一篇，这一篇目的就是在于网络爬虫的实现，对数据的获取，以便分析。----->爬虫实现原理网络爬虫基本技术处理网络爬虫是数据采集的一种方法，实际项目... [阅读全文]
JavaFX桌面应用未响应问题解决方案

日常使用软件的过程中，偶尔会遇到软件突然卡住，再点击几次就变成“未响应”的情况。在javafx应用中同样也会出现这种情况，在开发过程中应该尽量避免这种情况的出现... [阅读全文]
java调用回调机制详解

调用和回调机制在一个应用系统中, 无论使用何种语言开发, 必然存在模块之间的调用, 调用的方式分为几种:1.同步调用同步调用是最基本并且最简单的一种调用方式, ... [阅读全文]
Springboot项目因为kackson版本问题启动报错解决方案

问题现象org.springframework.context.applicationcontextexception: unable to start emb... [阅读全文]
idea中database不显示问题的解决

【问题】一般情况下，database会显示在idea的最右边，就像这个样子：一无所有。。。【理想界面】：【解决方法】方法一：1）点击view 2)点击tool ... [阅读全文]
Java多线程下的其他组件之CyclicBarrier、Callable、Future和FutureTask详解

cyclicbarrier 接着讲多线程下的其他组件，第一个要讲的就是cyclicbarrier。cyclicbarrier从字面理解是指循环屏障，它可以协同多... [阅读全文]
IDEA POJO开发神器之Groovy的使用详解

暂时只对 mysql进行了测试项目使用 lombok mybatis-plus一：使用步骤首先在项目右侧找到 database 如图没有请参考 2.点开之后进... [阅读全文]
idea+git合并分支解决冲突及详解步骤

git分支详解参考：分支管理组成1.1、master主干在版本管理中，代码库应该仅有一个主干。此主干是和当前生产保持一致的，是可用的、稳定的可直接发布的版本，不... [阅读全文]