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

RabbitMQ消息队列

news 2025/9/14 3:00:30

简介

MQ(message queue)，从字面意思上看就个 FIFO 先入先出的队列，只不过队列中存放的内容是 message 而已，它是一种具有接收数据、存储数据、发送数据等功能的技术服务。

作用：流量削峰、应用解耦、异步处理。

在这里插入图片描述
生产者将消息发送到消息队列中，消息队列负责转发消息给消费者，消费者在处理完消息后会对消息队列进行应答，消息队列收到应答信息会将相应的消息进行丢弃。

批量应答会导致高并发时消息的丢失，所以尽力以channel.ack()进行手动应答。

docker安装

拉取镜像并后台运行

docker run -id --name=rabbitmq -v rabbitmq-home:/var/lib/rabbitmq -p 15672:15672 -p 5672:5672 -e RABBITMQ_DEFAULT_USER=yi -e RABBITMQ_DEFAULT_PASS=123456 rabbitmq

需要将RABBITMQ_DEFAULT_USER、RABBITMQ_DEFAULT_PASS改成自己的用户名、密码。

开启manager插件，可以在网页进行管理。

 docker exec -it 容器id /bin/bash  #这里可以用docker ps 查询刚刚开启的容器id#进入容器后输入，开启rabbitmq-plugins enable rabbitmq_management

可以登录 http://服务器IP:15672 访问web管理界面，访问成功则代表开启成功。

JAVA环境搭建

jar包：

<dependencies><dependency><groupId>com.rabbitmq</groupId><artifactId>amqp-client</artifactId><version>5.8.0</version></dependency><dependency><groupId>org.slf4j</groupId><artifactId>slf4j-simple</artifactId><version>1.7.25</version></dependency></dependencies>

Helloworld实例

生产者

public static void main(String[] args) throws IOException, TimeoutException {//创建连接工厂ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("127.0.0.1");connectionFactory.setUsername("yi");connectionFactory.setPassword("123456");//获取连接Connection connection = connectionFactory.newConnection();//获取信道，一个连接中有多个信道Channel channel = connection.createChannel();//声明一个队列 String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> argumentsAMQP.Queue.DeclareOk declareOk = channel.queueDeclare(QUEUE_NAME,false,false,false,null);String message="hello world";//(String exchange, String routingKey, AMQP.BasicProperties props, byte[] body)channel.basicPublish("",QUEUE_NAME,null,message.getBytes());System.out.println("发送成功");}

消费者

public static void main(String[] args) throws IOException, TimeoutException {ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("127.0.0.1");connectionFactory.setUsername("yi");connectionFactory.setPassword("123456");Connection connection = connectionFactory.newConnection();Channel channel = connection.createChannel();DeliverCallback deliverCallback=(consumerTag,message)->{System.out.println(new String(message.getBody()));};CancelCallback cancelCallback=(String var1)->{System.out.println("消息消费被中断");};channel.basicConsume(QUEUE_NAME,true,deliverCallback,cancelCallback);}

工作队列（任务队列）

RabbitMQ默认为工作队列模式，消费者C1,C2为竞争关系，接收到的消息将轮询发送给C1，C2处理，即C1一条C2一条依次循环。
在这里插入图片描述

手动应答ack

因为自动应答不会考虑消息是否处理成功，所以可能会导致消息丢失，需要在代码中将自动应答改为手动应答。批量应答在高并发的时候也容易丢失消息，也应该关闭。

生产者的代码无需修改。
消费者：

public static void main(String[] args) throws IOException, TimeoutException {Channel channel = RabbitMQUtils.getChannel();System.out.println("work2 waiting:");DeliverCallback deliverCallback= (String s, Delivery delivery)->{System.out.println(new String(delivery.getBody()));// do something//手动回复ack，false为关闭批量应答channel.basicAck(delivery.getEnvelope().getDeliveryTag(),false);};CancelCallback cancelCallback=(s)->{System.out.println("消息被打断");};//false表示不自动应答ackchannel.basicConsume(QUEUE_NAME,false,deliverCallback,cancelCallback);}

不公平分发

会存在有些线程能力差耗时长，有些能力强耗时短的情况，不公平分发将实现能者多劳。
设立channel的basicQos即可实现不公平分发， basicQos的数值意味着channel的最大存储上限，channel为1时，消费者最多同时缓存一条待处理消息。

channel.basicQos(1);

发布确认

在开启队列持久化、消息持久化后，RabbitMQ服务器仍然可能在将消息存储在磁盘前宕机，需要发布确认才能保证消息不丢失，即RabbitMQ在存储磁盘成功后，发送确认给生产者。

单个发布确认

每条消息存储在磁盘后进行发布确认，只有发送者在接收到消费者对应的发布确认消息后才会给此消费者发送下一条消息。

public static void publicMsgIndividual()throws Exception{Channel channel = RabbitMQUtils.getChannel();String QUEUE_NAME = UUID.randomUUID().toString();//开启持久化channel.queueDeclare(QUEUE_NAME,true,false,false,null);channel.confirmSelect();//开启发布确认long begin = System.currentTimeMillis();for (int i = 0; i < 1000; i++) {channel.basicPublish("",QUEUE_NAME,null, new String(i+" ").getBytes());       boolean flag = channel.waitForConfirms(); //等待发布确认if(flag){System.out.println("消息发送成功");}}long end = System.currentTimeMillis();System.out.println("发布1000条耗时:"+(end-begin)+"ms");}

批量发布确认

每发送100条消息进行一次发布确认。速度快，但是不知道具体是哪一条消息发送失败了。

public static void publicMsgIndividual()throws Exception{Channel channel = RabbitMQUtils.getChannel();String QUEUE_NAME = UUID.randomUUID().toString();//开启持久化channel.queueDeclare(QUEUE_NAME,true,false,false,null);channel.confirmSelect();//开启发布确认long begin = System.currentTimeMillis();for (int i = 0; i < 1000; i++) {channel.basicPublish("",QUEUE_NAME,null, new String(i+" ").getBytes());if(i%100==0){boolean flag = channel.waitForConfirms(); //等待发布确认if(flag){System.out.println("消息发送成功");}}}long end = System.currentTimeMillis();System.out.println("发布1000条耗时:"+(end-begin)+"ms");}

异步发布确认

推荐使用，需要加入确认发布监听器confirmListener，并且记录序列号与消息的关联（ConcurrentSkipListMap）。

 public static void publicMsgAsync()throws Exception{Channel channel = RabbitMQUtils.getChannel();String QUEUE_NAME = UUID.randomUUID().toString();//开启持久化channel.queueDeclare(QUEUE_NAME,true,false,false,null);channel.confirmSelect();//开启发布确认//       将序列号与信息相关联，ConcurrentSkipListMap concurrentSkipListMap = new ConcurrentSkipListMap<Long,String>();//加入确认监听器channel.addConfirmListener(new ConfirmListener() {@Overridepublic void handleAck(long msgTag, boolean multiply) throws IOException {System.out.println("消息发送成功:"+msgTag);if(multiply) { //如果是批量确认，批量删除//headMap返回小于msgTag的map视图ConcurrentNavigableMap concurrentNavigableMap = concurrentSkipListMap.headMap(msgTag);//清理已经标记的MapconcurrentNavigableMap.clear();}else {concurrentSkipListMap.remove(msgTag);}}@Overridepublic void handleNack(long msgTag, boolean multiply) throws IOException {System.out.println("未确认的消息:"+concurrentSkipListMap.get(msgTag));}});long begin = System.currentTimeMillis();for (int i = 0; i < 1000; i++) {channel.basicPublish("",QUEUE_NAME,null, new String(i+" ").getBytes());//记录发送的信息与其序列号concurrentSkipListMap.put(channel.getNextPublishSeqNo(),new String(i+" "));}long end = System.currentTimeMillis();System.out.println("发布1000条耗时:"+(end-begin)+"ms");}

发布/订阅模式（fanout交换机）

首先要弄明白交换机和队列的关系，交换机负责信息的接收，通过不同的RountingKey将消息转发到不同的队列，每个队列上的接收者都是竞争关系（即队列上的消息只会被处理一次），那么当一个交换机对应多个队列时，每个队列仅有一个消费者，这个时候即发布/订阅模式，消息会被每个消费者接收。
在这里插入图片描述

生产者代码：向交换机中发送消息

public static final String EXCHANGE_NAME="logs";public static void main(String[] args) throws IOException, TimeoutException {Channel channel = RabbitMQUtils.getChannel();Scanner scanner = new Scanner(System.in);while (scanner.hasNext()){String next = scanner.next();channel.basicPublish(EXCHANGE_NAME,"", null,next.getBytes());}}

消费者代码：声明匿名队列，将队列绑定到交换机上，不同的消费者用相同的RountingKey，以便同时接收到消息。

public static final String EXCHANGE_NAME="logs";public static void main(String[] args) throws IOException, TimeoutException {Channel channel = RabbitMQUtils.getChannel();channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT); //FANOUT煽出，就是发布订阅模式String queue = channel.queueDeclare().getQueue(); //声明匿名队列channel.queueBind(queue,EXCHANGE_NAME,""); //将队列绑定到交换机上，RountingKey为“”DeliverCallback deliverCallback=(consumerTag,message)->{System.out.println("接收到消息:"+new String(message.getBody()));};channel.basicConsume(queue,true,deliverCallback, (consumerTag)->{});}

Direct交换机

与fanout模式相比，不同的队列有不同的Rounting key，通过Rounting Key能够直接向指定队列发送消息。

Topic交换机

rountingKey作为匹配串，发送消息时，匹配上的则能进行发送。
routingKey必须是单词列表，用.隔开。如aa.bb.cc
*可以代表一个单词，#可以代表若干个单词
比如向rountingKey为aa.orange.rabbit发送消息，Q1和Q2都能接收到消息，而向aa.orange.bb发送消息则只有Q1能够接收到消息。
在这里插入图片描述
当队列的rountingKey绑定的#，则相当于fanout煽出交换机。
当队列的rountingKey绑定不带#*时，相当于direct交换机。