【RabbitMQ】消息队列-RabbitMQ篇章

文章目录

    • 1、RabbitMQ是什么
      • 1.1、RabbitMQ---使用场景
          • 一般场景
          • 解耦
          • 削峰
          • 异步
    • 2、Dokcer安装RabbitMQ
      • 2.1安装Dokcer
      • 2.2安装rabbitmq
    • 3、RabbitMQ入门案例 - Simple 简单模式
    • 4、RabbitMQ的核心组成部分
      • 4.1 RabbitMQ整体架构
      • 4.2RabbitMQ的运行流程
    • 5、RabbitMQ的模式
      • 5.1 发布订阅模式--fanout
      • 5.2路由模式-direct模式
      • 5.3路由模式-Topic模式
      • 5.4轮询模式 - Work模式
        • 5.4.1Work模式 - 轮询模式(Round-Robin)
        • 5.4.1Work模式 - 公平分发模式(Round-Robin)

1、RabbitMQ是什么

RabbitMQ是一个开源的遵循AMQP协议实现的基于Erlang语言编写,支持多种客户端(语言)。用于在分布式系统中存储消息,转发消息,具有高可用高可扩性易用性等特征。

1.1、RabbitMQ—使用场景

一般场景

像一般的下订单业务如下图:

将订单信息写入数据库成功后,发送注册邮件,再发送注册短信。以上三个任务全部完成后,返回给客户端,
像这样耗时就很大 = 所有服务操作的耗时总和,而且若是这一整条执行链某个环节出了问题触发回滚,得不偿失

在这里插入图片描述

public void makeOrder(){// 1 :保存订单 orderService.saveOrder();// 2: 发送短信服务messageService.sendSMS("order");//1-2 s// 3: 发送email服务emailService.sendEmail("order");//1-2 s// 4: 发送APP服务appService.sendApp("order");    
}

那么当我们开辟一个线程池去异步处理的话,也存在缺点:(最大的原因就是自己去实现起来,因素过多,实现复杂)
存在问题:
1:耦合度高
2:需要自己写线程池自己维护成本太高
3:出现了消息可能会丢失,需要你自己做消息补偿
4:如何保证消息的可靠性你自己写
5:如果服务器承载不了,你需要自己去写高可用
在这里插入图片描述

所以MQ就诞生了
在这里插入图片描述
只管下单,下单后直接就给用户提示下单成功,别的事交给mq去派发,让别的服务去mq拿消息处理

用户响应耗时 = 下单(主要)耗时(50ms) 别的(次要)处理服务全放到消息队列当中等待处理

public void makeOrder(){// 1 :保存订单 orderService.saveOrder();   rabbitTemplate.convertSend("ex","2","消息内容");
}
解耦

发送方将消息发送到消息队列中,接收方从队列中获取消息进行处理。这种松耦合的通信模式可以提高系统的可扩展性和灵活性。

这样使得下单服务并不受,发短信、发邮件、等等服务的影响(前提是下单不依赖任何一个服务的返回值)
在这里插入图片描述
在这里插入图片描述

削峰

当系统面临突然的请求高峰时,消息队列可以起到缓冲的作用。请求先进入消息队列排队,然后逐个被处理,使得系统能够逐渐消化高峰期的请求压力,避免过载和故障

也就是如果某个时刻有大量的请求,此时都会到mq里面去,而不会瞬间开启很多线程去异步执行,从而达到销峰的效果,使得即便大量的用户请求来了,那系统处理请求还是非常平滑的
在这里插入图片描述

异步

消息队列支持异步处理,即发送方发送消息后,并不需要等待接收方立即处理完成,而是继续执行其他任务。接收方在合适的时间从队列中获取消息进行处理。这种异步处理可以提高系统的性能和响应速度,尤其适用于处理耗时的操作。

在这里插入图片描述

好处
1:完全解耦,用MQ建立桥接
2:有独立的线程池和运行模型
3:出现了消息可能会丢失,MQ有持久化功能
4:如何保证消息的可靠性,死信队列和消息转移的等
5:如果服务器承载不了,你需要自己去写高可用,HA镜像模型高可用。
按照以上约定,用户的响应时间相当于是订单信息写入数据库的时间,也就是50毫秒。注册邮件,发送短信写入消息队列后,直接返回,因此写入消息队列的速度很快,基本可以忽略,因此用户的响应时间可能是50毫秒。因此架构改变后,系统的吞吐量提高到每秒20 QPS。比串行提高了3倍,比并行提高了两倍

2、Dokcer安装RabbitMQ

2.1安装Dokcer

  1. yum 包更新到最新
> yum update
  1. 安装软件包,yum-util提供yum-config-manager功能,另外两个是devicemapper驱动依赖的
> yum install -y yum-utils device-mapper-persistent-data lvm2
  1. 设置yum源为阿里云
> yum-config-manager --add-repo
> http://mirrors.aliyun.com/docker-ce/linux/centos/docker-ce.repo
  1. 安装docker
> yum install docker-ce-y
  1. 安装后查看docker版本
> docker-v
  1. 安装加速镜像

从阿里云获取镜像加速器:
https://cr.console.aliyun.com/cn-hangzhou/instances/mirrors

sudo mkdir -p /etc/docker
sudo tee /etc/docker/daemon.json <<-'EOF'
{"registry-mirrors": ["https://spukdfwp.mirror.aliyuncs.com"]
}
EOF
sudo systemctl daemon-reload
sudo systemctl restart docke

2.2安装rabbitmq

  1. 路径:https://www.rabbitmq.com/download.html

在这里插入图片描述

  1. 点击上图中标红线的 community Docker
    image,跳转到如下地址:https://registry.hub.docker.com/_/rabbitmq/

当前可以看到安装镜像的时候可以设置用户名,密码,ip。就不用安装完进入容器内部设置
在这里插入图片描述
3. 官网给的安装案例

$ docker run -d --hostname my-rabbit --name some-rabbit -e RABBITMQ_DEFAULT_USER=user -e RABBITMQ_DEFAULT_PASS=password rabbitmq:3-management

4.命令讲解

docker run -id --hostname my-rabbit --name=myrabbit -p 15672:15672 rabbitmq:3-management--hostname:指定容器主机名称
--name:指定容器名称
-p:将mq端口号映射到本地
-e 设置

5.修改命令创建并安装

docker run -di  --name myrabbit -e RABBITMQ_DEFAULT_USER=admin -e RABBITMQ_DEFAULT_PASS=admin -p 15672:15672 -p 5672:5672 -p 25672:25672 -p 61613:61613 -p 1883:1883 rabbitmq:3-management

6.阿里云开放上方命令 设置的端口号

-p 15672:15672 -p 5672:5672 -p 25672:25672 -p 61613:61613 -p 1883:1883

在这里插入图片描述

7.安装成功

[root@iZbp1av1izm1qqcdfa0nndZ ~]# docker images
REPOSITORY   TAG            IMAGE ID       CREATED        SIZE
rabbitmq     3-management   6c3c2a225947   7 months ago   253MB
[root@iZbp1av1izm1qqcdfa0nndZ ~]# docker ps
CONTAINER ID   IMAGE                   COMMAND                  CREATED         STATUS         PORTS                                                                                                                                                                                                                                                                               NAMES
1de1f1e10cb0   rabbitmq:3-management   "docker-entrypoint.s…"   6 minutes ago   Up 6 minutes   4369/tcp, 0.0.0.0:1883->1883/tcp, :::1883->1883/tcp, 5671/tcp, 0.0.0.0:5672->5672/tcp, :::5672->5672/tcp, 15671/tcp, 0.0.0.0:15672->15672/tcp, :::15672->15672/tcp, 0.0.0.0:25672->25672/tcp, :::25672->25672/tcp, 0.0.0.0:61613->61613/tcp, :::61613->61613/tcp, 15691-15692/tcp   myrabbit
[root@iZbp1av1izm1qqcdfa0nndZ ~]#

8.停掉手动安装的rabbimq

systemctl stop rabbitmq-server

9.启动docker的rabbitmq容器

##查看容器
[root@iZbp1av1izm1qqcdfa0nndZ ~]# docker ps
CONTAINER ID   IMAGE                   COMMAND                  CREATED         STATUS         PORTS                                                                                                                                                                                                                                                                               NAMES
1de1f1e10cb0   rabbitmq:3-management   "docker-entrypoint.s…"   9 minutes ago   Up 9 minutes   4369/tcp, 0.0.0.0:1883->1883/tcp, :::1883->1883/tcp, 5671/tcp, 0.0.0.0:5672->5672/tcp, :::5672->5672/tcp, 15671/tcp, 0.0.0.0:15672->15672/tcp, :::15672->15672/tcp, 0.0.0.0:25672->25672/tcp, :::25672->25672/tcp, 0.0.0.0:61613->61613/tcp, :::61613->61613/tcp, 15691-15692/tcp   myrabbit
##启动容器 docker start 容器id(CONTAINER ID)
[root@iZbp1av1izm1qqcdfa0nndZ ~]# docker start 1de1f1e10cb0
1de1f1e10cb0
[root@iZbp1av1izm1qqcdfa0nndZ ~]#

10.通过服务器(虚拟机ip+端口号(15672))访问RabbitMQ主页http://192.168.157.128:15672

默认登录账号和密码都是admin
在这里插入图片描述

并且在admin账号下可以通过增加用户,给用户不同角色,也就对应不同的操作权限:
在这里插入图片描述
详情如下:
在这里插入图片描述

3、RabbitMQ入门案例 - Simple 简单模式

1.实现步骤:

1:jdk1.8
2:构建一个maven工程
3:导入rabbitmq的maven依赖
4:启动rabbitmq-server服务
5:定义生产者
6:定义消费者
7:观察消息的在rabbitmq-server服务中的过程

2.构建一个maven工程
在这里插入图片描述
3.导入rabbitmq的maven依赖

<dependency><groupId>com.rabbitmq</groupId><artifactId>amqp-client</artifactId><version>5.10.0</version>
</dependency><dependency><groupId>org.springframework.amqp</groupId><artifactId>spring-amqp</artifactId><version>2.2.5.RELEASE</version>
</dependency>
<dependency><groupId>org.springframework.amqp</groupId><artifactId>spring-rabbit</artifactId><version>2.2.5.RELEASE</version>
</dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

4.启动rabbitmq-server服务

systemctl start rabbitmq-server
或者
docker start myrabbit

5、定义生产者

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
public class Producer {public static void main(String[] args) {// 1: 创建连接工厂ConnectionFactory connectionFactory = new ConnectionFactory();// 2: 设置连接属性connectionFactory.setHost("192.168.157.128");connectionFactory.setPort(5672);connectionFactory.setVirtualHost("/");connectionFactory.setUsername("admin");connectionFactory.setPassword("admin");Connection connection = null;Channel channel = null;try {// 3: 从连接工厂中获取连接connection = connectionFactory.newConnection("生产者");// 4: 从连接中获取通道channelchannel = connection.createChannel();// 5: 申明队列queue存储消息/**  如果队列不存在,则会创建*  Rabbitmq不允许创建两个相同的队列名称,否则会报错。**  @params1: queue 队列的名称*  @params2: durable 队列是否持久化*  @params3: exclusive 是否排他,即是否私有的,如果为true,会对当前队列加锁,其他的通道不能访问,并且连接自动关闭*  @params4: autoDelete 是否自动删除,当最后一个消费者断开连接之后是否自动删除消息。*  @params5: arguments 可以设置队列附加参数,设置队列的有效期,消息的最大长度,队列的消息生命周期等等。* */channel.queueDeclare("queue1", false, false, true, null);// 6: 准备发送消息的内容String message = "你好,学相伴!!!";// 7: 发送消息给中间件rabbitmq-server// @params1: 交换机exchange// @params2: 队列名称/routing// @params3: 属性配置// @params4: 发送消息的内容channel.basicPublish("", "queue1", null, message.getBytes());System.out.println("消息发送成功!");} catch (Exception ex) {ex.printStackTrace();System.out.println("发送消息出现异常...");} finally {// 7: 释放连接关闭通道if (channel != null && channel.isOpen()) {try {channel.close();} catch (Exception ex) {ex.printStackTrace();}}// 8: 关闭连接if (connection != null) {try {connection.close();} catch (Exception ex) {ex.printStackTrace();}}}}
}

1:执行发送,这个时候可以在web控制台查看到这个队列queue的信息

在这里插入图片描述

2:我们可以进行对队列的消息进行预览和测试如下:

在这里插入图片描述

3:进行预览和获取消息进行测试

在这里插入图片描述

NACK 只是做消息预览,不会吧消息从队列移除
ACK相当于手动的把消息处理了,这个时候就会把消息从队列剔除,导致消息丢失

6、定义消费者

import com.rabbitmq.client.*;import java.io.IOException;public class Consumer {public static void main(String[] args) {// 所有的中间件技术都是基于tcp/ip协议基础上构建新型协议规范,只不过rabbitmq遵循的是amqp// ip port// 1: 创建连接工厂ConnectionFactory connectionFactory = new ConnectionFactory();// 2: 设置连接属性connectionFactory.setHost("192.168.157.128");connectionFactory.setPort(5672);connectionFactory.setVirtualHost("/");connectionFactory.setUsername("admin");connectionFactory.setPassword("admin");Connection connection = null;Channel channel = null;try {// 3: 从连接工厂中获取连接connection = connectionFactory.newConnection("消费者");// 4: 从连接中获取通道channelchannel = connection.createChannel();channel.basicConsume("queue1", true, new DeliverCallback() {public void handle(String consumerTag, Delivery message) throws IOException {System.out.println("收到的消息是:" + new String(message.getBody(), "UTF-8"));}}, new CancelCallback() {public void handle(String s) throws IOException {System.out.println("接收失败了。。。");}});System.out.println("开始接收消息");System.in.read();}catch (Exception e){e.printStackTrace();System.out.println("发送消息出现异常...");}finally {// 7: 释放连接关闭通道if (channel != null && channel.isOpen()) {try {channel.close();} catch (Exception ex) {ex.printStackTrace();}}if (connection != null) {try {connection.close();} catch (Exception ex) {ex.printStackTrace();}}}}
}

消费者和生产者的区别在于,消费者是从mq中取消息,而生产者是从mq中存消息
在这里插入图片描述

4、RabbitMQ的核心组成部分

在这里插入图片描述
核心概念:

  1. Server:又称Broker ,接受客户端的连接,实现AMQP实体服务。 安装rabbitmq-server
  2. Connection:连接,应用程序与Broker的网络连接 TCP/IP/ 三次握手和四次挥手
  3. Channel:网络信道,几乎所有的操作都在Channel中进行,Channel是进行消息读写的通道,客户端可以建立对各Channel,每个Channel代表一个会话任务。
  4. Message:消息:服务与应用程序之间传送的数据,由Properties和body组成,Properties可是对消息进行修饰,比如消息的优先级,延迟等高级特性,Body则就是消息体的内容。
  5. Virtual Host 虚拟地址,用于进行逻辑隔离,最上层的消息路由,一个虚拟主机理由可以有若干个Exhange和Queueu,同一个虚拟主机里面不能有相同名字的Exchange
  6. Exchange:交换机,接受消息,根据路由键发送消息到绑定的队列。(不具备消息存储的能力)
  7. Bindings:Exchange和Queue之间的虚拟连接,binding中可以保护多个routing key.
  8. Routing key:是一个路由规则,虚拟机可以用它来确定如何路由一个特定消息。
  9. Queue:队列:也成为Message Queue,消息队列,保存消息并将它们转发给消费者。

4.1 RabbitMQ整体架构

在这里插入图片描述

4.2RabbitMQ的运行流程

在这里插入图片描述

所以发送消息的时候没有设置交换机,rabbitmq发送消息一定会有默认一个交换机,并且消息不是直接到队列当中的,而是由交换机根据路由键发送消息到绑定的队列

在这里插入图片描述
在这里插入图片描述

5、RabbitMQ的模式

5.1 发布订阅模式–fanout

特点:Fanout—发布与订阅模式,是一种广播机制,它是没有路由key的模式。

也就是只要生产者发送一条消息经过交换机加入队列中,左右的消费者都能拿到消息

在这里插入图片描述
这里就直接用web界面演示

  1. 新建一个fanout模式的交换机(让交换机代替生产者去发消息)
    在这里插入图片描述
    在这里插入图片描述

  2. 创建3个消息队列q1、q2、q3
    在这里插入图片描述

  3. 将队列绑定到交换机上
    在这里插入图片描述

  4. 由交换机代替生产者发送消息
    在这里插入图片描述

  5. 然后三个队列都会有一个交换机发来的消息
    在这里插入图片描述

  6. q1队列消息正常被消费者拾取(其他队列一样)
    在这里插入图片描述

  7. q1队列消息正常被消费者拾取之后,队列消息-1

ACK后  页面在自动会更新队列消息条目,默认5

在这里插入图片描述

在这里插入图片描述

5.2路由模式-direct模式

Direct模式是fanout模式上的一种叠加,增加了路由RoutingKey的模式
在这里插入图片描述

在这里插入图片描述

这样就可以给指定设置了路由key的队列发送消息,并且一个队列可以有多个路由key,当发送消息指定了路由key,则只有设置了相对应的路由key的队列才能接收到消息

5.3路由模式-Topic模式

Topic模式是direct模式上的一种叠加,增加了模糊路由RoutingKey的模式。

在这里插入图片描述

* 代表一级(必须有一级)
# 代表0级或者多级

在这里插入图片描述

注意最好用代码的形式来进行绑定

在实际开发中,我们既可以在RabbitMq的web界面进行交换机的创建,队列的创建,绑定路由key等等操作。
还可以在生产者代码里面通过channel.XXX的方式设置交换机,设置队列,设置路由key,等等,效果是一样的

例如下面代码:生产者(消费者也可以声明) 代码实现【交换机和队列】的声明和绑定

package com.xxx.rabbitmq.all;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;/*** RabbitMQ入门案例 - 完整的声明方式创建* 代码实现创建交换机和队列,并绑定关系* 生产者*/
public class Producer {public static void main(String[] args) {// 1: 创建连接工厂ConnectionFactory connectionFactory = new ConnectionFactory();// 2: 设置连接属性connectionFactory.setHost("121.196.153.197");connectionFactory.setPort(5672);connectionFactory.setVirtualHost("/");connectionFactory.setUsername("admin");connectionFactory.setPassword("admin");Connection connection = null;Channel channel = null;try {// 3: 从连接工厂中获取连接connection = connectionFactory.newConnection("生产者");// 4: 从连接中获取通道channelchannel = connection.createChannel();// 5: 准备发送消息的内容String message = "你好,交换机";// 6:准备交换机。取名规范 :  类型_业务模块_交换机String exchangeName = "direct_order_exchange";// 7: 定义路由keyString routeKeyOrder = "order";String routeKeyCourse = "course";// 8: 指定交换机的类型String exchangeType = "direct";// 9: 声明交换机/注册交换机// @params1: 交换机名称// @params2: 交换机类型// @params3: 是否持久化 所谓的持久化就是值:交换机不会随着服务器的重启造成丢失,如果是true代表不丢失,false重启丢失channel.exchangeDeclare(exchangeName, exchangeType, true);// 10: 声明队列/注册队列// @params1: 队列名称// @params2: 是否持久化// @params3: 是不是排他性,是否是独占独立// @params4: 是不是自动删除 随着最后一个消费者消息完毕消息以后是否把队列自动删除// @params5: 是不是有参数 参数携带可能会引发headers模式channel.queueDeclare("queue5", true, false, false, null);channel.queueDeclare("queue6", true, false, false, null);channel.queueDeclare("queue7", true, false, false, null);// 11: 绑定队列// @params1: 队列名称// @params2: 交换机名称// @params3: routeKeychannel.queueBind("queue5", exchangeName, routeKeyOrder);channel.queueBind("queue6", exchangeName, routeKeyOrder);channel.queueBind("queue7", exchangeName, routeKeyCourse);// 12: 发送消息给中间件rabbitmq-server// @params1: 交换机exchange// @params2: 队列名称// @params3: 属性配置// @params4: 发送消息的内容channel.basicPublish(exchangeName, routeKeyOrder, null, message.getBytes());System.out.println("消息发送成功!");} catch (Exception ex) {ex.printStackTrace();System.out.println("发送消息出现异常...");} finally {// 13: 释放连接关闭通道if (channel != null && channel.isOpen()) {try {channel.close();} catch (Exception ex) {ex.printStackTrace();}}if (connection != null) {try {connection.close();} catch (Exception ex) {ex.printStackTrace();}}}}
}

消费者去队列拿消息:

package com.xxx.rabbitmq.all;import com.rabbitmq.client.*;import java.io.IOException;/*** RabbitMQ入门案例 - 完整的声明方式创建* 消费者*/
public class Consumer {private static Runnable runnable = new Runnable() {public void run() {// 1: 创建连接工厂ConnectionFactory connectionFactory = new ConnectionFactory();// 2: 设置连接属性connectionFactory.setHost("121.196.153.197");connectionFactory.setPort(5672);connectionFactory.setVirtualHost("/");connectionFactory.setUsername("admin");connectionFactory.setPassword("admin");//获取队列的名称final String queueName = Thread.currentThread().getName();Connection connection = null;Channel channel = null;try {// 3: 从连接工厂中获取连接connection = connectionFactory.newConnection();// 4: 从连接中获取通道channelchannel = connection.createChannel();// 5: 申明队列queue存储消息/**  如果队列不存在,则会创建*  Rabbitmq不允许创建两个相同的队列名称,否则会报错。**  @params1: queue 队列的名称*  @params2: durable 队列是否持久化*  @params3: exclusive 是否排他,即是否私有的,如果为true,会对当前队列加锁,其他的通道不能访问,并且连接自动关闭*  @params4: autoDelete 是否自动删除,当最后一个消费者断开连接之后是否自动删除消息。*  @params5: arguments 可以设置队列附加参数,设置队列的有效期,消息的最大长度,队列的消息生命周期等等。* */// 这里如果queue已经被创建过一次了,可以不需要定义//channel.queueDeclare("queue1", false, false, false, null);// 6: 定义接受消息的回调Channel finalChannel = channel;finalChannel.basicConsume(queueName, true, new DeliverCallback() {@Overridepublic void handle(String s, Delivery delivery) throws IOException {System.out.println(queueName + ":收到消息是:" + new String(delivery.getBody(), "UTF-8"));}}, new CancelCallback() {@Overridepublic void handle(String s) throws IOException {}});System.out.println(queueName + ":开始接受消息");System.in.read();} catch (Exception ex) {ex.printStackTrace();System.out.println("发送消息出现异常...");} finally {// 7: 释放连接关闭通道if (channel != null && channel.isOpen()) {try {channel.close();} catch (Exception ex) {ex.printStackTrace();}}if (connection != null && connection.isOpen()) {try {connection.close();} catch (Exception ex) {ex.printStackTrace();}}}}};public static void main(String[] args) {// 启动三个线程去执行new Thread(runnable, "queue5").start();new Thread(runnable, "queue6").start();new Thread(runnable, "queue7").start();}}

5.4轮询模式 - Work模式

5.4.1Work模式 - 轮询模式(Round-Robin)

轮询模式的分发:一个消费者一条,按均分配;

在这里插入图片描述

生产者:

package com.xuexiangban.rabbitmq.work.lunxun;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
/*** @author: 学相伴-飞哥* @description: Producer 简单队列生产者* @Date : 2021/3/2*/
public class Producer {public static void main(String[] args) {// 1: 创建连接工厂ConnectionFactory connectionFactory = new ConnectionFactory();// 2: 设置连接属性connectionFactory.setHost("192.168.157.128");connectionFactory.setPort(5672);connectionFactory.setVirtualHost("/");connectionFactory.setUsername("admin");connectionFactory.setPassword("admin");Connection connection = null;Channel channel = null;try {// 3: 从连接工厂中获取连接connection = connectionFactory.newConnection("生产者");// 4: 从连接中获取通道channelchannel = connection.createChannel();// 6: 准备发送消息的内容//===============================end topic模式==================================for (int i = 1; i <= 20; i++) {//消息的内容String msg = "学相伴:" + i;// 7: 发送消息给中间件rabbitmq-server// @params1: 交换机exchange// @params2: 队列名称/routingkey// @params3: 属性配置// @params4: 发送消息的内容channel.basicPublish("", "queue1", null, msg.getBytes());}System.out.println("消息发送成功!");} catch (Exception ex) {ex.printStackTrace();System.out.println("发送消息出现异常...");} finally {// 7: 释放连接关闭通道if (channel != null && channel.isOpen()) {try {channel.close();} catch (Exception ex) {ex.printStackTrace();}}if (connection != null) {try {connection.close();} catch (Exception ex) {ex.printStackTrace();}}}}
}

消费者work1:

package com.xuexiangban.rabbitmq.work.lunxun;
import com.rabbitmq.client.*;
import java.io.IOException;
/*** @author: 学相伴-飞哥* @description: Consumer* @Date : 2021/3/2*/
public class Work1 {public static void main(String[] args) {// 1: 创建连接工厂ConnectionFactory connectionFactory = new ConnectionFactory();// 2: 设置连接属性connectionFactory.setHost("192.168.157.128");connectionFactory.setPort(5672);connectionFactory.setVirtualHost("/");connectionFactory.setUsername("admin");connectionFactory.setPassword("admin");Connection connection = null;Channel channel = null;try {// 3: 从连接工厂中获取连接connection = connectionFactory.newConnection("消费者-Work1");// 4: 从连接中获取通道channelchannel = connection.createChannel();// 5: 申明队列queue存储消息/**  如果队列不存在,则会创建*  Rabbitmq不允许创建两个相同的队列名称,否则会报错。**  @params1: queue 队列的名称*  @params2: durable 队列是否持久化*  @params3: exclusive 是否排他,即是否私有的,如果为true,会对当前队列加锁,其他的通道不能访问,并且连接自动关闭*  @params4: autoDelete 是否自动删除,当最后一个消费者断开连接之后是否自动删除消息。*  @params5: arguments 可以设置队列附加参数,设置队列的有效期,消息的最大长度,队列的消息生命周期等等。* */// 这里如果queue已经被创建过一次了,可以不需要定义
//            channel.queueDeclare("queue1", false, false, false, null);// 同一时刻,服务器只会推送一条消息给消费者// 6: 定义接受消息的回调Channel finalChannel = channel;finalChannel.basicQos(1);finalChannel.basicConsume("queue1", true, new DeliverCallback() {@Overridepublic void handle(String s, Delivery delivery) throws IOException {try{System.out.println("Work1-收到消息是:" + new String(delivery.getBody(), "UTF-8"));Thread.sleep(2000);}catch(Exception ex){ex.printStackTrace();}}}, new CancelCallback() {@Overridepublic void handle(String s) throws IOException {}});System.out.println("Work1-开始接受消息");System.in.read();} catch (Exception ex) {ex.printStackTrace();System.out.println("发送消息出现异常...");} finally {// 7: 释放连接关闭通道if (channel != null && channel.isOpen()) {try {channel.close();} catch (Exception ex) {ex.printStackTrace();}}if (connection != null && connection.isOpen()) {try {connection.close();} catch (Exception ex) {ex.printStackTrace();}}}}
}

work2

package com.xuexiangban.rabbitmq.work.lunxun;
import com.rabbitmq.client.*;
import java.io.IOException;
/*** @author: 学相伴-飞哥* @description: Consumer* @Date : 2021/3/2*/
public class Work2 {public static void main(String[] args) {// 1: 创建连接工厂ConnectionFactory connectionFactory = new ConnectionFactory();// 2: 设置连接属性connectionFactory.setHost("192.168.157.128");connectionFactory.setPort(5672);connectionFactory.setVirtualHost("/");connectionFactory.setUsername("admin");connectionFactory.setPassword("admin");Connection connection = null;Channel channel = null;try {// 3: 从连接工厂中获取连接connection = connectionFactory.newConnection("消费者-Work2");// 4: 从连接中获取通道channelchannel = connection.createChannel();// 5: 申明队列queue存储消息/**  如果队列不存在,则会创建*  Rabbitmq不允许创建两个相同的队列名称,否则会报错。**  @params1: queue 队列的名称*  @params2: durable 队列是否持久化*  @params3: exclusive 是否排他,即是否私有的,如果为true,会对当前队列加锁,其他的通道不能访问,并且连接自动关闭*  @params4: autoDelete 是否自动删除,当最后一个消费者断开连接之后是否自动删除消息。*  @params5: arguments 可以设置队列附加参数,设置队列的有效期,消息的最大长度,队列的消息生命周期等等。* */// 这里如果queue已经被创建过一次了,可以不需要定义//channel.queueDeclare("queue1", false, true, false, null);// 同一时刻,服务器只会推送一条消息给消费者//channel.basicQos(1);// 6: 定义接受消息的回调Channel finalChannel = channel;finalChannel.basicQos(1);finalChannel.basicConsume("queue1", true, new DeliverCallback() {@Overridepublic void handle(String s, Delivery delivery) throws IOException {try{System.out.println("Work2-收到消息是:" + new String(delivery.getBody(), "UTF-8"));Thread.sleep(200);}catch(Exception ex){ex.printStackTrace();}}}, new CancelCallback() {@Overridepublic void handle(String s) throws IOException {}});System.out.println("Work2-开始接受消息");System.in.read();} catch (Exception ex) {ex.printStackTrace();System.out.println("发送消息出现异常...");} finally {// 7: 释放连接关闭通道if (channel != null && channel.isOpen()) {try {channel.close();} catch (Exception ex) {ex.printStackTrace();}}if (connection != null && connection.isOpen()) {try {connection.close();} catch (Exception ex) {ex.printStackTrace();}}}}
}

work1和work2的消息处理能力不同,但是最后处理的消息条数相同,是“按均分配”。

往队列发送20条消息,
在这里插入图片描述

结果就是轮询给消费者拿消息,即便有的消费者消费很快(也只能按照新顺序拿消息),也只能按照轮询一个一个拿,

也就是说,不会因为某个消费者所在的服务器满,而导致少消费,一定是公平消费
在这里插入图片描述

在这里插入图片描述

5.4.1Work模式 - 公平分发模式(Round-Robin)

根据消费者的消费能力进行公平分发,处理快的处理的多,处理慢的处理的少;按劳分配

相比较轮询模式,公平分发的不同在于:修改应答方式为手动
在这里插入图片描述

qos设置为1,就代表消费者拿到cpu的执行权就每次从只拿走一条消息,一条一条的拿。若不设置,就是默认轮询拿一条
所以根据队列堆积的消息条数以及内存和磁盘空间来合理设置qos

在这里插入图片描述
这个时候,性能好的消费者就会消费得多,而性能差的消费者就消费得少,能者多劳
在这里插入图片描述
在这里插入图片描述

更新中------
参考来自:狂神

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

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

相关文章

大数据-玩转数据-Flink App市场推广统计

一、说明 电商网站中已经有越来越多的用户来自移动端&#xff0c;相比起传统浏览器的登录方式&#xff0c;手机APP成为了更多用户访问电商网站的首选。对于电商企业来说&#xff0c;一般会通过各种不同的渠道对自己的APP进行市场推广&#xff0c;而这些渠道的统计数据&#xf…

iOS自定义下拉刷新控件

自定义下拉刷新控件 概述 用了很多的别人的下拉刷新控件&#xff0c;想写一个玩玩&#xff0c;自定义一个在使用的时候也会比较有意思。使应用更加的灵动一些&#xff0c;毕竟谁不喜欢各种动画恰到好处的应用呢。 使用方式如下&#xff1a; tableview.refreshControl XRef…

小程序多图片组合

目录 子组件 index.js 子组件 index.wxml 子组件 index.wxss 父组件引用&#xff1a; 子组件&#xff1a;preview-image 子组件 index.js Component({properties: {previewData: {type: Array,default: [],observer: function (newVal, oldVal) {console.log(newVal, ol…

回归预测 | MATLAB实现GWO-SVM灰狼优化算法优化支持向量机多输入单输出回归预测(多指标,多图)

回归预测 | MATLAB实现GWO-SVM灰狼优化算法优化支持向量机多输入单输出回归预测&#xff08;多指标&#xff0c;多图&#xff09; 目录 回归预测 | MATLAB实现GWO-SVM灰狼优化算法优化支持向量机多输入单输出回归预测&#xff08;多指标&#xff0c;多图&#xff09;效果一览基…

图论基础和表示(Java 实例代码)

目录 图论基础和表示 一、概念及其介绍 二、适用说明 三、图的表达形式 Java 实例代码 src/runoob/graph/DenseGraph.java 文件代码&#xff1a; src/runoob/graph/SparseGraph.java 文件代码&#xff1a; 图论基础和表示 一、概念及其介绍 图论(Graph Theory)是离散数…

MySQL基础篇(二)

DML 定义&#xff1a;Data Manipulation Language、数据操作语言&#xff08;增删改&#xff09; 添加数据&#xff08;INSERT&#xff09;修改数据&#xff08;UPDATE&#xff09;删除数据&#xff08;DELETE&#xff09; 添加数据&#xff08;INSERT&#xff09; 给指定的…

Nginx:网站服务

nginx&#xff1a;一个高性能、轻量级的web服务软件 1、稳定性高&#xff08;没有apache稳&#xff09; 2、系统资源消耗低&#xff08;处理http请求的并发能力很高&#xff0c;单台物理服务器可以处理3万到5万个并发请求&#xff09; 稳定&#xff1a;一般在企业中&#xff…

系统架构设计专业技能 · 信息安全技术

系列文章目录 系统架构设计专业技能 网络技术&#xff08;三&#xff09; 系统架构设计专业技能 系统安全分析与设计&#xff08;四&#xff09;【系统架构设计师】 系统架构设计高级技能 软件架构设计&#xff08;一&#xff09;【系统架构设计师】 系统架构设计高级技能 …

<数据结构与算法>堆的应用二叉树的链式实现

目录 前言 一、堆的应用 1. 堆排序 1.1 排升序&#xff0c;建大堆 1.2 时间复杂度计算 2. Top k问题 二、 二叉树的链式实现 1. 二叉树的遍历 2. 二叉树基础OJ 2.2 100. 相同的树 总结 前言 学习完堆的数据结构&#xff0c;我们要清楚&#xff0c;它虽然实现了排序功能&am…

售后服务管理软件怎么选择?售后服务管理系统有什么用?

随着企业信息化发展&#xff0c;越来越多的企业纷纷选择售后服务管理软件来服务客户和进行内部人员管理。借助这款软件&#xff0c;企业能够高效地满足客户提出的需求&#xff0c;并提高客户对售后服务的满意度。售后服务通常涉及客户、客服、维修师傅和服务管理人员等各种角色…

用C++/JS/Python/Java代码描述秋天的味道

前言 秋天是一个充满诗意和浪漫的季节&#xff0c;它带来了清新、芬芳和美食的味道。让我们一起探索如何用编程语言来写出秋天味道的代码吧&#xff01;无论是C、JavaScript、Python还是Java&#xff0c;以下是几个简单的步骤来帮助你创造出充满秋天味道的代码&#xff1a; …

Redis缓存问题(穿透, 击穿, 雪崩, 污染, 一致性)

目录 1.什么是Redis缓存问题&#xff1f; 2.缓存穿透 3.缓存击穿 4.缓存雪崩 5.缓存污染&#xff08;或满了&#xff09; 5.1 最大缓存设置多大 5.2 缓存淘汰策略 6.数据库和缓存一致性 6.1 4种相关模式 6.2 方案&#xff1a;队列重试机制 6.3 方案&#xff1a;异步更新缓…

[C++11]

文章目录 1. 自动类型推导1.1 auto1.1.1 推导规则1.1.2 auto的限制1.1.3 auto的应用1.1.4 范围for 1.2 decltype1.2.1 推导规则1.2.2 decltype的应用 1.3 返回类型后置 2.可调用对象包装器、绑定器2.1 可调用对象包装器2.1.1 基本用法2.1.2 作为回调函数使用 2.2 绑定器 3. usi…

idea创建javaweb项目,jboss下没有web application

看看下图这个地方有没有web application

C++_模板进阶_非类型模板参数_模板特化_分离编译

一、非类型模板参数 模板参数&#xff0c;分为类型形参和非类型形参。 类型形参就是在模板中跟在typename和class之后的参数类型名称&#xff0c;非类型形参就是用一个常量作为类模板或者函数模板的一个参数&#xff0c;在类模板和函数模板中&#xff0c;可以将该参数当作常量…

【JUC系列-01】深入理解JMM内存模型的底层实现原理

一&#xff0c;深入理解JMM内存模型 1&#xff0c;什么是可见性 在谈jmm的内存模型之前&#xff0c;先了解一下并发并发编程的三大特性&#xff0c;分别是&#xff1a;可见性&#xff0c;原子性&#xff0c;有序性。可见性指的就是当一个线程修改某个变量的值之后&#xff0c…

卷积神经网络全解:(AlexNet/VGG/ GoogLeNet/LeNet/ResNet/卷积/激活/池化/全连接)、现代卷积神经网络、经典卷积神经网络

CNN&#xff0c;卷积神经网络&#xff0c;Convolution Neural Network 卷积计算公式&#xff1a;N &#xff08;W-F2p&#xff09;/s1 这个公式每次都得看看&#xff0c;不能忘 1 经典网络 按照时间顺序 1.1 LeNet LeNet是 Yann LeCun在1998年提出&#xff0c;用于解决手…

Lua 数据结构

一、Lua 中的数据结构 Lua 中并没有像 java、kotlin 语言有专门的数组、列表、集合等数据结构的类&#xff0c;而只提供 table 类型&#xff0c;但他很强大而且也很灵活&#xff0c;而且也在很多场景中天然的就解决了如何对接和使用的问题&#xff08;例如模块的引入&#xff…

MYSQL完全卸载、安装与账号创建、权限控制

一、卸载mysql CentOS 卸载 MySQL 1. 查看安装情况 使用以下命令查看当前安装mysql情况&#xff0c;查找以前是否装有mysql rpm -qa|grep -i mysql这里显示我安装的 MySQL 服务有有&#xff1a; 2. 停止 mysql 服务、删除之前安装的 mysql 删除命令&#xff1a;rpm -e –n…

C++笔记之条件变量(Condition Variable)与cv.wait 和 cv.wait_for的使用

C笔记之条件变量&#xff08;Condition Variable&#xff09;与cv.wait 和 cv.wait_for的使用 参考博客&#xff1a;C笔记之各种sleep方法总结 code review! 文章目录 C笔记之条件变量&#xff08;Condition Variable&#xff09;与cv.wait 和 cv.wait_for的使用1.条件变量&…