RabbitMQ 的基本概念

news/2024/7/27 7:55:41/文章来源:https://blog.csdn.net/weixin_65700815/article/details/135573284

一 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.luyixian.cn/news_show_926294.aspx

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

相关文章

使用CSS计算高度铺满屏幕

使用CSS计算高度铺满屏幕

前言 今天写项目时出现高度设置百分百却不占满屏幕&#xff0c;第一反应看自己设置的是块级元素还是行级元素。看了几篇博客&#xff0c;发现并不能解决问题。脱离文档流的做法都没考虑&#xff0c;前期模板搭建脱离文档流&#xff0c;后面开发会出现很多问题。 以上图片是我…
阅读更多...
如何优雅的进行异步编排

如何优雅的进行异步编排

Future模式 Future模式是高并发设计和开发过程中常见的设计模式。它的核心思想是异步调用&#xff0c;对于Future模式来说&#xff0c;它不是立即返回我们所需要的结果&#xff0c;但是它会返回一个异步任务&#xff0c;将来我们可以通过这个异步任务获取到我们所需要的结果。…
阅读更多...
Linux-ARM裸机(十一)-UART串口通信

Linux-ARM裸机(十一)-UART串口通信

无论单片机开发还是嵌入式 Linux 开发&#xff0c;串口都是最常用到的外设。可通过串口将开发板与电脑相连&#xff0c;然后在电脑上通过串口调试助手来调试程序。还有很多的模块&#xff0c;比如蓝牙、GPS、 GPRS 等都使用的串口来与主控进行通信的&#xff0c;在嵌入式 Linux…
阅读更多...
基于信号完整性的PCB设计原则

基于信号完整性的PCB设计原则

最小化单根信号线质量的一些PCB设计建议 1. 使用受控阻抗线&#xff1b; 2. 理想情况下&#xff0c;所有信号都应该使用完整的电源或地平面作为其返回路径&#xff0c;关键信号则使用地平面作为返回路径&#xff1b; 3. 信号的返回参考面发生变化时&#xff0c;在尽可能接近…
阅读更多...
wins安装paddle框架

wins安装paddle框架

一、安装 https://www.paddlepaddle.org.cn/install/quick?docurl/documentation/docs/zh/install/pip/windows-pip.html 装包&#xff08;python 的版本是否满足要求&#xff1a; 3.8/3.9/3.10/3.11/3.12&#xff0c; pip 版本为 20.2.2 或更高版本 &#xff09; CPU 版:…
阅读更多...
LLM论文：ALCE （Enabling Large Language Models to Generate Text with Citations）

LLM论文：ALCE （Enabling Large Language Models to Generate Text with Citations）

这是一篇RAG领域的文章&#xff0c;原文在这&#xff1a;https://aclanthology.org/2023.emnlp-main.398.pdf 时间[Submitted on 24 May 2023 (v1), last revised 31 Oct 2023 (this version, v2)]背景LLM在信息搜索、生成带引用的文本时存在幻觉问题&#xff0c;即事实准确性…
阅读更多...
《 乱弹篇（四）》

《 乱弹篇（四）》

既然是“乱弹”&#xff0c;弹&#xff08;谈&#xff09;题便可以包罗万象&#xff1b;天上地下&#xff0c;飞的走的&#xff0c;你的我的他的事儿&#xff0c;甚至还有许许多多八竿子都打不着的怪涎事儿&#xff0c;都可成为弹&#xff08;谈&#xff09;资 。比如&#xff…
阅读更多...
计算机毕业设计 | 大型SpringBoot宠物医院管理 宠物商城购物系统(附源码)

计算机毕业设计 | 大型SpringBoot宠物医院管理 宠物商城购物系统(附源码)

写在前面 Le Dao宠物医院管理系统是一个超大型的&#xff0c;完成度很高的&#xff0c;集宠物医疗、宠物美容、宠物交易、宠物周边等各种功能于一身的&#xff0c;权限涵盖普通用户、医生、化验师、美容师、仓库主管、采购员等多种角色于一体的大型宠物医疗&#xff0c;购物系…
阅读更多...
Rust-Panic

Rust-Panic

什么是panic 在Rust中&#xff0c;有一类错误叫作panic。示例如下&#xff1a; 编译&#xff0c;没有错误&#xff0c;执行这段程序&#xff0c;输出为&#xff1a; 这种情况就引发了一个panic。在这段代码中&#xff0c;我们调用了Option::unwrap()方法&#xff0c;正是这个方…
阅读更多...
【开源】基于JAVA语言的网上药店系统

【开源】基于JAVA语言的网上药店系统

目录 一、摘要1.1 项目介绍1.2 项目录屏 二、功能模块2.1 数据中心模块2.2 药品类型模块2.3 药品档案模块2.4 药品订单模块2.5 药品收藏模块2.6 药品资讯模块 三、系统设计3.1 用例设计3.2 数据库设计3.2.1 角色表3.2.2 药品表3.2.3 药品订单表3.2.4 药品收藏表3.2.5 药品留言表…
阅读更多...
【Django开发】美多商城项目第2篇：Django用户注册和登录开发（附代码，已分享）

【Django开发】美多商城项目第2篇：Django用户注册和登录开发（附代码，已分享）

本系列文章md笔记&#xff08;已分享&#xff09;主要讨论django商城项目相关知识。项目利用Django框架开发一套前后端不分离的商城项目&#xff08;4.0版本&#xff09;含代码和文档。功能包括前后端不分离&#xff0c;方便SEO。采用Django Jinja2模板引擎 Vue.js实现前后端…
阅读更多...
Kubernetes (K8S) 3 小时快速上手 + 实践

Kubernetes (K8S) 3 小时快速上手 + 实践

1. Kubernetes 简介 k8s即Kubernetes。其为google开发来被用于容器管理的开源应用程序&#xff0c;可帮助创建和管理应用程序的容器化。用一个的例子来描述&#xff1a;"当虚拟化容器Docker有太多要管理的时候&#xff0c;手动管理就会很麻烦&#xff0c;于是我们便可以通…
阅读更多...
【SQL注入】SQLMAP v1.7.11.1 汉化版

【SQL注入】SQLMAP v1.7.11.1 汉化版

下载链接 【SQL注入】SQLMAP v1.7.11.1 汉化版 简介 SQLMAP是一款开源的自动化SQL注入工具&#xff0c;用于扫描和利用Web应用程序中的SQL注入漏洞。它在安全测试领域被广泛应用&#xff0c;可用于检测和利用SQL注入漏洞&#xff0c;以验证应用程序的安全性。 SQL注入是一种…
阅读更多...
初识OpenCV

初识OpenCV

首先你得保证你的虚拟机Ubuntu能上网 可看 http://t.csdnimg.cn/bZs6c 打开终端输入 sudo apt-get install libopencv-dev 回车 输入密码 回车 遇到Y/N 回车 OpenCV在线文档 opencv 文档链接 点zip可以下载&#xff0c;点前面的直接在线浏览&#xff0c;但是很慢 https…
阅读更多...
k8s云原生环境搭建笔记——第二篇

k8s云原生环境搭建笔记——第二篇

目录 1、使用普通方式安装prometheus和grafana1.1、安装kube-state-metrics容器1.1.1、下载并修改yaml文件1.1.2、导入kube-state-metrics镜像1.1.3、执行yaml文件目录 1.2、安装node-exploer1.2.1、创建名称空间prometheus1.2.2、执行yaml 1.3、安装prometheus1.3.1、创建集群…
阅读更多...
基于python集成学习算法XGBoost农业数据可视化分析预测系统

基于python集成学习算法XGBoost农业数据可视化分析预测系统

文章目录 基于python集成学习算法XGBoost农业数据可视化分析预测系统一、项目简介二、开发环境三、项目技术四、功能结构五、功能实现模型构建封装类用于网格调参训练模型系统可视化数据请求接口模型评分 0.5*mse 六、系统实现七、总结 基于python集成学习算法XGBoost农业数据可…
阅读更多...
多行SQL转成单行SQL

多行SQL转成单行SQL

如下图所示 将以上多行SQL转成单行SQL 正则表达式如下 (?s)$[^a-zA-Z()0-9]*结果如下 灵活使用,也未必只能使用Sublime Text 提供了一个在线工具
阅读更多...
[Docker] Docker为什么出现

[Docker] Docker为什么出现

Docker为什么出现 一款产品&#xff1a; 开发–上线 -->两套环境 | 应用配置 开发即运维&#xff01; 环境配置十分麻烦&#xff0c;每一个机器都要部署环境&#xff08;Redis, ES, Hadoop&#xff09; 费时费力 项目带上配置环境安装打包。 传统&#xff1a; 开发jar&…
阅读更多...
基于pyqt5+scapy 根据ip 具体端口 进行扫描 的程序

基于pyqt5+scapy 根据ip 具体端口 进行扫描 的程序

先给出代码 import sysfrom PyQt5 import uic from PyQt5.QtWidgets import *from scapy.all import * import argparse import logging from scapy.layers.inet import IP, TCP from scapy.sendrecv import sr1class MyWindow(QWidget):def __init__(self):super().__init__(…
阅读更多...
AI人工智能工程师证书专业认证培训班有用吗？

AI人工智能工程师证书专业认证培训班有用吗？

当然有用&#xff0c;它即让自身技术技能有所提升&#xff0c;也拿到行内有含金量的证书&#xff0c;让自己在选择职业、升职加薪中更有竞争力。但是要擦亮眼睛&#xff0c;建议大家如果要找人工智能培训&#xff0c;就找性价比较高的培训班&#xff0c; 人工智能AI培训班怎么…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023

友情链接： 我的编程经验分享 一条长河网 尧图网站开发 尧图建网站 广利置业