RabbitMQ 的基本概念

一 MQ 的基本概念

1 MQ概述

MQ全称 Message Queue（消息队列），是在消息的传输过程中保存消息的容器。多用于分布式系统之间进行通信。消息队列就是所谓的存放消息的队列。 消息队列解决的不是存放消息的队列的⽬的，解决的是通信问题。

  • 传统方式，系统之间直接调用 （http协议 httpclient/openFeign）

  • 中间件

2 MQ 的优势

异步、 解耦、 削峰

1 应用解耦

系统的耦合性越高，容错性就越低，可维护性就越低。以购物为例子

使用 MQ 使得应用间解耦，提升容错性和可维护性。

2 异步提速

一个下单操作耗时：20 + 300 + 300 + 300 = 920ms，用户点击完下单按钮后，需要等待920ms才能得到下单响应，太慢！

用户点击完下单按钮后，只需等待25ms就能得到下单响应 (20 + 5 = 25ms)。提升用户体验和系统吞吐量（单位时间内处理请求的数目）。

3 削峰填谷

使用了 MQ 之后，限制消费消息的速度为1000，这样一来，高峰期产生的数据势必会被积压在 MQ 中，高峰就被“削”掉了，但是因为消息积压，在高峰期过后的一段时间内，消费消息的速度还是会维持在1000，直到消费完积压的消息，这就叫做“填谷”。使用MQ后，可以提高系统稳定性。

3 MQ 的劣势

1 系统可用性降低

系统引入的外部依赖越多，系统稳定性越差。一旦 MQ 宕机，就会对业务造成影响。如何保证MQ的高可用？

2系统复杂度提高

MQ 的加入大大增加了系统的复杂度，以前系统间是同步的远程调用，现在是通过 MQ 进行异步调用。如何保证消息不被丢失等情况？

4 常见的 MQ 产品

二 RabbitMQ安装

1 上传软件

erlang­18.3­1.el7.centos.x86_64.rpm

socat­1.7.3.2­5.el7.lux.x86_64.rpm

rabbitmq­server­3.6.5­1.noarch.rpm

2 安装Erlang

rpm -ivh erlang-18.3-1.el7.centos.x86_64.rpm

3 安装RabbitMQ

rpm -ivh socat-1.7.3.2-1.1.el7.x86_64.rpm --force --nodeps

rpm -ivh rabbitmq-server-3.6.5-1.noarch.rpm

4 开启管理界面及配置

rabbitmq-plugins enable rabbitmq_management

5 启动

service rabbitmq-server start # 启动服务

service rabbitmq-server stop # 停止服务

service rabbitmq-server restart # 重启服务

6 登录

需要关闭防火墙， 远程服务器开启15672和5672开启

http://192.168.56.140:15672/

如果登录报错， 这是因为rabbitmq从3.3.0开始禁止使用guest/guest权限通过除localhost外的访问

vi /usr/lib/rabbitmq/lib/rabbitmq_server-3.6.5/ebin/rabbit.app

删除loopback_users 中的 <<"guest">>

云服务器记得开放15672端口

默认账号和密码都是 guest

三 界面介绍和操作

1 添加用户

# 角色说明： 1、 超级管理员(administrator) 可登陆管理控制台，可查看所有的信息，并且可以对用户，策略(policy)进行操作。 2、 监控者(monitoring) 可登陆管理控制台，同时可以查看rabbitmq节点的相关信息(进程数，内存使用情 况，磁盘使用情况等) 3、 策略制定者(policymaker) 可登陆管理控制台, 同时可以对policy进行管理。但无法查看节点的相关信息(上图红 框标识的部分)。 4、 普通管理者(management) 仅可登陆管理控制台，无法看到节点信息，也无法对策略进行管理。 5、 其他 无法登陆管理控制台，通常就是普通的生产者和消费者。

2 创建虚拟机

1 点击图中的Virtual Hosts

2 创建虚拟机路径，记得要带 / 

3 将虚拟机分配给用户

四 RabbitMQ概念

1 架构图

2 相关概念

Publisher - ⽣产者：发布消息到RabbitMQ中的Exchange

Consumer - 消费者：监听RabbitMQ中的Queue中的消息

Broker：接收和分发消息的应用，RabbitMQ Server就是 Message Broker，也就是我们的RabbitMQ服务器

Virtual host：出于多租户和安全因素设计的，在RabbitMQ中可以创建出多个虚拟消息服务器VirtualHost。

Connection：publisher／consumer 和 broker 之间的 TCP 连接

channel-信道： 网络信道，几乎所有操作都在channel中进行，channel是消息读写的通道。客户端可以建立多个channel，每个channel表示一个会话任务 ， 信道有特定的功能，比如创建交换机，创建队列。

Exchange - 交换机：和⽣产者建⽴连接并接收⽣产者的消息 ，并且不能保存消息。

Queue - 队列：Exchange会将消息分发到指定的Queue，Queue和消费者进⾏交互 ，队列是可以保存消息的。

Routes - 路由：交换机以什么样的策略将消息发布到Queue。生产者发消息的时候，可以给消息贴一个标签，为了让指定的消费者接收消息。

  • 结构解读：

首先安装好的RabbitMQ就是一个Broker，如果我们想将MQ给多个用户使用并且互不影响，那我们就需要将MQ通过虚拟化的方式分割成多个提供MQ的服务，也就是Virtual host，每个Virtual host都有独立的路径，并且和用户绑定。这样我们就可以在自己的世界里发消息了。

  • 通信解读：一条消息到底是怎么从生产者到了消费者的？

    • 首先生产者通过连接的方式连接到MQ的一个虚拟机，需要知道MQ的ip,端口，虚拟机路径，用户名和密码，准备好了以后就可以建立连接了TCP 连接Connection连接，

    • 但是建立和关闭TCP连接是有代价的，频繁的建立关闭TCP连接对于系统的性能有很大的影响，而且TCP的连接数也有限制，这也限制了系统处理高并发的能力。但是，在TCP连接中建立Channel是没有上述代价的，所以我们使用信道changel的方式发送和接受消息。

    • 消息进入MQ的第一站是Exchange交换机，交换机的作用：① 接收生产者发送的消息 ②和队列绑定。交换机是不保存信息的。生产者发消息的时候可以指定一个路由键，路由键可以理解为就是给消息贴了一个标签(做标记作用，消费者接收消息的时候有用)

    • 消息进入第二站queue，消费者要接收消息，需要一直监听着queue，那么消费者在监听queue的时候需要先指定队列要和那个交换机绑定，绑定的时候也需要指定路由键，如果发消息时的路由键和接收消息时候路由键一样，那么这个消息就会进入到这个队列。

    • 最后消费者就拿到消息了。需要说明的一点，所有的交换机和队列创建的时候都是需要起名字的。

3 RabbitMQ的通讯

官网介绍：RabbitMQ Tutorials — RabbitMQ

主题

五 案例解释

新建maven工程，Spring整合MQ。因为MQ中有很多概念在boot中是体会不到的，boot屏蔽了很多概念。

1 简单队列模式

1 代码

生产者和消费者都导入maven依赖

<dependency><groupId>com.rabbitmq</groupId><artifactId>amqp-client</artifactId><version>5.10.0</version>
</dependency>

  • 生产者代码，记得最后需要关闭资源。

    package com.xinzhi.product;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class MyProduct {//队列名private static final String QUEUE_NAME = "my_queue";public static void main(String[] args) throws IOException, TimeoutException {// 1.创建连接ConnectionFactory connectionFactory = new ConnectionFactory();// 2.设置连接地址connectionFactory.setHost("192.168.32.11");// 3.设置端口号:connectionFactory.setPort(5672);// 4.设置账号和密码connectionFactory.setUsername("laohan123");connectionFactory.setPassword("laohan123");// 5.设置VirtualHostconnectionFactory.setVirtualHost("/laohan");Connection connection = connectionFactory.newConnection();// 6.获取信道Channel channel = connection.createChannel();// 7.创建队列,声明并创建一个队列，如果队列已存在，则使用这个队列// 7.1第一个参数：队列名称// 7.2第二个参数：是否持久化，false对应不持久化数据，MQ停掉数据就会丢失// 7.3第三个参数：该队列是否是私有的// 7.4第四个：是否自动删除,false代表连接停掉后不自动删除掉这个队列// 7.5队列的其他参数, 一般都是nullchannel.queueDeclare("my_queue", false, false, false, null);String message = "欣知大数据";//四个参数//exchange 交换机，如果使⽤了""，表示使⽤了默认交换机，默认交换机会隐式绑定到队列，//routingKey路由键：如果使⽤了默认交换机，那么路由键就可以用队列名来代替。//props header信息，一般设置null//最后一个参数是要传递的消息字节数组channel.basicPublish("",  //使⽤默认交换机"my_queue",    //因为⽤了默认交换机，于是参数就是队列名称null,message.getBytes()        消息内容);channel.close();connection.close();System.out.println("发送成功");}
}

  • 消费者代码

    package com.xinzhi;import com.rabbitmq.client.*;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class MyConsumer {public static void main(String[] args) throws IOException, TimeoutException {// 1.创建连接ConnectionFactory connectionFactory = new ConnectionFactory();// 2.设置连接地址connectionFactory.setHost("192.168.32.11");// 3.设置端口号:connectionFactory.setPort(5672);// 4.设置账号和密码connectionFactory.setUsername("laohan123");connectionFactory.setPassword("laohan123");// 5.设置VirtualHostconnectionFactory.setVirtualHost("/laohan");Connection connection = connectionFactory.newConnection();// 6.获取信道Channel channel = connection.createChannel();// 7.声明队列channel.queueDeclare("my_queue", false, false, false, null);// 8.创建消费者Consumer consumer = new DefaultConsumer(channel) {// consumerTag 消息的唯一标识，一般用来确认消息是否被消费// envelope  封装了mq的基本方法// properties 封装了mq的基本属性// body       监听到的消息@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println(new String(body));}};// 9.消费者监听某个队列  autoAck自动签收channel.basicConsume("my_queue", false, consumer);}
}

2 代码解读：

envelope：单词 信封的意思，在这里是封装了MQ的一些基本方法

- getDeliveryTag() 获取此参数信封中包含的交货标签
- isRedeliver()    如果这是在 ack 失败后是否重新投递
- getExchange()
- getRoutingKey()

3 流程解读

这是RabbitMQ最简单的工作方式

  • 生产者声明好队列，然后把信息给了MQ默认的交换机，交换机将信息发给队列

  • 消费者也声明好队列，然后监听队列获取信息

4 抽出工具类

因为生产者和消费者都是相同的获取信道的方式

public static Connection getConnection(){// 1.创建连接ConnectionFactory connectionFactory = new ConnectionFactory();// 2.设置连接地址connectionFactory.setHost("192.168.56.140");// 3.设置端口号:connectionFactory.setPort(5672);// 4.设置账号和密码connectionFactory.setUsername("laohan123");connectionFactory.setPassword("laohan123");// 5.设置VirtualHostconnectionFactory.setVirtualHost("/laohan");Connection connection = null;try {connection = connectionFactory.newConnection();} catch (IOException e) {e.printStackTrace();} catch (TimeoutException e) {e.printStackTrace();}return connection;}

2 work queue

队列模式： 能者多劳模式

1 代码

  • 生产者

    package com.xinzhi.work.product;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.xinzhi.utils.RabbitUtil;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class MyProduct {public static void main(String[] args) throws IOException, TimeoutException {//1 获取连接和信道Connection connection = RabbitUtil.getConnection();Channel channel = connection.createChannel();//2 声明队列channel.queueDeclare("work_queue", true, false, false, null);//3 发消息(消息先到了默认交换机，交换机和队列绑定了，所以信息也会直接到了queue)for (int i = 1; i <101 ; i++) {String message = "xinzhi"+i;channel.basicPublish("","work_queue",null,message.getBytes());}//4 提示和释放资源System.out.println("发送成功");channel.close();connection.close();}
}

  • 消费者 将下面的代码再复制两份MyConsumer1，MyConsumer2，等待时间设置成100,500

    package com.xinzhi.work.consumer;import com.rabbitmq.client.*;
import com.xinzhi.utils.RabbitUtil;import java.io.IOException;public class MyConsumer1 {public static void main(String[] args) throws IOException {//1 获取连接和信道Connection connection = RabbitUtil.getConnection();Channel channel = connection.createChannel();//2 声明队列channel.queueDeclare("work_queue", true, false, false, null);//3 声明消费者一次只接受一条消息channel.basicQos(1);// 4 声明消费者Consumer consumer = new DefaultConsumer(channel) {@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {try {Thread.sleep(10);} catch (InterruptedException e) {e.printStackTrace();}System.out.println("消费标签是" + consumerTag + "消息体是" + new String(body));// 消息消费成功以后的唯一标识System.out.println(envelope.getDeliveryTag());// 确认签收当前的一条消息，如果是true是签收队列里面所有的消息channel.basicAck(envelope.getDeliveryTag(), false);}};channel.basicConsume("work_queue", consumer);}
}

2 代码解读

在简单模式的基础上添加了多个消费者，每个消费者添加了等待时间。

生产者一次往队列里投放多条消息，消费者根据能力来消费这里面的所有消息，性能强的消费的消息多，所以是能者多劳

3 订阅发布

平分秋色

交换机类型 fanout

发布订阅，这次使用了交换机，之前的两种方式都是没有显式的声明使用交换机，之前其实用的系统默认的交换机。

这次使用了交换机，但是 没有使用路由键。只要和交换机绑定了的对了都可以接受到消息，也就是上图两个队列中可以收到相同的消息。

1 代码

  • 生产者

    package com.xinzhi.fanout;import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.xinzhi.utils.RabbitUtil;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Product {public static void main(String[] args) throws IOException, TimeoutException {//1 获取连接和信道Connection connection = RabbitUtil.getConnection();Channel channel = connection.createChannel();//2 声明交换机和类型channel.exchangeDeclare("fanout_exchange", BuiltinExchangeType.FANOUT);//3 将信息发给交换机for (int i = 1; i <101 ; i++) {String message = "laohan"+i;channel.basicPublish("fanout_exchange","",null,message.getBytes());}System.out.println("success");channel.close();connection.createChannel();}
}

  • 消费者1

    package com.xinzhi.fanout;import com.rabbitmq.client.*;
import com.xinzhi.utils.RabbitUtil;import java.io.IOException;public class MyConsumer1 {public static void main(String[] args) throws IOException {//1 获取连接和信道Connection connection = RabbitUtil.getConnection();Channel channel = connection.createChannel();//2 声明队列channel.queueDeclare("fanout_queue1", true, false, false, null);//3 声明交换机channel.exchangeDeclare("fanout_exchange", BuiltinExchangeType.FANOUT);//4 交换机和队列绑定channel.queueBind("fanout_queue1", "fanout_exchange", "", null);// 5 声明消费者Consumer consumer = new DefaultConsumer(channel) {@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("tag:" + consumerTag + ",message:" + new String(body));// 消息消费成功以后的唯一标识System.out.println(envelope.getDeliveryTag());// 确认签收当前的一条消息，如果是true是签收队列里面所有的消息channel.basicAck(envelope.getDeliveryTag(), false);}};channel.basicConsume("fanout_queue1", consumer);}
}

  • 消费者2

    package com.xinzhi.fanout;import com.rabbitmq.client.*;
import com.xinzhi.utils.RabbitUtil;import java.io.IOException;public class MyConsumer2 {public static void main(String[] args) throws IOException {//1 获取连接和信道Connection connection = RabbitUtil.getConnection();Channel channel = connection.createChannel();//2 声明队列channel.queueDeclare("fanout_queue2", true, false, false, null);//3 声明交换机channel.exchangeDeclare("fanout_exchange", BuiltinExchangeType.FANOUT);//4 交换机和队列绑定channel.queueBind("fanout_queue2", "fanout_exchange", "", null);// 5 声明消费者Consumer consumer = new DefaultConsumer(channel) {@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("tag:" + consumerTag + ",message:" + new String(body));// 消息消费成功以后的唯一标识System.out.println(envelope.getDeliveryTag());// 确认签收当前的一条消息，如果是true是签收队列里面所有的消息channel.basicAck(envelope.getDeliveryTag(), false);}};channel.basicConsume("fanout_queue2", consumer);}
}

4 .路由 routing

暗送秋波

1 概念

交换机direct

  • 在⽣产者发送消息时指明routing-key

  • 在消费者声明队列和交换机的绑定关系时，指明routing-key

  • 解决的问题是：

    • 因为交换机和两个队列都绑定了，但是为了给队列里发送的消息不一样，也就是区分给那个队列发什么样 的消息，就有了routing key的概念。发消息的时候指定一下路由键，接收消息的时候队列要和交换机绑定，这时候也需要指定路由键，如果这两次的路由键一样，那么这个消息就放着这个队列里面

2 代码

  • 生产者

    package com.xinzhi.direct;import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.xinzhi.utils.RabbitUtil;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Product {public static void main(String[] args) throws IOException, TimeoutException {//1 获取连接和信道Connection connection = RabbitUtil.getConnection();Channel channel = connection.createChannel();//2 声明交换机和类型,并且持久化channel.exchangeDeclare("direct_exchange", BuiltinExchangeType.DIRECT,true);//3 将信息发给交换机，并且指定路由键String message1 = "laohan1";String message2 = "laohan2";channel.basicPublish("direct_exchange","han",null,message1.getBytes());channel.basicPublish("direct_exchange","man",null,message2.getBytes());System.out.println("success");channel.close();connection.close();}
}

  • 消费者1

    package com.xinzhi.direct;import com.rabbitmq.client.*;
import com.xinzhi.utils.RabbitUtil;import java.io.IOException;public class MyConsumer1 {public static void main(String[] args) throws IOException {//1 获取连接和信道Connection connection = RabbitUtil.getConnection();Channel channel = connection.createChannel();//2 声明队列channel.queueDeclare("direct_queue1", true, false, false, null);//3 声明交换机channel.exchangeDeclare("direct_exchange", BuiltinExchangeType.DIRECT,true);//4 交换机和队列绑定channel.queueBind("direct_queue1", "direct_exchange", "han");// 5 声明消费者Consumer consumer = new DefaultConsumer(channel) {@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("tag:" + consumerTag + ",message:" + new String(body));// 消息消费成功以后的唯一标识System.out.println(envelope.getDeliveryTag());// 确认签收当前的一条消息，如果是true是签收队列里面所有的消息channel.basicAck(envelope.getDeliveryTag(), false);}};channel.basicConsume("direct_queue1", consumer);}
}

  • 消费者2

    package com.xinzhi.direct;import com.rabbitmq.client.*;
import com.xinzhi.utils.RabbitUtil;import java.io.IOException;public class MyConsumer2 {public static void main(String[] args) throws IOException {//1 获取连接和信道Connection connection = RabbitUtil.getConnection();Channel channel = connection.createChannel();//2 声明队列channel.queueDeclare("direct_queue2", true, false, false, null);//3 声明交换机channel.exchangeDeclare("direct_exchange", BuiltinExchangeType.DIRECT,true);//4 交换机和队列绑定channel.queueBind("direct_queue2", "direct_exchange", "man");// 5 声明消费者Consumer consumer = new DefaultConsumer(channel) {@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("tag:" + consumerTag + ",message:" + new String(body));// 消息消费成功以后的唯一标识System.out.println(envelope.getDeliveryTag());// 确认签收当前的一条消息，如果是true是签收队列里面所有的消息channel.basicAck(envelope.getDeliveryTag(), false);}};channel.basicConsume("direct_queue2", consumer);}
}

5 通配符模式

你的心思我要

1 概念

交换机是 topic

  • 因为路由模式里是精确匹配，比较局限，使用通配符方式，通配符，提⾼了匹配的范围，扩展业务。

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

  • 通配符规则：# 匹配一个或多个词，* 匹配不多不少恰好1个词，例如：item.# 能够匹配 item.insert.abc 或者 item.insert，item.* 只能匹配 item.insert。

2 代码

  • 生产者

    package com.xinzhi.topic;import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.xinzhi.utils.RabbitUtil;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Product {public static void main(String[] args) throws IOException, TimeoutException {//1 获取连接和信道Connection connection = RabbitUtil.getConnection();Channel channel = connection.createChannel();//2 声明交换机和类型,并且持久化channel.exchangeDeclare("topic_exchange", BuiltinExchangeType.TOPIC,true);//3 将信息发给交换机，并且指定路由键String message1 = "laohanxueit";channel.basicPublish("topic_exchange","xinzhi.15",null,message1.getBytes());System.out.println("success");channel.close();connection.close();}
}

  • 消费者

    package com.xinzhi.topic;import com.rabbitmq.client.*;
import com.xinzhi.utils.RabbitUtil;import java.io.IOException;public class MyConsumer {public static void main(String[] args) throws IOException {//1 获取连接和信道Connection connection = RabbitUtil.getConnection();Channel channel = connection.createChannel();//2 声明队列channel.queueDeclare("topic_queue", true, false, false, null);//3 声明交换机channel.exchangeDeclare("topic_exchange", BuiltinExchangeType.TOPIC,true);//4 交换机和队列绑定channel.queueBind("topic_queue", "topic_exchange", "xinzhi.#");// 5 声明消费者Consumer consumer = new DefaultConsumer(channel) {@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("tag:" + consumerTag + ",message:" + new String(body));// 消息消费成功以后的唯一标识System.out.println(envelope.getDeliveryTag());// 确认签收当前的一条消息，如果是true是签收队列里面所有的消息channel.basicAck(envelope.getDeliveryTag(), false);}};channel.basicConsume("topic_queue", consumer);}
}

六 SpringBoot整合

1 发布订阅

1 新建boot项目

2 导入依赖

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

3 配置文件

server:port: 8099
spring:rabbitmq:host: 192.168.56.140port: 5672username: laohan123password: laohan123virtual-host: /laohan

4 配置类

import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class RabbitConfig {public static final String EXCHANGE_NAME = "fanout_exchage";public static final String QUEUE_NAME = "fanout_queue";@Bean("queue")public Queue queue(){
//        return new Queue(QUEUE_NAME, true, false, false);return QueueBuilder.durable(QUEUE_NAME).build();}@Bean("exchange")public Exchange exchange(){
//        return new FanoutExchange(EXCHANGE_NAME, true, false);return ExchangeBuilder.fanoutExchange(EXCHANGE_NAME).durable(true).build();}@Beanpublic Binding binding(@Qualifier("queue") Queue queue, @Qualifier("exchange") Exchange exchange){return BindingBuilder.bind(queue).to(exchange).with("").noargs();}}

5监听类

import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;import java.io.IOException;@Component
public class RabbitListen {@RabbitListener(queues = {RabbitConfig.QUEUE_NAME})
public void listener(String body,Message message, Channel channel) throws IOException {long msgTag = message.getMessageProperties().getDeliveryTag();System.out.println("msgTag==>"+msgTag);System.out.println("message==>"+message);System.out.println("body==>"+body);}
}

6 测试类发送消息

@Autowired
private RabbitTemplate rabbitTemplate;@Test
void contextLoads() {rabbitTemplate.convertAndSend(RabbitConfig.EXCHANGE_NAME,"","老韩学it");
}

2 topic

  • 在发布订阅的基础上修改交换机名称和路由绑定就可以了

    package com.xinzhi.config;import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class RabbitConfig {
//    public static final String EXCHANGE_NAME = "fanout_exchage";public static final String EXCHANGE_NAME = "topic_exchange";
//    public static final String QUEUE_NAME = "fanout_queue";public static final String QUEUE_NAME = "topic_queue";@Bean("queue")public Queue queue(){
//        return new Queue(QUEUE_NAME, true, false, false);return QueueBuilder.durable(QUEUE_NAME).build();}@Bean("exchange")public Exchange exchange(){
//        return new FanoutExchange(EXCHANGE_NAME, true, false);
//        return ExchangeBuilder.fanoutExchange(EXCHANGE_NAME).durable(true).build();return ExchangeBuilder.topicExchange(EXCHANGE_NAME).durable(true).build();}@Beanpublic Binding binding(@Qualifier("queue") Queue queue, @Qualifier("exchange") Exchange exchange){return BindingBuilder.bind(queue).to(exchange).with("xinzhi.#").noargs();}}

  • 发送消息验证

     @Autowiredprivate RabbitTemplate rabbitTemplate;@Testvoid contextLoads() {rabbitTemplate.convertAndSend(RabbitConfig.EXCHANGE_NAME,"xinzhi.15","老韩学it");}

七 消息的可靠性投递

1 什么是消息的可靠性投递

  • 保证消息一定能发到消息队列中

  • 细节

    • 保证mq节点成功接受消息

    • 消息发送端需要接受到mq服务端接收到消息的确认应答

    • 完善的消息补偿机制，发送失败的消息可以再感知并二次处理

  • RabbitMQ消息投递路径

    • 生产者-->交换机-->队列-->消费者

    • 通过两个点的控制，保证消息的可靠性投递

      • 生产者到交换机 confirmCallback

      • 交换机到队列 returnCallbakc

  • 建议

    • 开启消息确认机制以后，保证了消息的准确送达，但由于频繁的确认交互，RabbitMQ的整体效率变低，吞吐量下降严重，不是非常重要的消息不建议用消息确认机制

2 confirmCallback

  • 机制：

    生产者投递消息以后，如果Broker收到消息以后，会给生产者一个ACK，生产者通过ACK可以确认这条消息是否成功发送到Broker。

  • 开启confirmCallback

    spring.rabbitmq.publisher-confirm-type: correlated

  • 发送代码

@Test
void confirm(){rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {/*** 消息到交换机的确认* @param correlationData  配置信息* @param ack              交换机确认  true消息接受成功  false消息接受失败* @param cause             消息发送失败原因*/@Overridepublic void confirm(CorrelationData correlationData, boolean ack, String cause) {System.out.println("ConfirmCallback==========>");System.out.println("correlationData==========>"+correlationData);System.out.println("ack==========>"+ack);System.out.println("cause==========>"+cause);if(ack){System.out.println("发送成功");// 更新数据库  成功}else {System.out.println("发送失败,日志或数据库纪录");// 更新数据库  失败}}});rabbitTemplate.convertAndSend(RabbitConfig.EXCHANGE_NAME,"xinzhi.15","老韩学it");
}

  • 模拟失败场景,修改发送时候交换机名称

2 returnCallback

  • return机制保证消息在rabbitmq中能够成功的投递到队列⾥

  • 两种模式：

    • 交换机到队列不成功，则丢弃消息（默认）

    • 交换机到队列不成功，返回生产者，触发returnCallback

  • 开启returnCallback，交换机到队列的可靠性投递

    spring.rabbitmq.publisher-returns=true

  • 修改投递到队列失败的策略

    spring.rabbitmq.template.mandatory=true

  • 发送消息验证.

     @Test
void returnCallback(){rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {@Overridepublic void returnedMessage(ReturnedMessage returned) {int code = returned.getReplyCode();System.out.println("code==>"+code);System.out.println("returned==>"+returned);}});rabbitTemplate.convertAndSend(RabbitConfig.EXCHANGE_NAME,"xinzhi.15","老韩学it");
}

  • 发送消息以后，没有任何提示，我们修改路由键

八 消息确认

1 背景

保证消息从队列到消费者的过程。

2 ACK介绍

  • 消费者从RabbitMQ中获取消息并且处理完成以后，反馈给RabbitMQ，RabbitMQ收到确认消息以后才能把消息从队列中删除

  • 消费者在处理消息的时候出现了网络不稳定、服务器异常等情况，那么就不会有ACK反馈，RabbitMQ认为这个消息没有正常消费，就将这个消息放回队列里面

  • 只有当消费者正确发送ack以后，RabbitMQ才会把消息从队列中删除

  • 消息的ack确认机制默认是打开的，消息如果未被进行ack的消息确认机制，这条消息将被锁定

3 确认方式

  • 自动

  • 手动manual

    spring.rabbitmq.listener.simple.acknowledge-mode=manual

  • 发送消息，并且开启监听模式，虽然消息被消费了，但是因为开启了手动确认模式配置，但是代码里没有手动确认所以队列里的消息不会删除

  • 代码中开启确认机制

 channel.basicAck(msgTag,false);

  • 消息拒绝

// false 一次拒绝一条   true 重新回到队列
channel.basicNack(msgTag,false,true);

结果就会看到控制台一直接受消息，因为对列有消息就会被监听到，监听以后拒绝了又放到队列里面，然后 又监听...

  • DeliveryTag

    表示消息投递的序号，每次消费消息或者消息重新投递以后，DeliveryTag都会+1

  • basicReject

    也是消息拒绝的，一次只能拒绝一条消息，也可以设置是否重新回如队列

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mzph.cn/news/618970.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

三、GCC编译：链接

三、GCC编译：链接

代码准备 main.c extern int shared; extern void func(int *a, int *b); int main(){int a 100;func(&a, &shared);return 0; }func.c int shared 1; int tmp 0; void func(int *a, int *b){tmp *a;*a *b;*b tmp; }静态链接 编译 gcc -static -fno-stack-p…
阅读更多...
TypeScript学习笔记、鸿蒙开发学习笔记

TypeScript学习笔记、鸿蒙开发学习笔记

变量定义方式 # 变量声明 let msg: string douzi console.log(msg) let num: number 20 console.log(num) let end: boolean true console.log("end" end) let a: any 10 a douzi console.log(a) let p {name:"douzi",age:20} console.log(p.name)…
阅读更多...
cmake-将源文件编译为可执行文件

cmake-将源文件编译为可执行文件

文章目录 准备工作创建cmake文件编译源代码需要的文件不在同一个文件夹的情况 准备工作 首先我们准备三个文件 add.h int add(int a,int b);add.cpp int add(int a,int b) {return ab; }main.cpp #include "add.h" #include <iostream> int main() {in…
阅读更多...
Multi-Concept Customization of Text-to-Image Diffusion——【代码复现】

Multi-Concept Customization of Text-to-Image Diffusion——【代码复现】

本文是发表于CVPR 2023上的一篇论文&#xff1a;[2212.04488] Multi-Concept Customization of Text-to-Image Diffusion (arxiv.org) 一、引言 本文主要做的工作是对stable-diffusion的预训练模型进行微调&#xff0c;需要的显存相对较多&#xff0c;论文中测试时是在两块GP…
阅读更多...
ssh远程登录协议

ssh远程登录协议

目录 一、ssh协议定义及特点 &#xff08;一&#xff09;ssh协议简介 &#xff08;二&#xff09;ssh协议的主要特点 二、ssh传输原理 三、sshd服务 &#xff08;一&#xff09;配置文件 &#xff08;二&#xff09;基本用法 1.直接连接 2.连接指定用户 3.指定端口号…
阅读更多...
设计模式之策略模式【行为型模式】

设计模式之策略模式【行为型模式】

提示&#xff1a;文章写完后&#xff0c;目录可以自动生成&#xff0c;如何生成可参考右边的帮助文档> 学习的最大理由是想摆脱平庸&#xff0c;早一天就多一份人生的精彩&#xff1b;迟一天就多一天平庸的困扰。各位小伙伴&#xff0c;如果您&#xff1a; 想系统/深入学习某…
阅读更多...
十分钟教你在 k8s 中部署一个前后端应用

十分钟教你在 k8s 中部署一个前后端应用

好多开发人员&#xff0c;尤其是没接触过 k8s 的人员对如何在k8s中部署一个 前后端应用很模糊&#xff0c;不知如何下手&#xff0c;所以本篇讲一下如何快速在 k8s 部署一个前后端应用&#xff0c;让大家对k8s有个快速认识 前置依赖 k8s集群&#xff0c;如果没有安装&#xff0…
阅读更多...
什么是RoPE-旋转位置编码？

什么是RoPE-旋转位置编码？

RoPE位置编码是大模型中最常见的位置编码之一。像是谷歌的PaLM和meta的LLaMA等开源大模型都是RoPE位置编码&#xff0c;那么RoPE有什么特点呢&#xff1f; 本文将介绍如下内容&#xff1a; RoPE旋转位置编码概要什么是位置编码&#xff1f;RoPE及其特点总结 一、RoPE旋转位置…
阅读更多...
stm32学习笔记：DMA

stm32学习笔记：DMA

每个DMA通道硬件触发源是不一样的&#xff0c;要使用某个外设的硬件触发源&#xff0c;就必须使用它连接的那个通道 12个独立可配置的通道&#xff1a;DMA1(7个通道)&#xff0c;DMA2(5个通道) 每个通道都支持软件触发和特定的硬件触发 C8T6 DMA资源&#xff1a;DMA1 &#xff…
阅读更多...
Python自动化我选DrissionPage，弃用Selenium

Python自动化我选DrissionPage，弃用Selenium

DrissionPage 是一个基于 python 的网页自动化工具。 它既能控制浏览器&#xff0c;也能收发数据包&#xff0c;还能把两者合而为一。 可兼顾浏览器自动化的便利性和 requests 的高效率。 它功能强大&#xff0c;内置无数人性化设计和便捷功能。 它的语法简洁而优雅&#x…
阅读更多...
数据库和表的操作

数据库和表的操作

文章目录 前言一、库的操作创建数据库字符集和校验规则操纵数据库查看数据库显示创建语句修改数据库删除数据库备份和恢复数据库还原查看连接情况 二、表的操作创建表查看表结构修改表修改表名添加一列修改某一列属性删除某一列 删除表 前言 一、库的操作 创建数据库 语法&am…
阅读更多...
【Python机器学习】分类器的不确定估计——决策函数

【Python机器学习】分类器的不确定估计——决策函数

scikit-learn接口的分类器能够给出预测的不确定度估计&#xff0c;一般来说&#xff0c;分类器会预测一个测试点属于哪个类别&#xff0c;还包括它对这个预测的置信程度。 scikit-learn中有两个函数可以用于获取分类器的不确定度估计&#xff1a;decidion_function和predict_pr…
阅读更多...
day06

day06

1.八大基本数据类型整型: byte short int long 浮点: double float字符: char布尔: boolean2.声明变量的语法格式数据类型 变量名字 初始化的值; 3.使用for循环打印等腰三角形for (int i 1; i < 4; i) {for (int j 1; j < 4 - i; j) {System.out.print(" &qu…
阅读更多...
计算机毕业设计----SSH电子相册管理系统

计算机毕业设计----SSH电子相册管理系统

项目介绍 本项目分为普通用户与管理员两个角色&#xff1b; 管理员角色包含以下功能&#xff1a; 管理员登陆,用户信息管理,登陆密码修改等功能。 用户角色包含以下功能&#xff1a; 用户登陆,个人信息管理,相册管理,上传照片等功能。 环境需要 1.运行环境&#xff1a;最好…
阅读更多...
高效办公：在文件夹名称左边插入关键字，提高文件管理效率

高效办公：在文件夹名称左边插入关键字，提高文件管理效率

在繁忙的工作环境中&#xff0c;经常要处理大量的文件和文件夹。有效的文件管理是一个挑战&#xff0c;大量的文件和文件夹难以找到所需的资料。下面一起来看云炫文件管理器如何在文件夹名称左边批量插入关键字。 文件夹名称左边添加关键字前后对比图。 文件夹名称左边批量插…
阅读更多...
P1379 八数码难题

P1379 八数码难题

题目描述 在 33 的棋盘上&#xff0c;摆有八个棋子&#xff0c;每个棋子上标有 1 至 8 的某一数字。棋盘中留有一个空格&#xff0c;空格用 0 来表示。空格周围的棋子可以移到空格中。要求解的问题是&#xff1a;给出一种初始布局&#xff08;初始状态&#xff09;和目标布局&…
阅读更多...
哪里能找到好用的PPT模板？12个免费模板网站让你畅快办公！

哪里能找到好用的PPT模板？12个免费模板网站让你畅快办公！

你是否有过这样的经历&#xff0c;在准备重要会议或者演讲的时候&#xff0c;为找不到合适的PPT模板而困扰&#xff1f;或是在网上漫无目的地搜寻&#xff0c;结果收获的是设计平淡无奇的PPT模板&#xff1f; 如果你有同样的疑问&#xff0c;那么你来对地方了&#xff01;在这…
阅读更多...
电脑扩容升级硬盘选1T还是2T

电脑扩容升级硬盘选1T还是2T

SSD固态有必要升级2TB吗&#xff1f;----------吴中函 某大二学生用的一台笔记本电脑&#xff0c;512GB的硬盘空间已经严重不够用了&#xff0c;想给笔记本扩容升级一下硬盘&#xff1b; 这位学生是学设计专业的、平时也喜欢摄影、电脑里面也装了一些游戏&#xff0c;经常整理、…
阅读更多...
Linux技术，winSCP连接服务器超时故障解决方案

Linux技术，winSCP连接服务器超时故障解决方案

知识改变命运&#xff0c;技术就是要分享&#xff0c;有问题随时联系&#xff0c;免费答疑&#xff0c;欢迎联系&#xff01; 故障现象 使用 sftp 协议连接主机时, 明显感觉缓慢且卡顿,并且时常出现如下报错: 点击重新连接后,又有概率重新连接上; 总之在"连接上"和&…
阅读更多...
【Web】NSSCTF Round#16 Basic个人wp(全)

【Web】NSSCTF Round#16 Basic个人wp(全)

出题友好&#xff0c;适合手生复健。 目录 ①RCE但是没有完全RCE ②了解过PHP特性吗 ①RCE但是没有完全RCE 上来就是一段md5八股 (string)就是不让用数组了&#xff0c;然后强比较需要md5碰撞 ?md5_1%4d%c9%68%ff%0e%e3%5c%20%95%72%d4%77%7b%72%15%87%d3%6f%a7%b2%1b%dc…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023