当前位置：首页 > news >正文

RabbitMQ-API

news 2025/7/8 21:33:12

这里写目录标题

Hello word 模式
- 添加依赖
- 生产者
- 消费者
- 获取信道工具类
Work Queues模式
- 消费者代码 C1
- 开启多线程运行
- 启动消费者代码 C2
- 生产者代码
消息应答
- 自动应答
- 消息应答的方法
- Multiple 的解释
- 消息自动重新入队
- 消息手动应答代码
- 消费者API
队列持久化
消息持久化
不公平分发
消息预取值
确认发布
- 单个确认发布
- 批量确认发布
- 异步批量确认发布
- 如何处理异步未确认消息

Hello word 模式

“ P”是我们的生产者，“ C”是我们的消费者。中间的框是一个队列-RabbitMQ 代表使用者保留的消息缓冲区
在这里插入图片描述

添加依赖

<!--rabbitmq 依赖客户端--><dependency><groupId>com.rabbitmq</groupId><artifactId>amqp-client</artifactId><version>5.8.0</version></dependency><!--操作文件流的一个依赖--><dependency><groupId>commons-io</groupId><artifactId>commons-io</artifactId><version>2.6</version></dependency>

生产者

package com.wlj.rabbitmq.one;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.wlj.rabbitmq.util.MQUtil;import java.io.IOException;
import java.util.concurrent.TimeoutException;
/***@创建人 wlj*@创建时间 2023/7/20*@描述 消息生产者 Hello World简单队列模式*/
public class Producer {private static String QUEUE_NAME="hello";public static void main(String[] args) {//创建一个连接工厂ConnectionFactory factory = new ConnectionFactory();//工厂IP连接MQ的队列factory.setHost("localhost");//设置用户名factory.setUsername("guest");factory.setPassword("guest");try {//创建连接Connection connection = factory.newConnection();//获取信道Channel channel = connection.createChannel();//生成一个队列/*** 生成一个队列* 1.队列名称* 2.队列里面的消息是否持久化 默认消息存储在内存中* 3.该队列是否只供一个消费者进行消费 是否进行共享 true 可以多个消费者消费* 4.是否自动删除 最后一个消费者端开连接以后 该队列是否自动删除 true 自动删除* 5.其他参数*/channel.queueDeclare(QUEUE_NAME,false,false,false,null);String message="hello world2";/*** 发送一个消息* 1.发送到那个交换机* 2.路由的 key 是哪个* 3.其他的参数信息* 4.发送消息的消息体*/channel.basicPublish("",QUEUE_NAME,null,message.getBytes());System.out.println("消息发送完毕");} catch (IOException e) {e.printStackTrace();} catch (TimeoutException e) {e.printStackTrace();}}}

消费者

package com.wlj.rabbitmq.one;import com.rabbitmq.client.*;
/***@创建人 wlj*@创建时间 2023/7/20*@描述 消息消费者 Hello World简单队列模式*/
public class Consumer {private static String QUEUE_NAME="hello";public static void main(String[] args) {//创建有一个连接工厂ConnectionFactory factory = new ConnectionFactory();//工厂IP连接MQ的队列factory.setHost("localhost");//设置用户名factory.setUsername("guest");factory.setPassword("guest");//创建连接try {Connection connection = factory.newConnection();//获取信道Channel channel = connection.createChannel();/***消费者消费消息* 1.消费那个队列* 2. 消费成功之后 是否手动应答 true 自动应答 false 手动应答* 3.消费者成功消费的回调* 4. 消费者取消消费的回调*///接收消息DeliverCallback deliverCallback=(conusmerTag,message)->{System.out.println("消费的消息======="+ new String( message.getBody()));};//取消接收消息的回调CancelCallback cancelCallback =(e)->{System.out.println("消费消息被中断=======");};channel.basicConsume(QUEUE_NAME,true,deliverCallback,cancelCallback);} catch (Exception e) {e.printStackTrace();}}
}

获取信道工具类

package com.wlj.rabbitmq.util;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;import java.io.IOException;
import java.util.concurrent.TimeoutException;/*** @创建人 wlj* @创建时间 2023/7/19* @描述 连接MQ获取信道工具类*/
public class MQUtil {public static  Channel    getMQ( ) {//创建一个连接工厂ConnectionFactory factory = new ConnectionFactory();//工厂IP连接MQ的队列factory.setHost("localhost");//设置用户名factory.setUsername("guest");factory.setPassword("guest");try {//创建连接Connection connection = factory.newConnection();//获取信道Channel channel = connection.createChannel();return channel;} catch (IOException e) {e.printStackTrace();} catch (TimeoutException e) {e.printStackTrace();}return null;}}

Work Queues模式

在这里插入图片描述

一个生产者对应多个消费者。每个消费者是竞争关系，一个消息只能被一个消费者消费

消费者代码 C1

package com.wlj.rabbitmq.two;import com.rabbitmq.client.Channel;
import com.wlj.rabbitmq.util.MQUtil;import java.io.IOException;/***@创建人 wlj*@创建时间 2023/7/20*@描述 这是一个工作线程 也是一个消费者*/
public class WorkerConsumer {private static String QUEUE_NAME="hello";public static void main(String[] args) {Channel channel = MQUtil.getMQ();System.out.println("C1");//消息的接收try {channel.basicConsume(QUEUE_NAME,true,(tag,msg)->{System.out.println("消费的消息====="+ new String(msg.getBody()));},e->{System.out.println("消费取消");});} catch (IOException e) {}}
}

开启多线程运行

在这里插入图片描述

启动消费者代码 C2

只需要更改输出语句即可

System.out.println("C2");

生产者代码

从控制台输入发送消息

package com.wlj.rabbitmq.two;import com.rabbitmq.client.*;
import com.wlj.rabbitmq.util.MQUtil;import java.util.Scanner;/***@创建人 wlj*@创建时间 2023/7/20*@描述 消息消费者 工作线程模式*/
public class WorkerProducer {private static String QUEUE_NAME="hello";public static void main(String[] args) {Channel channel = MQUtil.getMQ();try {channel.queueDeclare(QUEUE_NAME, false, false, false, null);//从控制台当中接受信息Scanner scanner = new Scanner(System.in);while (scanner.hasNext()){String message = scanner.next();/*** 生成一个队列* 1.队列名称* 2.队列里面的消息是否持久化 默认消息存储在内存中* 3.该队列是否只供一个消费者进行消费 是否进行共享 true 可以多个消费者消费* 4.是否自动删除 最后一个消费者端开连接以后 该队列是否自动删除 true 自动删除* 5.其他参数*/channel.basicPublish("",QUEUE_NAME,null,message.getBytes());System.out.println("发送消息完成:"+message);}} catch (Exception e) {e.printStackTrace();}}
}

启动C1、C2和生产者进行测试

消息应答

消费者完成一个任务可能需要一段时间，如果其中一个消费者处理一个长的任务并仅只完成
了部分突然它挂掉了，会发生什么情况。RabbitMQ 一旦向消费者传递了一条消息，便立即将该消
息标记为删除。在这种情况下，突然有个消费者挂掉了，我们将丢失正在处理的消息。以及后续
发送给该消费这的消息，因为它无法接收到。
为了保证消息在发送过程中不丢失，rabbitmq 引入消息应答机制，消息应答就是:消费者在接
收到消息并且处理该消息之后，告诉 rabbitmq 它已经处理了，rabbitmq 可以把该消息删除了。

自动应答

消息发送后立即被认为已经传送成功，这种模式需要在高吞吐量和数据传输安全性方面做权
衡,因为这种模式如果消息在接收到之前，消费者那边出现连接或者 channel 关闭，那么消息就丢
失了,当然另一方面这种模式消费者那边可以传递过载的消息，没有对传递的消息数量进行限制，
当然这样有可能使得消费者这边由于接收太多还来不及处理的消息，导致这些消息的积压，最终
使得内存耗尽，最终这些消费者线程被操作系统杀死，所以这种模式仅适用在消费者可以高效并
以某种速率能够处理这些消息的情况下使用。

消息应答的方法

Channel.basicAck(用于肯定确认)
RabbitMQ 已知道该消息并且成功的处理消息，可以将其丢弃了
Channel.basicNack(用于否定确认)
Channel.basicReject(用于否定确认)
与 Channel.basicNack 相比少一个参数
不处理该消息了直接拒绝，可以将其丢弃了

Multiple 的解释

手动应答的好处是可以批量应答并且减少网络拥堵
在这里插入图片描述

multiple 的 true 和 false 代表不同意思
true 代表批量应答 channel 上未应答的消息
比如说 channel 上有传送 tag 的消息 5,6,7,8 当前 tag 是 8 那么此时
5-8 的这些还未应答的消息都会被确认收到消息应答
false 同上面相比
只会应答 tag=8 的消息 5,6,7 这三个消息依然不会被确认收到消息应答

消息自动重新入队

如果消费者由于某些原因失去连接(其通道已关闭，连接已关闭或 TCP 连接丢失)，导致消息
未发送 ACK 确认，RabbitMQ 将了解到消息未完全处理，并将对其重新排队。如果此时其他消费者
可以处理，它将很快将其重新分发给另一个消费者。这样，即使某个消费者偶尔死亡，也可以确
保不会丢失任何消息。
在这里插入图片描述

消息手动应答代码

默认消息采用的是自动应答，所以我们要想实现消息消费过程中不丢失，需要在接收消息是把自动应答改
为手动应答。之后需要在消息处理完成之后进行手动应答

消费者API

** 首先是消费消息时，取消自动应答**
消费消息完成时，进行手动应答

package com.wlj.rabbitmq.two;import com.rabbitmq.client.Channel;
import com.wlj.rabbitmq.util.MQUtil;import java.io.IOException;/***@创建人 wlj*@创建时间 2023/7/20*@描述 这是一个工作线程 也是一个消费者*/
public class WorkerConsumer {private static String QUEUE_NAME="hello";public static void main(String[] args) {Channel channel = MQUtil.getMQ();System.out.println("C2");//消息的接收try {/***消费者消费消息* 1.消费那个队列* 2. 消费成功之后 是否手动应答 true 自动应答 false 手动应答* 3.消费者成功消费的回调* 4. 消费者取消消费的回调*///取消自动应答channel.basicConsume(QUEUE_NAME,false,(tag,msg)->{System.out.println("消费的消息====="+ new String(msg.getBody()));//进行手动应答 参数一：消息标记tag ，参数二：false代表只应答接收到的那个传递的消息。true代表为应答所有消息包括传递过来的消息channel.basicAck(msg.getEnvelope().getDeliveryTag(),true);},e->{System.out.println("消费取消");});} catch (IOException e) {}}
}

队列持久化

队列为开启持久化时。MQ重启队列就会被删除掉，如果想要队列实现持久化，需要在声明队列的时候吧durable参数设置为持久化（true）

 boolean durable =true;channel.queueDeclare(QUEUE_NAME,durable,false,false,null);

需要注意的是,如果之前生声明的队列不是持久化的,需要把原来的队列删除之后,重新声明持久化队列
查看mq队列列表，出现D就是持久化成功
在这里插入图片描述

消息持久化

要想让消息实现持久化需要在消息生产者修改代码,，MessageProperties.PERSISTENT_TEXT_PLAIN 添
加这个属性可以理解为告诉队列把消息实现持久化保存到磁盘上

 boolean durable =true;channel.queueDeclare(QUEUE_NAME,durable,false,false,null);String message="hello world2";/*** 发送一个消息* 1.发送到那个交换机* 2.路由的 key 是哪个* 3.其他的参数信息* 4.发送消息的消息体*/channel.basicPublish("",QUEUE_NAME,MessageProperties.PERSISTENT_TEXT_PLAIN,message.getBytes());

将消息标记为持久化并不能完全保证不会丢失消息。尽管它告诉 RabbitMQ 将消息保存到磁盘，但是
这里依然存在当消息刚准备存储在磁盘的时候但是还没有存储完，消息还在缓存的一个间隔点。此时并没
有真正写入磁盘。持久性保证并不强，但是对于我们的简单任务队列而言，这已经绰绰有余了。如果需要
更强有力的持久化策略，参考MQ的发布确认。

不公平分发

MQ默认是采用轮询的方式分发消息，但是有的消费者处理很慢，就会导致消息积压，可以设置不公平分发，消费者进行应答之后，才会接收下一条消息

生产者代码无变动

package com.wlj.rabbitmq.bugongpingfenfa;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.MessageProperties;import java.io.IOException;
import java.util.Scanner;
import java.util.concurrent.TimeoutException;/***@创建人 wlj*@创建时间 2023/7/20*@描述 消息生产者 Hello World简单队列模式*/
public class Producer {private static String QUEUE_NAME="hello3";public static void main(String[] args) {//创建一个连接工厂ConnectionFactory factory = new ConnectionFactory();//工厂IP连接MQ的队列factory.setHost("localhost");//设置用户名factory.setUsername("guest");factory.setPassword("guest");try {//创建连接Connection connection = factory.newConnection();//获取信道Channel channel = connection.createChannel();//生成一个队列/*** 生成一个队列* 1.队列名称* 2.队列里面的消息是否持久化 默认消息存储在内存中* 3.该队列是否只供一个消费者进行消费 是否进行共享 true 可以多个消费者消费* 4.是否自动删除 最后一个消费者端开连接以后 该队列是否自动删除 true 自动删除* 5.其他参数*///从控制台当中接受信息channel.queueDeclare(QUEUE_NAME, false, false, false, null);Scanner scanner = new Scanner(System.in);while (scanner.hasNext()) {String message = scanner.next();//发布消息channel.basicPublish("", QUEUE_NAME, null, message.getBytes());System.out.println("发送消息完成:" + message);}} catch (IOException e) {e.printStackTrace();} catch (TimeoutException e) {e.printStackTrace();}}}

消费者代码在消息消费之前 channel.basicQos(1); 同时开通了手动应答，手动应答应该先启动消费者，在启动生产者

package com.wlj.rabbitmq.bugongpingfenfa;import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import com.wlj.rabbitmq.util.MQUtil;
import com.wlj.rabbitmq.util.ThreadSleep;import java.io.IOException;/***@创建人 wlj*@创建时间 2023/7/20*@描述 进行不公平分发测试，处理慢的消费者接受的消息少，处理快的消费者处理的消息多*/
public class WorkerConsumer {private static String QUEUE_NAME="hello3";public static void main(String[] args) {Channel channel = MQUtil.getMQ();System.out.println("50秒消费的消息=====");        //消息的接收try {channel.basicQos(1);/***消费者消费消息* 1.消费那个队列* 2. 消费成功之后 是否手动应答 true 自动应答 false 手动应答* 3.消费者成功消费的回调* 4. 消费者取消消费的回调*///接收消息DeliverCallback deliverCallback=(conusmerTag, message)->{System.out.println("消费的消息======="+ new String( message.getBody()));ThreadSleep.sleep(50000);channel.basicAck(message.getEnvelope().getDeliveryTag(),false);System.out.println("应答结束");};//取消接收消息的回调CancelCallback cancelCallback =(e)->{System.out.println("消费消息被中断=======");};channel.basicConsume(QUEUE_NAME,false,deliverCallback,cancelCallback);} catch (IOException e) {}}
}

进行测试

启动消费者一，获取消息之后，睡眠十秒再应答。
启动消费者二，获取消息之后，睡眠二十秒再应答。
启动生产者
测试发送第一条消息，由第一个消费者消费。再发送第二条消息，此时消费者二是空闲的，所以消费者二消费消息
发送第三条数据，如果第一个消费者十秒结束，进行应答，那么会得到第三条消息。如果没有进行应答，则不能接收到消息
所以得出结论，不公平分发，是消费者消息处理完应答之后，才会接收到下一条消息

消息预取值

确认发布

这是一种简单的确认方式。他是一种同步确认发布的方式，也就是发布一个消息之后，只有它被确认发布，后续的消息才能被发布。waitForCnfirmsOrDie(long)这个方法只有在消息被确认的时候才返回，如果指定时间内没有被确认，则会抛出异常

单个确认发布

生产者API
需要使用信道开启发布确认

package com.wlj.rabbitmq.three;import com.rabbitmq.client.Channel;
import com.wlj.rabbitmq.util.MQUtil;import java.util.Scanner;/*** @创建人 wlj* @创建时间 单个发布确认模式* @描述*/
public class ComfirmSelecr {private static String QUEUE_NAME = "confirms";public static void main(String[] args) {Channel channel = MQUtil.getMQ();try {//开启发布确认channel.confirmSelect();channel.queueDeclare(QUEUE_NAME, false, false, false, null);long l = System.currentTimeMillis();for (int i = 0; i < 1000; i++) {//发送消息channel.basicPublish("", QUEUE_NAME, null, (i+"").getBytes());System.out.println("发布确认模式:" + i);boolean b = channel.waitForConfirms(1000);if (b) {System.out.println("发布成功");}}long l1 = System.currentTimeMillis();System.out.println("发布时长= === " + (l1 - l));} catch (Exception e) {e.printStackTrace();}}
}

批量确认发布

package com.wlj.rabbitmq.three;import com.rabbitmq.client.Channel;
import com.wlj.rabbitmq.util.MQUtil;/*** @创建人 wlj* @创建时间 批量发布确认模式* @描述*/
public class BatchComfirmSelecr {private static String QUEUE_NAME = "confirms";public static void main(String[] args) {Channel channel = MQUtil.getMQ();//开启发布确认try {channel.confirmSelect();channel.queueDeclare(QUEUE_NAME, false, false, false, null);long l = System.currentTimeMillis();for (int i = 0; i < 1000; i++) {/*** 生成一个队列* 1.队列名称* 2.队列里面的消息是否持久化 默认消息存储在内存中* 3.该队列是否只供一个消费者进行消费 是否进行共享 true 可以多个消费者消费* 4.是否自动删除 最后一个消费者端开连接以后 该队列是否自动删除 true 自动删除* 5.其他参数*/channel.basicPublish("", QUEUE_NAME, null, (i + "").getBytes());System.out.println("发布确认模式:" + i);// 每一百个一次确认发布if (i % 100 == 0) {boolean b = channel.waitForConfirms(1000);if (b) {System.out.println("发布成功");}}}long l1 = System.currentTimeMillis();System.out.println("发布时长= === " + (l1 - l));} catch (Exception e) {e.printStackTrace();}}
}

异步批量确认发布

需要添加添加一个异步确认的监听器处理已被处理的消息或者是未被处理的消息 channel.addConfirmListener(ackCallback, nackCallback);

package com.wlj.rabbitmq.three;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;
import com.wlj.rabbitmq.util.MQUtil;import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;/*** @创建人 wlj* @创建时间 批量发布确认模式* @描述*/
public class SyncBatchComfirmSelecr {private static String QUEUE_NAME = "confirms";public static void main(String[] args) {Channel channel = MQUtil.getMQ();//开启发布确认try {channel.confirmSelect();/*** 线程安全有序的一个哈希表，适用于高并发的情况* 1.轻松的将序号与消息进行关联* 2.轻松批量删除条目 只要给到序列号* 3.支持并发访问*/ConcurrentSkipListMap<Long, String> outstandingConfirms = newConcurrentSkipListMap<>();/*** 确认收到消息的一个回调* 1.消息序列号* 2.true 可以确认小于等于当前序列号的消息* false 确认当前序列号消息*/ConfirmCallback ackCallback = (sequenceNumber, multiple) -> {String message = outstandingConfirms.get(sequenceNumber);System.out.println("发布的消息---------------" + message + "已被确认，序列号" + sequenceNumber);//是否是批量确认if (multiple) {//返回的是小于等于当前序列号的未确认消息 是一个 mapConcurrentNavigableMap<Long, String> confirmed =outstandingConfirms.headMap(sequenceNumber, true);//清除该部分未确认消息confirmed.clear();} else {//只清除当前序列号的消息outstandingConfirms.remove(sequenceNumber);}};ConfirmCallback nackCallback = (sequenceNumber, multiple) -> {String message = outstandingConfirms.get(sequenceNumber);System.out.println("发布的消息*************" + message + "未被确认，序列号" + sequenceNumber);};/*** 添加一个异步确认的监听器* 1.确认收到消息的回调* 2.未收到消息的回调*///  channel.addConfirmListener(ackCallback, null);channel.addConfirmListener(ackCallback, nackCallback);channel.queueDeclare(QUEUE_NAME, false, false, false, null);long l = System.currentTimeMillis();for (int i = 0; i < 1000; i++) {String message = "消息" + i;/*** channel.getNextPublishSeqNo()获取下一个消息的序列号* 通过序列号与消息体进行一个关联* 全部都是未确认的消息体*/outstandingConfirms.put(channel.getNextPublishSeqNo(), message);/*** 生成一个队列* 1.队列名称* 2.队列里面的消息是否持久化 默认消息存储在内存中* 3.该队列是否只供一个消费者进行消费 是否进行共享 true 可以多个消费者消费* 4.是否自动删除 最后一个消费者端开连接以后 该队列是否自动删除 true 自动删除* 5.其他参数*/channel.basicPublish("", QUEUE_NAME, null, (i + "").getBytes());}long l1 = System.currentTimeMillis();System.out.println("异步批量确认发布时长= === " + (l1 - l));} catch (Exception e) {e.printStackTrace();}}
}

如何处理异步未确认消息

最好的解决的解决方案就是把未确认的消息放到一个基于内存的能被发布线程访问的队列，
比如说用 ConcurrentLinkedQueue 这个队列在 confirm callbacks 与发布线程之间进行消息的传
递.上面代码就是用了此方法,发消息是进行记录,在未确认回调函数里面进行处理