安装RabbitMQ（Centos6）（入门使用教程）（消息丢失的解决方案）以及Spring AMQP的使用

86 阅读 0 评论 57 点赞

我是靠谱客的博主爱撒娇项链，最近开发中收集的这篇文章主要介绍安装RabbitMQ（Centos6）（入门使用教程）（消息丢失的解决方案）以及Spring AMQP的使用，觉得挺不错的，现在分享给大家，希望可以做个参考。

概述

一、RabbitMQ的安装和配置以及用户管理

1、搜索与商品服务的问题

目前我们已经完成了商品详情和搜索系统的开发。我们思考一下，是否存在问题？

商品的原始数据保存在数据库中，增删改查都在数据库中完成。
搜索服务数据来源是索引库，如果数据库商品发生变化，索引库数据不能及时更新。
商品详情做了页面静态化，静态页面数据也不会随着数据库商品发生变化。

如果我们在后台修改了商品的价格，搜索页面和商品详情页显示的依然是旧的价格，这样显然不对。该如何解决？

这里有两种解决方案：

方案1：每当后台对商品做增删改操作，同时要修改索引库数据及静态页面
方案2：搜索服务和商品页面服务对外提供操作接口，后台在商品增删改后，调用接口

以上两种方式都有同一个严重问题：就是代码耦合，后台服务中需要嵌入搜索和商品页面服务，违背了微服务的独立原则。

所以，我们会通过另外一种方式来解决这个问题：消息队列

2、消息队列（MQ）

（1）什么是消息队列

消息队列，即MQ，Message Queue。
在这里插入图片描述

（2）AMQP和JMS

MQ是消息通信的模型，并发具体实现。现在实现MQ的有两种主流方式：AMQP、JMS。
在这里插入图片描述
两者间的区别和联系：

JMS是定义了统一的接口，来对消息操作进行统一；AMQP是通过规定协议来统一数据交互的格式
JMS限定了必须使用Java语言；AMQP只是协议，不规定实现方式，因此是跨语言的。
JMS规定了两种消息模型；而AMQP的消息模型更加丰富

（3）常见MQ产品

在这里插入图片描述

ActiveMQ：基于JMS
RabbitMQ：基于AMQP协议，erlang语言开发，稳定性好
RocketMQ：基于JMS，阿里巴巴产品，目前交由Apache基金会
Kafka：分布式消息系统，高吞吐量

（4）RabbitMQ

RabbitMQ是基于AMQP的一款消息管理系统

官网： http://www.rabbitmq.com/

在这里插入图片描述

官方教程：http://www.rabbitmq.com/getstarted.html

在这里插入图片描述

RabbitMQ基于Erlang语言开发：

在这里插入图片描述

3、下载和安装

（1）下载

官网下载地址：http://www.rabbitmq.com/download.html

在这里插入图片描述

也可以使用我提供的软件：
https://download.csdn.net/download/qq_44757034/21713768

在这里插入图片描述

（2）安装Erlang

1）在`/home/leyou/`下创建一个新目录

在这里插入图片描述

我们并没有提供Erlang安装包，直接采用yum仓库安装：在mq目录下

sudo yum install esl-erlang_17.3-1~centos~6_amd64.rpm
sudo yum install esl-erlang-compat-R14B-1.el6.noarch.rpm

在这里插入图片描述

（3）安装RabbitMQ

1）安装

sudo rpm -ivh rabbitmq-server-3.4.1-1.noarch.rpm --force --nodeps

在这里插入图片描述

2）修改配置文件

将配置文件模板复制到etc目录：

sudo cp /usr/share/doc/rabbitmq-server-3.4.1/rabbitmq.config.example /etc/rabbitmq/rabbitmq.config

通过vim命令编辑：

sudo vim /etc/rabbitmq/rabbitmq.config

修改下面内容：
注意要去掉前面的百分号和后面的逗号
在这里插入图片描述

（4）设置开机启动

输入下面命令

sudo chkconfig rabbitmq-server on

（5）启动命令

通过下面命令来控制RabbitMQ：

sudo service rabbitmq-server start
sudo service rabbitmq-server stop
sudo service rabbitmq-server restart

在这里插入图片描述

（6）开启web管理界面

RabbitMQ提供了用来管理的控制界面，十分方便，不过默认是关闭的。

我们通过命令开启web管理插件：

sudo rabbitmq-plugins enable rabbitmq_management

然后重启RabbitMQ：

sudo service rabbitmq-server restart

（7）开放端口

RabbitMQ默认使用15672端口进行web访问，我们开启防火墙端口：

iptables -A INPUT -ptcp --dport  15672  -j ACCEPT

service iptables save

然后在主机中通过地址：http://192.168.206.66:15672即可访问到管理界面

4、管理界面介绍

第一次访问需要登录，默认的账号密码为：guest/guest

在这里插入图片描述

（1）主页

在这里插入图片描述

connections：无论生产者还是消费者，都需要与RabbitMQ建立连接后才可以完成消息的生产和消费，在这里可以查看连接情况
channels：通道，建立连接后，会形成通道，消息的投递获取依赖通道。
Exchanges：交换机，用来实现消息的路由
Queues：队列，即消息队列，消息存放在队列中，等待消费，消费后被移除队列。

在这里插入图片描述
修改密码

（2）添加用户

在这里插入图片描述
上面的Tags选项，其实是指定用户的角色，可选的有以下几个：

超级管理员(administrator)

可登陆管理控制台，可查看所有的信息，并且可以对用户，策略(policy)进行操作。
监控者(monitoring)

可登陆管理控制台，同时可以查看rabbitmq节点的相关信息(进程数，内存使用情况，磁盘使用情况等)
策略制定者(policymaker)

可登陆管理控制台, 同时可以对policy进行管理。但无法查看节点的相关信息(上图红框标识的部分)。
普通管理者(management)

仅可登陆管理控制台，无法看到节点信息，也无法对策略进行管理。
其他

无法登陆管理控制台，通常就是普通的生产者和消费者。

（3）创建虚拟主机（Virtual Hosts）

为了让各个用户可以互不干扰的工作，RabbitMQ添加了虚拟主机（Virtual Hosts）的概念。其实就是一个独立的访问路径，不同用户使用不同路径，各自有自己的队列、交换机，互相不会影响。

在这里插入图片描述
创建好虚拟主机，我们还要给用户添加访问权限：

点击添加好的虚拟主机：

在这里插入图片描述
进入虚拟主机设置界面：

在这里插入图片描述

二、五种消息模型

RabbitMQ提供了6种消息模型，但是第6种其实是RPC，并不是MQ，因此不予学习。那么也就剩下5种。

但是其实3、4、5这三种都属于订阅模型，只不过进行路由的方式不同。
在这里插入图片描述

1、导入demo工程

我们通过一个demo工程来了解下RabbitMQ的工作方式：

导入工程：下载代码

https://download.csdn.net/download/qq_44757034/21718711

在这里插入图片描述

2、基本消息模型

说明

官方文档说明：

RabbitMQ是一个消息的代理者（Message Broker）：它接收消息并且传递消息。

你可以认为它是一个邮局：当你投递邮件到一个邮箱，你很肯定邮递员会终究会将邮件递交给你的收件人。与此类似，RabbitMQ 可以是一个邮箱、邮局、同时还有邮递员。

不同之处在于：RabbitMQ不是传递纸质邮件，而是二进制的数据

基本消息模型图：
在这里插入图片描述
在上图的模型中，有以下概念：

P：生产者，也就是要发送消息的程序
C：消费者：消息的接受者，会一直等待消息到来。
queue：消息队列，图中红色部分。类似一个邮箱，可以缓存消息；生产者向其中投递消息，消费者从其中取出消息。

生产者

连接工具类：
修改ip地址和对应的账号密码以及访问路径
在这里插入图片描述

public class ConnectionUtil {
    /**
     * 建立与RabbitMQ的连接
     * @return
     * @throws Exception
     */
    public static Connection getConnection() throws Exception {
        //定义连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        //设置服务地址
        factory.setHost("192.168.206.66");
        //端口
        factory.setPort(5672);
        //设置账号信息，用户名、密码、vhost
        factory.setVirtualHost("/leyou");
        factory.setUsername("leyou");
        factory.setPassword("123321");
        // 通过工程获取连接
        Connection connection = factory.newConnection();
        return connection;
    }
}

生产者发送消息：
在这里插入图片描述

package cn.itcast.rabbitmq.simple;

import cn.itcast.rabbitmq.util.ConnectionUtil;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.MessageProperties;
/**
 * 生产者
 */
public class Send {

    private final static String QUEUE_NAME = "simple_queue";

    public static void main(String[] argv) throws Exception {
        // 获取到连接
        Connection connection = ConnectionUtil.getConnection();
        // 从连接中创建通道，使用通道才能完成消息相关的操作
        Channel channel = connection.createChannel();
        // 声明（创建）队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        // 消息内容
        String message = "Hello World!";
        // 向指定的队列中发送消息
        channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
        
        System.out.println(" [x] Sent '" + message + "'");

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

在这里插入图片描述
控制台运行结果

web控制台查看消息(已创建的leyou用户登录，密码是123321)

进入队列页面，可以看到新建了一个队列：simple_queue
在这里插入图片描述
点击队列名称simple_queue，进入详情页，可以查看消息：

在控制台查看消息并不会将消息消费，所以消息还在

消费者获取消息

package cn.itcast.rabbitmq.simple;

import java.io.IOException;

import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;

import cn.itcast.rabbitmq.util.ConnectionUtil;

/**
 * 消费者
 */
public class Recv {
    private final static String QUEUE_NAME = "simple_queue";

    public static void main(String[] argv) 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, BasicProperties properties,
                    byte[] body) throws IOException {
                // body 即消息体
                String msg = new String(body);
                System.out.println(" [x] received : " + msg + "!");
            }
        };
        // 监听队列，第二个参数：是否自动进行消息确认。
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}

在这里插入图片描述

控制台

这个时候，队列中的消息就没了：

在这里插入图片描述

消费者的消息确认机制(Acknowlage)防止消息丢失

通过刚才的案例可以看出，消息一旦被消费者接收，队列中的消息就会被删除。

那么问题来了：RabbitMQ怎么知道消息被接收了呢？

这就要通过消息确认机制（Acknowlege）来实现了。当消费者获取消息后，会向RabbitMQ发送回执ACK，告知消息已经被接收。不过这种回执ACK分两种情况：

自动ACK：消息一旦被接收，消费者自动发送ACK（如果代码当中出现异常，代码没有执行完毕的情况下消息就丢失了，正常情况下代码没有执行完毕是不可能删除消息的）
手动ACK：消息接收后，不会发送ACK，需要手动调用

大家觉得哪种更好呢？

这需要看消息的重要性：

如果消息不太重要，丢失也没有影响，那么自动ACK会比较方便
如果消息非常重要，不容丢失。那么最好在消费完成后手动ACK，否则接收消息后就自动ACK，RabbitMQ就会把消息从队列中删除。如果此时消费者宕机，那么消息就丢失了。

在Recv当中测试

先运行Send
在这里插入图片描述

在Recv当中添加错误代码
在这里插入图片描述

运行Recv
在这里插入图片描述
运行报错

在这里插入图片描述

我们之前的测试都是自动ACK的，如果要手动ACK，需要改动我们的代码：

package cn.itcast.rabbitmq.simple;

import java.io.IOException;

import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;

import cn.itcast.rabbitmq.util.ConnectionUtil;

/**
 * 消费者,手动进行ACK
 */
public class Recv2 {
    private final static String QUEUE_NAME = "simple_queue";

    public static void main(String[] argv) throws Exception {
        // 获取到连接
        Connection connection = ConnectionUtil.getConnection();
        // 创建通道
        final 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, 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);
    }
}

注意到最后一行代码：

channel.basicConsume(QUEUE_NAME, false, consumer);

如果第二个参数为true，则会自动进行ACK；如果为false，则需要手动ACK。

运行测试
先通过Send类
在这里插入图片描述
浏览器当中查看

在Recv2当中写一个错误代码

运行Recv2

在这里插入图片描述
报错

消息依旧在

3、work消息模型

说明

在刚才的基本模型中，一个生产者，一个消费者，生产的消息直接被消费者消费。比较简单。

Work queues，也被称为（Task queues），任务模型。

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

在这里插入图片描述
角色：

P：生产者：任务的发布者
C1：消费者，领取任务并且完成任务，假设完成速度较慢
C2：消费者2：领取任务并完成任务，假设完成速度快

案例一、

消费者1

在这里插入图片描述

消费者2

在这里插入图片描述
与消费者1基本类似，就是没有设置消费耗时时间。

这里是模拟有些消费者快，有些比较慢。

接下来，两个消费者一同启动，然后发送50条消息：
在这里插入图片描述

可以发现，两个消费者各自消费了25条消息，而且各不相同，这就实现了任务的分发。

案例二、能者多劳

在两个消费放当中都添加

 // 设置每个消费者同时只能处理一条消息
        channel.basicQos(1);

刚才的实现有问题吗？

消费者1比消费者2的效率要低，一次任务的耗时较长
然而两人最终消费的消息数量是一样的
消费者2大量时间处于空闲状态，消费者1一直忙碌

现在的状态属于是把任务平均分配，正确的做法应该是消费越快的人，消费的越多。

怎么实现呢？

我们可以修改设置，让消费者同一时间只接收一条消息，这样处理完成之前，就不会接收更多消息，就可以让处理快的人，接收更多消息：

生产者

生产者与案例1中的几乎一样：

在这里插入图片描述

package cn.itcast.rabbitmq.work;

import cn.itcast.rabbitmq.util.ConnectionUtil;

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

// 生产者
public class Send {
    private final static String QUEUE_NAME = "test_work_queue";

    public static void main(String[] argv) throws Exception {
        // 获取到连接
        Connection connection = ConnectionUtil.getConnection();
        // 获取通道
        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();
    }
}

不过这里我们是循环发送50条消息。

消费者1

在这里插入图片描述

package cn.itcast.rabbitmq.work;

import java.io.IOException;

import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;

import cn.itcast.rabbitmq.util.ConnectionUtil;

// 消费者1
public class Recv {
    private final static String QUEUE_NAME = "test_work_queue";

    public static void main(String[] argv) 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, BasicProperties properties,
                    byte[] body) throws IOException {
                // body 即消息体
                String msg = new String(body);
                System.out.println(" [消费者1] received : " + msg + "!");
                try {
                    // 模拟完成任务的耗时：1000ms
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                }
                // 手动ACK
                channel.basicAck(envelope.getDeliveryTag(), false);
            }
        };
        // 监听队列。
        channel.basicConsume(QUEUE_NAME, false, consumer);
    }
}

消费者2

在这里插入图片描述

package cn.itcast.rabbitmq.work;

import java.io.IOException;

import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;

import cn.itcast.rabbitmq.util.ConnectionUtil;

//消费者2
public class Recv2 {
    private final static String QUEUE_NAME = "test_work_queue";

    public static void main(String[] argv) 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, BasicProperties properties,
                    byte[] body) throws IOException {
                // body 即消息体
                String msg = new String(body);
                System.out.println(" [消费者2] received : " + msg + "!");
                // 手动ACK
                channel.basicAck(envelope.getDeliveryTag(), false);
            }
        };
        // 监听队列。
        channel.basicConsume(QUEUE_NAME, false, consumer);
    }
}

运行测试

同时启动Recv和Recv2

在这里插入图片描述

然后启动Send发送50条信息
在这里插入图片描述

消费者1 和消费者2都在拼命消费
在这里插入图片描述

4、订阅模型分类

订阅模型示意图：
在这里插入图片描述
前面2个案例中，只有3个角色：

P：生产者，也就是要发送消息的程序
C：消费者：消息的接受者，会一直等待消息到来。
queue：消息队列，图中红色部分。类似一个邮箱，可以缓存消息；生产者向其中投递消息，消费者从其中取出消息。

而在订阅模型中，多了一个exchange角色，而且过程略有变化：

P：生产者，也就是要发送消息的程序，但是不再发送到队列中，而是发给X（交换机）
C：消费者，消息的接受者，会一直等待消息到来。
Queue：消息队列，接收消息、缓存消息。
Exchange：交换机，图中的X。一方面，接收生产者发送的消息。另一方面，知道如何处理消息，例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作，取决于Exchange的类型。Exchange有以下3种类型：
- Fanout：广播，将消息交给所有绑定到交换机的队列
- Direct：定向，把消息交给符合指定routing key 的队列
- Topic：通配符，把消息交给符合routing pattern（路由模式）的队列

Exchange（交换机）只负责转发消息，不具备存储消息的能力，因此如果没有任何队列与Exchange绑定，或者没有符合路由规则的队列，那么消息会丢失！

5、订阅模型-Fanout

Fanout，也称为广播。

流程说明

流程图：
在这里插入图片描述
在广播模式下，消息发送流程是这样的：

1）可以有多个消费者
2）每个消费者有自己的queue（队列）
3）每个队列都要绑定到Exchange（交换机）
4） 生产者发送的消息，只能发送到交换机，交换机来决定要发给哪个队列，生产者无法决定。
5）交换机把消息发送给绑定过的所有队列
6）队列的消费者都能拿到消息。实现一条消息被多个消费者消费

在这里插入图片描述

package cn.itcast.rabbitmq.fanout;

import cn.itcast.rabbitmq.util.ConnectionUtil;

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;

public class Send {

    private final static String EXCHANGE_NAME = "fanout_exchange_test";

    public static void main(String[] argv) 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();
    }
}

运行上述代码
在这里插入图片描述
交换机创建成功

消费者1

在这里插入图片描述

package cn.itcast.rabbitmq.fanout;

import java.io.IOException;

import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;

import cn.itcast.rabbitmq.util.ConnectionUtil;

//消费者1
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[] argv) throws Exception {
        // 获取到连接
        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, BasicProperties properties,
                    byte[] body) throws IOException {
                // body 即消息体
                String msg = new String(body);
                System.out.println(" [消费者1] received : " + msg + "!");
            }
        };
        // 监听队列，自动返回完成
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}

在这里插入图片描述

package cn.itcast.rabbitmq.fanout;

import java.io.IOException;

import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;

import cn.itcast.rabbitmq.util.ConnectionUtil;
// 消费者2
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[] argv) throws Exception {
        // 获取到连接
        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, BasicProperties properties,
                    byte[] body) throws IOException {
                // body 即消息体
                String msg = new String(body);
                System.out.println(" [消费者2] received : " + msg + "!");
            }
        };
        // 监听队列，手动返回完成
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}

同时运行消费者1和消费者2

在这里插入图片描述
运行Send

6、订阅模型-Direct

说明

在Fanout模式中，一条消息，会被所有订阅的队列都消费。但是，在某些场景下，我们希望不同的消息被不同的队列消费。这时就要用到Direct类型的Exchange。

在Direct模型下：

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

流程图：
在这里插入图片描述
图解：

P：生产者，向Exchange发送消息，发送消息时，会指定一个routing key。
X：Exchange（交换机），接收生产者的消息，然后把消息递交给与routing key完全匹配的队列
C1：消费者，其所在队列指定了需要routing key 为 error 的消息
C2：消费者，其所在队列指定了需要routing key 为 info、error、warning 的消息

生产者

此处我们模拟商品的增删改，发送消息的RoutingKey分别是：insert、update、delete
在这里插入图片描述

package cn.itcast.rabbitmq.direct;

import cn.itcast.rabbitmq.util.ConnectionUtil;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
/**
 * 生产者，模拟为商品服务
 */
public class Send {
    private final static String EXCHANGE_NAME = "direct_exchange_test";

    public static void main(String[] argv) throws Exception {
        // 获取到连接
        Connection connection = ConnectionUtil.getConnection();
        // 获取通道
        Channel channel = connection.createChannel();
        // 声明exchange，指定类型为direct
        channel.exchangeDeclare(EXCHANGE_NAME, "direct");
        // 消息内容
        String message = "商品删除了， id = 1001";
        // 发送消息，并且指定routing key 为：insert ,代表新增商品
        channel.basicPublish(EXCHANGE_NAME, "delete", null, message.getBytes());
        System.out.println(" [商品服务：] Sent '" + message + "'");

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

运行上述代码创建对应的交换机
在这里插入图片描述

消费者1

我们此处假设消费者1只接收两种类型的消息：更新商品和删除商品。

在这里插入图片描述

package cn.itcast.rabbitmq.direct;

import java.io.IOException;

import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;

import cn.itcast.rabbitmq.util.ConnectionUtil;
/**
 * 消费者1
 */
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[] argv) throws Exception {
        // 获取到连接
        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, BasicProperties properties,
                    byte[] body) throws IOException {
                // body 即消息体
                String msg = new String(body);
                System.out.println(" [消费者1] received : " + msg + "!");
            }
        };
        // 监听队列，自动ACK
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}

消费者2

我们此处假设消费者2接收所有类型的消息：新增商品，更新商品和删除商品。
在这里插入图片描述

package cn.itcast.rabbitmq.direct;

import java.io.IOException;

import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;

import cn.itcast.rabbitmq.util.ConnectionUtil;
/**
 * 消费者2
 */
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[] argv) throws Exception {
        // 获取到连接
        Connection connection = ConnectionUtil.getConnection();
        // 获取通道
        Channel channel = connection.createChannel();
        // 声明队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        
        // 绑定队列到交换机，同时指定需要订阅的routing key。订阅 insert、update、delete
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "insert");
        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, BasicProperties properties,
                    byte[] body) throws IOException {
                // body 即消息体
                String msg = new String(body);
                System.out.println(" [消费者2] received : " + msg + "!");
            }
        };
        // 监听队列，自动ACK
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}

同时运行

同时运行消费者一和消费者二
在这里插入图片描述
运行Send生产者，模拟为商品服务

同时运行修改Send设置为商品新增

在这里插入图片描述

package cn.itcast.rabbitmq.direct;

import cn.itcast.rabbitmq.util.ConnectionUtil;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
/**
 * 生产者，模拟为商品服务
 */
public class Send {
    private final static String EXCHANGE_NAME = "direct_exchange_test";

    public static void main(String[] argv) throws Exception {
        // 获取到连接
        Connection connection = ConnectionUtil.getConnection();
        // 获取通道
        Channel channel = connection.createChannel();
        // 声明exchange，指定类型为direct
        channel.exchangeDeclare(EXCHANGE_NAME, "direct");
        // 消息内容
        String message = "商品新增了， id = 1001";
        // 发送消息，并且指定routing key 为：insert ,代表新增商品
        channel.basicPublish(EXCHANGE_NAME, "insert", null, message.getBytes());
        System.out.println(" [商品服务：] Sent '" + message + "'");

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

运行send

只有Recv2
在这里插入图片描述
Recv没有

7、订阅模型-Topic

说明

Topic类型的Exchange与Direct相比，都是可以根据RoutingKey把消息路由到不同的队列。只不过Topic类型Exchange可以让队列在绑定Routing key 的时候使用通配符！

Routingkey 一般都是由一个或多个单词组成，多个单词之间以”.”分割，例如： item.insert

通配符规则：

     `#`：匹配一个或多个词

     `*`：匹配不多不少恰好1个词

举例：

     `audit.#`：能够匹配`audit.irs.corporate` 或者 `audit.irs`

     `audit.*`：只能匹配`audit.irs`

图示：
在这里插入图片描述
解释：

红色Queue：绑定的是usa.# ，因此凡是以 usa.开头的routing key 都会被匹配到
黄色Queue：绑定的是#.news ，因此凡是以 .news结尾的 routing key 都会被匹配

生产者

使用topic类型的Exchange，发送消息的routing key有3种： item.isnert、item.update、item.delete：

在这里插入图片描述

package cn.itcast.rabbitmq.topic;

import cn.itcast.rabbitmq.util.ConnectionUtil;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
/**
 * 生产者，模拟为商品服务
 */
public class Send {
    private final static String EXCHANGE_NAME = "topic_exchange_test";

    public static void main(String[] argv) throws Exception {
        // 获取到连接
        Connection connection = ConnectionUtil.getConnection();
        // 获取通道
        Channel channel = connection.createChannel();
        // 声明exchange，指定类型为topic
        channel.exchangeDeclare(EXCHANGE_NAME, "topic");
        // 消息内容
        String message = "新增商品 : id = 1001";
        // 发送消息，并且指定routing key 为：insert ,代表新增商品
        channel.basicPublish(EXCHANGE_NAME, "item.insert", null, message.getBytes());
        System.out.println(" [商品服务：] Sent '" + message + "'");

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

消费者1

我们此处假设消费者1只接收两种类型的消息：更新商品和删除商品
在这里插入图片描述

package cn.itcast.rabbitmq.topic;

import java.io.IOException;

import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;

import cn.itcast.rabbitmq.util.ConnectionUtil;
/**
 * 消费者1
 */
public class Recv {
    private final static String QUEUE_NAME = "topic_exchange_queue_1";
    private final static String EXCHANGE_NAME = "topic_exchange_test";

    public static void main(String[] argv) throws Exception {
        // 获取到连接
        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, "item.update");
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "item.delete");

        // 定义队列的消费者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            // 获取消息，并且处理，这个方法类似事件监听，如果有消息的时候，会被自动调用
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
                    byte[] body) throws IOException {
                // body 即消息体
                String msg = new String(body);
                System.out.println(" [消费者1] received : " + msg + "!");
            }
        };
        // 监听队列，自动ACK
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}

消费者2

我们此处假设消费者2接收所有类型的消息：新增商品，更新商品和删除商品。
在这里插入图片描述

package cn.itcast.rabbitmq.topic;

import java.io.IOException;

import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;

import cn.itcast.rabbitmq.util.ConnectionUtil;
/**
 * 消费者2
 */
public class Recv2 {
    private final static String QUEUE_NAME = "topic_exchange_queue_2";
    private final static String EXCHANGE_NAME = "topic_exchange_test";

    public static void main(String[] argv) throws Exception {
        // 获取到连接
        Connection connection = ConnectionUtil.getConnection();
        // 获取通道
        Channel channel = connection.createChannel();
        // 声明队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        
        // 绑定队列到交换机，同时指定需要订阅的routing key。订阅 insert、update、delete
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "item.*");

        // 定义队列的消费者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            // 获取消息，并且处理，这个方法类似事件监听，如果有消息的时候，会被自动调用
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
                    byte[] body) throws IOException {
                // body 即消息体
                String msg = new String(body);
                System.out.println(" [消费者2] received : " + msg + "!");
            }
        };
        // 监听队列，自动ACK
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}

运行测试

运行Send创建交换机
在这里插入图片描述

运行消费者1和消费者2

再次运行Send

消费者2 有消息

消费者1没有

修改Send

在这里插入图片描述

package cn.itcast.rabbitmq.topic;

import cn.itcast.rabbitmq.util.ConnectionUtil;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
/**
 * 生产者，模拟为商品服务
 */
public class Send {
    private final static String EXCHANGE_NAME = "topic_exchange_test";

    public static void main(String[] argv) throws Exception {
        // 获取到连接
        Connection connection = ConnectionUtil.getConnection();
        // 获取通道
        Channel channel = connection.createChannel();
        // 声明exchange，指定类型为topic
        channel.exchangeDeclare(EXCHANGE_NAME, "topic");
        // 消息内容
        String message = "删除商品 : id = 1001";
        // 发送消息，并且指定routing key 为：insert ,代表新增商品
        channel.basicPublish(EXCHANGE_NAME, "item.delete", null, message.getBytes());
        System.out.println(" [商品服务：] Sent '" + message + "'");

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

运行Send
在这里插入图片描述
消费者1和消费者2 当中都有消息

8、持久化

如何避免消息丢失？

1）消费者的ACK机制。可以防止消费者丢失消息。

2）但是，如果在消费者消费之前，MQ就宕机了，消息就没了。

是可以将消息进行持久化呢？

要将消息持久化，前提是：队列、Exchange都持久化

在虚拟机当中重启RabbitMQ

在这里插入图片描述
交换机数据全部消失了

交换机持久化

在这里插入图片描述

运行上面代码

package cn.itcast.rabbitmq.topic;

import cn.itcast.rabbitmq.util.ConnectionUtil;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
/**
 * 生产者，模拟为商品服务
 */
public class Send {
    private final static String EXCHANGE_NAME = "topic_exchange_test01";

    public static void main(String[] argv) throws Exception {
        // 获取到连接
        Connection connection = ConnectionUtil.getConnection();
        // 获取通道
        Channel channel = connection.createChannel();
        // 声明exchange，指定类型为topic
        channel.exchangeDeclare(EXCHANGE_NAME, "topic",true);
        // 消息内容
        String message = "删除商品 : id = 1001";
        // 发送消息，并且指定routing key 为：insert ,代表新增商品
        channel.basicPublish(EXCHANGE_NAME, "item.delete", null, message.getBytes());
        System.out.println(" [商品服务：] Sent '" + message + "'");

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

在这里插入图片描述
自己创建的也有了D（代表持久的）

队列持久化

在这里插入图片描述

package cn.itcast.rabbitmq.topic;

import java.io.IOException;

import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;

import cn.itcast.rabbitmq.util.ConnectionUtil;
/**
 * 消费者2
 */
public class Recv2 {
    private final static String QUEUE_NAME = "topic_exchange_queue_02";
    private final static String EXCHANGE_NAME = "topic_exchange_test01";

    public static void main(String[] argv) throws Exception {
        // 获取到连接
        Connection connection = ConnectionUtil.getConnection();
        // 获取通道
        Channel channel = connection.createChannel();
        // 声明队列
        channel.queueDeclare(QUEUE_NAME, true, false, false, null);
        
        // 绑定队列到交换机，同时指定需要订阅的routing key。订阅 insert、update、delete
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "item.*");

        // 定义队列的消费者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            // 获取消息，并且处理，这个方法类似事件监听，如果有消息的时候，会被自动调用
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
                    byte[] body) throws IOException {
                // body 即消息体
                String msg = new String(body);
                System.out.println(" [消费者2] received : " + msg + "!");
            }
        };
        // 监听队列，自动ACK
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}

运行多了一个队列并且也是持久的
在这里插入图片描述

消息持久化

在这里插入图片描述

package cn.itcast.rabbitmq.topic;

import cn.itcast.rabbitmq.util.ConnectionUtil;

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

/**
 * 生产者，模拟为商品服务
 */
public class Send {
    private final static String EXCHANGE_NAME = "topic_exchange_test01";

    public static void main(String[] argv) throws Exception {
        // 获取到连接
        Connection connection = ConnectionUtil.getConnection();
        // 获取通道
        Channel channel = connection.createChannel();
        // 声明exchange，指定类型为topic
        channel.exchangeDeclare(EXCHANGE_NAME, "topic", true);
        // 消息内容
        String message = "新增商品 : id = 1001";
        // 发送消息，并且指定routing key 为：insert ,代表新增商品
        channel.basicPublish(EXCHANGE_NAME, "item.insert", MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());
        System.out.println(" [商品服务：] Sent '" + message + "'");

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

运行
在这里插入图片描述
RabbitMQ挂了或者重启消息也不会消失

生产者确认

 //生产者确认
        channel.confirmSelect();

 channel.waitForConfirms(1000);//失败等待1秒后抛出异常

在这里插入图片描述

package cn.itcast.rabbitmq.topic;

import cn.itcast.rabbitmq.util.ConnectionUtil;

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

/**
 * 生产者，模拟为商品服务
 */
public class Send {
    private final static String EXCHANGE_NAME = "topic_exchange_test01";

    public static void main(String[] argv) throws Exception {
        // 获取到连接
        Connection connection = ConnectionUtil.getConnection();
        // 获取通道
        Channel channel = connection.createChannel();
        //开启生产者确认
        channel.confirmSelect();
        // 声明exchange，指定类型为topic
        channel.exchangeDeclare(EXCHANGE_NAME, "topic", true);
        // 消息内容
        String message = "新增商品 : id = 1001";
        // 发送消息，并且指定routing key 为：insert ,代表新增商品
        channel.basicPublish(EXCHANGE_NAME, "item.insert", MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());
        System.out.println(" [商品服务：] Sent '" + message + "'");

        channel.waitForConfirms(1000);//失败等待1秒后抛出异常
        channel.close();
        connection.close();
    }
}

消息丢失的解决方案

ack(消费者确认)
持久化
发送消息前，将消息持久化到数据库，并记录消息状态（可靠消息服务)
生产者确认(publisher confirm)

幂等性(同一接口被重复执行，其结果—致)﹔

三、Spring AMQP

1、简介

https://spring.io/projects
Sprin有很多不同的项目，其中就有对AMQP的支持：
在这里插入图片描述

Spring AMQP的页面：http://projects.spring.io/spring-amqp/
在这里插入图片描述
Spring-amqp是对AMQP协议的抽象实现，而spring-rabbit 是对协议的具体实现，也是目前的唯一实现。底层使用的就是RabbitMQ

2、依赖和配置

添加AMQP的启动器：

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

在这里插入图片描述

在application.yml中添加RabbitMQ地址：
在这里插入图片描述

spring:
  rabbitmq:
    host: 192.168.206.66
    username: leyou
    password: 123321
    virtual-host: /leyou

3、监听者

在SpringAmqp中，对消息的消费者进行了封装和抽象，一个普通的JavaBean中的普通方法，只要通过简单的注解，就可以成为一个消费者。
在这里插入图片描述

package cn.itcast.rabbitmq.spring;

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;


@Component
public class Listener {

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = "spring.test.queue", durable = "true"),
            exchange = @Exchange(
                    value = "spring.test.exchange",
                    ignoreDeclarationExceptions = "true",
                    type = ExchangeTypes.TOPIC
            ),
            key = {"#.#"}))
    public void listen(String msg){
        System.out.println("接收到消息：" + msg);
    }
}

@Componet：类上的注解，注册到Spring容器
@RabbitListener：方法上的注解，声明这个方法是一个消费者方法，需要指定下面的属性：
- bindings：指定绑定关系，可以有多个。值是@QueueBinding的数组。@QueueBinding包含下面属性：
  - value：这个消费者关联的队列。值是@Queue，代表一个队列
  - exchange：队列所绑定的交换机，值是@Exchange类型
  - key：队列和交换机绑定的RoutingKey

类似listen这样的方法在一个类中可以写多个，就代表多个消费者。

4、AmqpTemplate

Spring最擅长的事情就是封装，把他人的框架进行封装和整合。

Spring为AMQP提供了统一的消息处理模板：AmqpTemplate，非常方便的发送消息，其发送方法：

在这里插入图片描述
红框圈起来的是比较常用的3个方法，分别是：

指定交换机、RoutingKey和消息体
指定消息
指定RoutingKey和消息，会向默认的交换机发送消息

5、测试代码

（1）创建队列

在这里插入图片描述

（2）测试代码（发送消息）

在这里插入图片描述

package cn.itcast.rabbitmq.spring;

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

/**
 * @author: HuYi.Zhang
 * @create: 2018-05-23 18:08
 **/
@RunWith(SpringRunner.class)
@SpringBootTest(classes = Application.class)
public class MqDemo {

    @Autowired
    private AmqpTemplate amqpTemplate;

    @Test
    public void testSend() throws InterruptedException {
        String msg = "hello, Spring boot amqp";
        this.amqpTemplate.convertAndSend("simple_query", msg);
        // 等待10秒后再结束
        //Thread.sleep(10000);
    }
}

运行
在这里插入图片描述
运行成功

可以看到

点击simple_query

（3）测试代码（接收消息）

在这里插入图片描述

package cn.itcast.rabbitmq.spring;

import org.junit.Test;
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;


@Component
public class Listener {

    @RabbitListener(queues = "simple_query")
    public void listen(String msg){
        System.out.println("接收到消息：" + msg);
    }

}

在设置暂停10秒
在这里插入图片描述

package cn.itcast.rabbitmq.spring;

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

/**

 **/
@RunWith(SpringRunner.class)
@SpringBootTest(classes = Application.class)
public class MqDemo {

    @Autowired
    private AmqpTemplate amqpTemplate;

    @Test
    public void testSend() throws InterruptedException {
        String msg = "hello, Spring boot amqp";
        this.amqpTemplate.convertAndSend("simple_query", msg);
        // 等待10秒后再结束
        Thread.sleep(10000);
    }
}

测试运行
在这里插入图片描述
运行结果（因为在刚刚测试的时候已经发送过一条数据，现在又发送所以是两条数据）

（4）不用手动声明，自动声明队列和交换机

在这里插入图片描述
在MqDemo当中测试

package cn.itcast.rabbitmq.spring;

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

/**
 * @author: HuYi.Zhang
 * @create: 2018-05-23 18:08
 **/
@RunWith(SpringRunner.class)
@SpringBootTest(classes = Application.class)
public class MqDemo {

    @Autowired
    private AmqpTemplate amqpTemplate;

    @Test
    public void testSend() throws InterruptedException {
        String msg = "hello, Spring boot amqp";
        this.amqpTemplate.convertAndSend("spring.test.queue", msg);
        // 等待10秒后再结束
        Thread.sleep(10000);
    }
}

运行结果
在这里插入图片描述
查看
自动创建好对应的队列和信息

最后

以上就是爱撒娇项链为你收集整理的安装RabbitMQ（Centos6）（入门使用教程）（消息丢失的解决方案）以及Spring AMQP的使用的全部内容，希望文章能够帮你解决安装RabbitMQ（Centos6）（入门使用教程）（消息丢失的解决方案）以及Spring AMQP的使用所遇到的程序开发问题。

如果觉得靠谱客网站的内容还不错，欢迎将靠谱客网站推荐给程序员好友。

本图文内容来源于网友提供，作为学习参考使用，或来自网络收集整理，版权属于原作者所有。

本文分类：SpringBoot
浏览次数：86 次浏览
发布日期：2023-09-23 17:40:12
本文链接：https://www.kaopuke.com/article/k-p-k_14_uzocfx_14_z_2_0.html

安装RabbitMQ（Centos6）（入门使用教程）（消息丢失的解决方案）以及Spring AMQP的使用

概述

一、RabbitMQ的安装和配置以及用户管理

1、搜索与商品服务的问题

2、消息队列（MQ）

（1）什么是消息队列

（2）AMQP和JMS

（3）常见MQ产品

（4）RabbitMQ

3、下载和安装

（1）下载

（2）安装Erlang

1）在/home/leyou/下创建一个新目录

（3）安装RabbitMQ

1）安装

2）修改配置文件

（4） 设置开机启动

（5） 启动命令

（6） 开启web管理界面

（7）开放端口

4、管理界面介绍

（1）主页

（2）添加用户

（3）创建虚拟主机（Virtual Hosts）

二、五种消息模型

1、导入demo工程

2、基本消息模型

说明

生产者

web控制台查看消息(已创建的leyou用户登录，密码是123321)

消费者获取消息

消费者的消息确认机制(Acknowlage)防止消息丢失

3、work消息模型

说明

案例一、

消费者1

消费者2

案例二、能者多劳

生产者

消费者1

消费者2

运行测试

4、订阅模型分类

5、订阅模型-Fanout

流程说明

消费者1

6、订阅模型-Direct

说明

生产者

消费者1

消费者2

同时运行

同时运行修改Send设置为商品新增

7、订阅模型-Topic

说明

生产者

消费者1

消费者2

运行测试

修改Send

8、持久化

在虚拟机当中重启RabbitMQ

交换机持久化

队列持久化

消息持久化

生产者确认

消息丢失的解决方案

三、Spring AMQP

1、简介

2、依赖和配置

3、监听者

4、AmqpTemplate

5、测试代码

（1）创建队列

（2）测试代码（发送消息）

（3）测试代码（接收消息）

（4）不用手动声明，自动声明队列和交换机

最后

相关文章

评论列表共有 0 条评论

1）在`/home/leyou/`下创建一个新目录

（4）设置开机启动

（5）启动命令

（6）开启web管理界面

发表评论取消回复