怎么设置网站名称,图文视频怎么制作,全国感染的最新数据统计,列举网站开发常用的工具文章目录 十.RabbitMQ10.1 简单队列实现10.2 Work 模式#xff08;工作队列#xff09;10.3 公平分发10.4 RabbitMQ 消息应答与消息持久化消息应答概念配置 消息持久化概念配置 十.RabbitMQ
10.1 简单队列实现
简单队列通常指的是一个基本的消息队列#xff0c;它可以用于… 文章目录 十.RabbitMQ10.1 简单队列实现10.2 Work 模式工作队列10.3 公平分发10.4 RabbitMQ 消息应答与消息持久化消息应答概念配置 消息持久化概念配置 十.RabbitMQ
10.1 简单队列实现
简单队列通常指的是一个基本的消息队列它可以用于在生产者生产消息的一方和消费者消费消息的一方之间传递消息。 新创建Springboot项目
引入依赖
dependencygroupIdcom.rabbitmq/groupIdartifactIdamqp-client/artifactIdversion5.7.1/version/dependency连接工具类
public class ConnectionUtils
{public static Connection getConnection(){try{Connection connection null;//定义一个连接工厂ConnectionFactory factory new ConnectionFactory();//设置服务端地址域名地址/ipfactory.setHost(127.0.0.1);//设置服务器端口号factory.setPort(5672);//设置虚拟主机(相当于数据库中的库)factory.setVirtualHost(/);//设置用户名factory.setUsername(guest);//设置密码factory.setPassword(guest);connection factory.newConnection();return connection;}catch (Exception e){return null;}}
}创建生产者
public class Provider01 {public static void main(String[] args) {try {System.out.println(--------生产者-------);// 获取连接Connection conn ConnectionUtils.getConnection();//创建通道Channel channel conn.createChannel();// 创建队列// 队列名称是否持久化是否排他是否自动删除其他参数channel.queueDeclare(test4072, false, false, false, null);// 定义发送信息String msg hello rabbitmq-kwh;// 发送数据channel.basicPublish(, test4072, null, msg.getBytes());System.out.println(发送成功....);// 关闭资源channel.close();conn.close();} catch (Exception e) {e.printStackTrace();}}
}创建消费者
public class Consumer01 {public static void main(String[] args) {try {System.out.println(消费者);// 获取连接Connection conn ConnectionUtils.getConnection();//创建通道Channel channel conn.createChannel();// 创建队列(有就直接连接。没有则创建)// 队列名称是否持久化是否排他是否自动删除其他参数channel.queueDeclare(test4072, false, false, false, null);// 消费者消费消息DefaultConsumer consumer new DefaultConsumer(channel){Override //一旦有消息进入 将触发public void handleDelivery(String consumerTag,Envelope envelope, AMQP.BasicProperties properties,byte[] body) throws IOException {String str new String(body,utf-8);System.out.println(msg接收str);}};// 监听队列channel.basicConsume(test4072,true,consumer);}catch (Exception e) {e.printStackTrace();}}
}10.2 Work 模式工作队列
工作队列的概念
工作队列模式生产者将任务发送到队列中多个消费者从队列中取出任务并并行处理。这意味着多个消费者可以共同工作来处理同一个队列中的任务。负载均衡每个消费者只处理一个任务消息通过增加消费者数量任务的处理可以并行化提高整体处理能力。
工作队列的特点
任务分配RabbitMQ 将队列中的任务消息分配给可用的消费者通常是按照“轮询”或“平衡”方式分配即消费者可以公平地处理任务。任务处理并行化多个消费者可以并行地从同一个队列中消费消息从而实现任务的并行处理减轻单一消费者的负担。消息丢失的风险低通过合理配置队列和消息持久化机制即使 RabbitMQ 重启也能确保任务消息不丢失。 生产者
(只是在简单队列中的生产者中循环发送了信息。)
/*** Work 模式工作队列*/
public class Provider01 {public static void main(String[] args) {try {System.out.println(--------生产者-------);// 获取连接Connection conn ConnectionUtils.getConnection();//创建通道Channel channel conn.createChannel();// 创建队列// 队列名称是否持久化是否排他是否自动删除其他参数channel.queueDeclare(test4072, false, false, false, null);for (int i 0; i 50; i) {// 定义发送信息String msg hello rabbitmq-kwhi;// 发送数据channel.basicPublish(, test4072, null, msg.getBytes());Thread.sleep(1000);}System.out.println(发送成功....);// 关闭资源channel.close();conn.close();} catch (Exception e) {e.printStackTrace();}}
}消费者01
public class Consumer01 {public static void main(String[] args) {try {System.out.println(消费者01);// 获取连接Connection conn ConnectionUtils.getConnection();//创建通道Channel channel conn.createChannel();// 创建队列(有就直接连接。没有则创建)// 队列名称是否持久化是否排他是否自动删除其他参数channel.queueDeclare(test4072, false, false, false, null);// 消费者消费消息DefaultConsumer consumer new DefaultConsumer(channel){Override //一旦有消息进入 将触发public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {String str new String(body,utf-8);System.out.println(msg接收str);}};// 监听队列channel.basicConsume(test4072,true,consumer);}catch (Exception e) {e.printStackTrace();}}
}消费者02
public class Consumer02 {public static void main(String[] args) {try {System.out.println(消费者02);// 获取连接Connection conn ConnectionUtils.getConnection();//创建通道Channel channel conn.createChannel();// 创建队列(有就直接连接。没有则创建)// 队列名称是否持久化是否排他是否自动删除其他参数channel.queueDeclare(test4072, false, false, false, null);// 消费者消费消息DefaultConsumer consumer new DefaultConsumer(channel){Override //一旦有消息进入 将触发public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,byte[] body) throws IOException {String str new String(body,utf-8);System.out.println(msg接收str);}};// 监听队列channel.basicConsume(test4072,true,consumer);}catch (Exception e) {e.printStackTrace();}}
}. 消费者 1 与消费者 2 处理的数据条数一样。 . 消费者 1 偶数 ;消费者 2 奇数 这种方式叫轮询分发Round-robin。 10.3 公平分发
指消息被均匀地分配给多个消费者以便各个消费者的负载大致相等。通过这种方式RabbitMQ 旨在避免某些消费者过载而其他消费者空闲的情况。 在10.2 中现在有 2 个消费者所有的奇数的消息都是繁忙的而偶数则是轻松的。按照轮询的方式奇数的任务交给了第一个消费者所以一直在忙个不停。偶数的任务交给另一个消费者则立即完成任务然后闲得不行。而 RabbitMQ 则是不了解这些的。这是因为当消息进入队列RabbitMQ 就会分派消息。它不看消费者为应答的数目只是盲目的将消息发给轮询指定的消费者。
改造生产者
/*
同一时刻服务器只会发一条消息给消费者
1 限制发送给消费者不得超过一条消息
*/
channel.basicQos(1);/*** 公平分发*/
public class Provider01 {public static void main(String[] args) {try {System.out.println(--------生产者-------);// 获取连接Connection conn ConnectionUtils.getConnection();//创建通道Channel channel conn.createChannel();//同一时刻服务器只会发一条消息给消费者channel.basicQos(1);// 创建队列// 队列名称是否持久化是否排他是否自动删除其他参数channel.queueDeclare(test4072, false, false, false, null);for (int i 0; i 50; i) {// 定义发送信息String msg hello rabbitmq-kwhi;// 发送数据channel.basicPublish(, test4072, null, msg.getBytes());Thread.sleep(1000);}System.out.println(发送成功....);// 关闭资源channel.close();conn.close();} catch (Exception e) {e.printStackTrace();}}}消费者01
(在10.2 中消费者的基础上只添加 channel.basicQos(1);,限制每次只消费一个消息)
public class Consumer01 {public static void main(String[] args) {try {System.out.println(消费者01);// 获取连接Connection conn ConnectionUtils.getConnection();//创建通道Channel channel conn.createChannel();//限制每次只消费一个消息channel.basicQos(1);// 创建队列(有就直接连接。没有则创建)// 队列名称是否持久化是否排他是否自动删除其他参数channel.queueDeclare(test4072, false, false, false, null);// 消费者消费消息DefaultConsumer consumer new DefaultConsumer(channel){Override //一旦有消息进入 将触发public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,byte[] body) throws IOException {String str new String(body,utf-8);System.out.println(envelope.getDeliveryTag()msg接收str);// 休眠一秒钟try {Thread.sleep(1000);} catch (Exception e) {e.printStackTrace();}finally {// 手动确认消息// 第一个参数消息的序号// 第二个参数是否批量,false 单条消息应答 当为 true 时批量应答channel.basicAck(envelope.getDeliveryTag(),false);}}};// 监听队列// 自动应答设为 falsechannel.basicConsume(test4072,false,consumer);}catch (Exception e) {e.printStackTrace();}}
}消费者02
同消费者01 消费者 1 休眠 1 秒消费者 2 休眠 2 秒。 分别设置接收消息数手动反馈关闭自动应答 10.4 RabbitMQ 消息应答与消息持久化
消息应答
概念
**消息应答ack**是 RabbitMQ 中一个重要的机制用于保证消息在被消费者处理后得以正确确认确保消息不会丢失。如果消费者成功处理了消息应该发送一个确认应答如果消费者遇到问题或失败则可以选择拒绝该消息甚至重新放回队列供其他消费者处理。
应答类型
**自动应答auto-ack**自动应答是默认设置消费者从队列中获取消息后RabbitMQ 会立即认为该消息已经被成功处理即使消费者并未真正处理完成。在这种模式下消息会在被消费后立即从队列中删除而无需消费者确认。这种模式的缺点是如果消费者在处理消息时崩溃消息会丢失。**手动应答manual ack**消费者处理完消息后需要显式地发送确认应答通知 RabbitMQ 该消息已经处理完成。这样如果消费者没有发送确认应答RabbitMQ 会重新将消息发送给其他消费者。
配置
// 监听队列
// 参数2自动应答设为 false; true:开启自动应答
channel.basicConsume(test4072,false,consumer);参数2为true时自动确认
只要消息从队列中获取无论消费者获取到消息后是否成功消费都认为是消息已经成功消费。一旦rabbitmq 将消息分发给消费者就会从内存中删除。会丢失数据消息
参数2为false时手动确认
消费者从队列中获取消息后服务器会将该消息标记为不可用状态等待消费者的反馈如果消费者一直没有反馈那么该消息将一直处于不可用状态。如果有一个消费者挂掉就会交付给其他消费者。手动告诉 rabbitmq 消息处理完成后rabbitmq 删除内存中的消息。
反馈
//手动回馈
channel.basicAck(envelope.getDeliveryTag(),false);使用 Nack 让消息回到队列中
// 处理条数; 是否批量处理 ;是否放回队列 false 丢弃
channel.basicNack(envelope.getDeliveryTag(),false,true);生产者
/*** 消息应答*/
public class Provider01 {public static void main(String[] args) {try {System.out.println(--------生产者-------);// 获取连接Connection conn ConnectionUtils.getConnection();//创建通道Channel channel conn.createChannel();//同一时刻服务器只会发一条消息给消费者channel.basicQos(1);// 创建队列// 队列名称是否持久化是否排他是否自动删除其他参数channel.queueDeclare(test4072, false, false, false, null);for (int i 0; i 50; i) {// 定义发送信息String msg hello rabbitmq-kwhi;// 发送数据channel.basicPublish(, test4072, null, msg.getBytes());Thread.sleep(1000);}System.out.println(发送成功....);// 关闭资源channel.close();conn.close();} catch (Exception e) {e.printStackTrace();}}}消费者01
public class Consumer01 {public static void main(String[] args) {try {System.out.println(消费者01);// 获取连接Connection conn ConnectionUtils.getConnection();//创建通道Channel channel conn.createChannel();//限制每次只消费一个消息防止通道中消息阻塞channel.basicQos(1);// 创建队列(有就直接连接。没有则创建)// 队列名称是否持久化是否排他是否自动删除其他参数channel.queueDeclare(test4072, false, false, false, null);// 消费者消费消息DefaultConsumer consumer new DefaultConsumer(channel){Override //一旦有消息进入 将触发public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,byte[] body) throws IOException {String str ;try {str new String(body,utf-8);if(envelope.getDeliveryTag()3){int i1/0;}System.out.println(envelope.getDeliveryTag()消费者01msg接收str);//手动应答 处理完了// 手动确认消息即手动反馈// 第一个参数消息的序号// 第二个参数是否批量,false 单条消息应答 ;当为 true 时批量应答channel.basicAck(envelope.getDeliveryTag(),false);}catch(Exception e){// e.printStackTrace();System.out.println(消费者01处理第envelope.getDeliveryTag()条时报错消息内容为str);//手动应答 报错了// 第一个参数消息的序号// 第二个参数是否批量,false 单条消息应答 当为 true 时批量应答// 第三个参数是否放回队列 ;false丢弃 ,true 放回队列channel.basicNack(envelope.getDeliveryTag(),false,true);}// 休眠一秒钟try {Thread.sleep(1000);} catch (Exception e) {e.printStackTrace();}}};// 监听队列// 参数2自动应答设为 false; true:开启自动应答channel.basicConsume(test4072,false,consumer);}catch (Exception e) {e.printStackTrace();}}
}消费者02同消费者01 消息持久化
概念
RabbitMQ 的持久化机制是确保消息和队列在系统崩溃、重启或其他故障情况下不会丢失的关键功能。确保消息不会丢失需要做两件事将队列和消息都标记为持久化。
配置
持久化队列
// 创建队列
// 队列名称是否持久化(队列)是否排他是否自动删除其他参数channel.queueDeclare(test4072, true, false, false, null);消息持久化
// 发送数据
// MessageProperties.PERSISTENT_TEXT_PLAIN:持久化消息
//设置生成者发送消息为持久化信息要求保存到硬盘上保存在内存中
//MessageProperties.PERSISTENT_TEXT_PLAIN,指令完成持久化
channel.basicPublish(, test4072, MessageProperties.PERSISTENT_TEXT_PLAIN, msg.getBytes());如果改动队列参数配置需要删除原有的队列重新建因为在 rabbitmq 是不允许重新定义一个已存在的队列。 生产者
/*** 消息持久化*/
public class Provider01 {public static void main(String[] args) {try {System.out.println(--------生产者-------);// 获取连接Connection conn ConnectionUtils.getConnection();//创建通道Channel channel conn.createChannel();//同一时刻服务器只会发一条消息给消费者channel.basicQos(1);// 创建队列// 队列名称是否持久化(队列)是否排他是否自动删除其他参数channel.queueDeclare(test4072, true, false, false, null);for (int i 0; i 50; i) {// 定义发送信息String msg hello rabbitmq-kwhi;// 发送数据// MessageProperties.PERSISTENT_TEXT_PLAIN:持久化消息channel.basicPublish(, test4072, MessageProperties.PERSISTENT_TEXT_PLAIN, msg.getBytes());Thread.sleep(1000);}System.out.println(发送成功....);// 关闭资源channel.close();conn.close();} catch (Exception e) {e.printStackTrace();}}
}
