RabbitMQ系列—Java操作之简单队列、工作队列

朴灿烈づ我的快乐病毒、 2022-05-08 08:46 260阅读 0赞

RabbitMQ官网介绍了，它支持六种应用场景：简单队列、工作队列、发布/订阅、路由模式、Topics主题模式、RPC，接下来分别介绍。

![70][]

创建一个Maven项目命名rabbitmq，并引入rabbitmq依赖。

<dependency>
    	<groupId>com.rabbitmq</groupId>
    	<artifactId>amqp-client</artifactId>
    	<version>5.2.0</version>
    </dependency>

### 简单队列 ###

![70 1][]

其中P是生产者，红色部分是队列，C是消费者。工作原理就是生产者生产消息，将消息放到队列里，消费者负责在队列取出消息进行消费，其中队列是rabbitmq实现的，所以我们需要实现生产者和消费者。

创建连接工具类ConnectionUtils，后面都会用到。

package com.rabbitmq.util;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    
    /**
     * rabbitmq连接工具类
     * @author Administrator
     *
     */
    public class ConnectionUtils {
    	/**
    	 * 获取连接
    	 * @return
    	 * @throws IOException
    	 * @throws TimeoutException
    	 */
    	public static Connection getConnection() throws IOException, TimeoutException{
    		ConnectionFactory factory = new ConnectionFactory();
    		// 设置服务地址
    		factory.setHost("127.0.0.1");
    		// 端口
    		factory.setPort(5672);
    		// vhost
    		factory.setVirtualHost("/vhost_test");
    		// 用户名
    		factory.setUsername("admin");
    		// 密码
    		factory.setPassword("123456");
    		return factory.newConnection();
    	}
    	/**
    	 * 关闭连接
    	 * @param channel
    	 * @param con
    	 */
    	public static void close(Channel channel,Connection con){
    		if(channel != null){
    			try {
    				channel.close();
    			} catch (IOException e) {
    				e.printStackTrace();
    			} catch (TimeoutException e) {
    				e.printStackTrace();
    			}
    		}
    		if(con != null){
    			try {
    				con.close();
    			} catch (IOException e) {
    				// TODO Auto-generated catch block
    				e.printStackTrace();
    			}
    		}
    	}
    
    }

**生产者Sender**

package com.rabbitmq.simple;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.util.ConnectionUtils;
    /**
     * 简单消息队列——生产者
     * @author Administrator
     *
     */
    public class Sender {
    	/**
    	 * 队列名称
    	 */
    	private static final String QUEUE = "test_simple_queue";
    	
    	public static void main(String[] args) {
    		Connection con = null;
    		Channel channel = null;
    		try {
    			// 获取连接
    			con = ConnectionUtils.getConnection();
    			// 从连接中创建通道
    			channel = con.createChannel();
    			// 声明一个队列
    			channel.queueDeclare(QUEUE, false, false, false, null);
    			// 消息内容
    			String msg = "simple queue hello!";
    			// 发送消息
    			channel.basicPublish("", QUEUE, null, msg.getBytes());
    			System.out.println("send success");
    		} catch (IOException e) {
    			e.printStackTrace();
    		} catch (TimeoutException e) {
    			e.printStackTrace();
    		} finally {
    			// 关闭连接
    			ConnectionUtils.close(channel, con);
    		}
    		
    	}
    
    }

channel.queueDeclare()方法的作用是声明一个队列，它只在所声明的队列不存在的情况下生效，如果队列已经存在在不做任何操作，在此方法中具有如下参数：

*  String queue：队列名称。
 *  boolean durable：是否持久化。队列模式是在内存中的，如果重启rabbitmq消息会丢失，如果设置为true，会保存到erlang自带的数据库，重启后可以恢复。
 *  boolean exclusive：是否排外。作用一，连接关闭后是否自动删除当前队列；作用二，是否私有队列，如果为true，则其他通道不能访问当前队列。
 *  boolean autoDelete：当所有消费者客户端断开连接时是否自动删除队列。
 *  Map<String, Object> arguments：其他参数。

channel.basicPublish()方法的作用是发送消息到队列，它具有如下参数：

*  String exchange：交换机名称，简单队列用不到交换机，此处写""空字符串即可。
 *  String routingKey：队列映射的路由key，此处就是队列名称。
 *  BasicProperties props：消息的其他属性。
 *  byte\[\] body：发送信息的主体。rabbitmq一般不用来发送大数据类型的消息。

接下来运行Sender生产者。

![70 2][]

访问[http://localhost:15672][http_localhost_15672]控制台的Queues选项，发现多了一个队列。

![70 3][]

点击此队列，点击Get Message，就可以消费这个消息了。

![70 4][]

接下来构建消费者，消费者有两种写法，一种是旧的API，一种是新的API，我当前使用的rabbitmq的jar包是5.2.0的，已经不支持旧的API了，就不介绍了，只介绍新的API。

**消费者Recver**

package com.rabbitmq.simple;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    import com.rabbitmq.client.AMQP.BasicProperties;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.Consumer;
    import com.rabbitmq.client.DefaultConsumer;
    import com.rabbitmq.client.Envelope;
    import com.rabbitmq.util.ConnectionUtils;
    
    /**
     * 简单队列——消费者
     * 
     * @author Administrator
     *
     */
    public class Recver {
    	/**
    	 * 队列名称，和生产者的队列名称必须保持一致
    	 */
    	private static final String QUEUE = "test_simple_queue";
    
    	public static void main(String[] args) throws IOException, TimeoutException {
    		// 获取连接
    		Connection con = ConnectionUtils.getConnection();
    		// 从连接中创建通道
    		Channel channel = con.createChannel();
    		// 声明队列
    		channel.queueDeclare(QUEUE, false, false, false, null);
    		// 创建消费者
    		Consumer consumer = new DefaultConsumer(channel) {
                            // 获取消息
    			@Override
    			public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
    					throws IOException {
    				String msg = new String(body, "utf-8");
    				System.out.println("接收到消息——" + msg);
    			}
    
    		};
    		// 监听队列
    		channel.basicConsume(QUEUE, true, consumer);
    	}
    }

channel.basicConsume()方法的作用是监听消息队列，它具有三个参数：

*  String queue：队列名称。
 *  boolean autoAck：是否自动应答，如果为false，需要我们手动通知服务器接收到消息了。
 *  Consumer callback：回调方法。

接下来启动消费者，再启动生产者，发现消费者控制台打印了消息

![70 5][]

在接下来关掉消费者，启动两次生产者生产两条消息，再次启动消费者控制台打印

![70 6][]

**简单队列缺点**

耦合度高，队列名在一端改动，另一端也要跟着改动。生产者和消费者一一对应，不支持多个消费者。

### 工作队列 ###

![70 7][]

一般在实际应用中，生产者发送消息耗时较少，反应较快，反而是消费者因为要处理业务逻辑，处理时间可能会很慢，这样队列中会积压很多消息，所以需要多个消费者分摊压力，这个时候可以使用工作队列。

工作队列的逻辑就是队列拿到生产者的消息后会在消费者中选择一个把消息发送过去，并不是把消息同时发送给两个消费者。

**生产者Sender**

public class Sender {
    
    	private static final String QUEUE = "test_work_queue";
    
    	public static void main(String[] args) {
    		Connection con = null;
    		Channel channel = null;
    		try {
    			// 获取连接
    			con = ConnectionUtils.getConnection();
    			// 从连接中创建通道
    			channel = con.createChannel();
    			// 声明一个队列
    			channel.queueDeclare(QUEUE, false, false, false, null);
    			// 发送50条消息 
    			for (int i = 0; i < 50; i++) {
    				// 消息内容
    				String msg = "work queue " + i;
    				// 发送消息
    				channel.basicPublish("", QUEUE, null, msg.getBytes());
    				try {
    					Thread.sleep(i * 20);
    				} catch (InterruptedException e) {
    					e.printStackTrace();
    				}
    			}
    			System.out.println("send success");
    		} catch (IOException e) {
    			e.printStackTrace();
    		} catch (TimeoutException e) {
    			e.printStackTrace();
    		} finally {
    			// 关闭连接
    			ConnectionUtils.close(channel, con);
    		}
    	}
    }

**消费者Recver1**

public class Recver1 {
    	
    	private static final String QUEUE = "test_work_queue";
    
    	public static void main(String[] args) throws IOException, TimeoutException {
    		// 获取连接
    		Connection con = ConnectionUtils.getConnection();
    		// 从连接中创建通道
    		Channel channel = con.createChannel();
    		// 声明队列
    		channel.queueDeclare(QUEUE, false, false, false, null);
    		// 创建消费者
    		Consumer consumer = new DefaultConsumer(channel) {
                            // 获取消息
    			@Override
    			public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
    					throws IOException {
    				String msg = new String(body, "utf-8");
    				System.out.println("Recver1接收到消息——" + msg);
    				// 模拟延迟,每2秒处理一个消息
    				try {
    					Thread.sleep(2000);
    				} catch (InterruptedException e) {
    					e.printStackTrace();
    				}
    			}
    
    		};
    		// 监听队列
    		channel.basicConsume(QUEUE, true, consumer);
    	}
    }

**消费者Recver2**

public class Recver2 {
    	
    	private static final String QUEUE = "test_work_queue";
    
    	public static void main(String[] args) throws IOException, TimeoutException {
    		// 获取连接
    		Connection con = ConnectionUtils.getConnection();
    		// 从连接中创建通道
    		Channel channel = con.createChannel();
    		// 声明队列
    		channel.queueDeclare(QUEUE, false, false, false, null);
    		// 创建消费者
    		Consumer consumer = new DefaultConsumer(channel) {
    			// 获取消息
    			@Override
    			public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
    					throws IOException {
    				String msg = new String(body, "utf-8");
    				System.out.println("Recver2接收到消息——" + msg);
    				// 模拟延迟,每1秒处理一个消息
    				try {
    					Thread.sleep(1000);
    				} catch (InterruptedException e) {
    					e.printStackTrace();
    				}
    			}
    
    		};
    		// 监听队列
    		channel.basicConsume(QUEUE, true, consumer);
    	}
    }

依次运行消费者Recver1、消费者Recver2、生产者Sender，控制台打印：

![70 8][]

![70 9][]

发现消费者Recver1和消费者Recver2消费的消息数量是一样多的，消费者Recver1获取的都是奇数的消息，消费者Recver2获取的都是偶数的消息。尽管我们通过线程休眠模拟消费者Recver1更加忙一些，处理更慢一些，但是处理的消息数量不受影响。这种模式就是轮询分发，两个消费者总是均分所有消息。

这样虽然解决了简单队列不能同时存在多个消费者的问题，但是依然有问题，我们希望处理的快的消费者来处理更多的消息，分担更多的压力。这就不能再使用轮询分发了，而是公平分发。我们需要改造一下生产者和消费者。

**改造生产者**

public class Sender {
    
    	private static final String QUEUE = "test_work_queue";
    
    	public static void main(String[] args) {
    		Connection con = null;
    		Channel channel = null;
    		try {
    			// 获取连接
    			con = ConnectionUtils.getConnection();
    			// 从连接中创建通道
    			channel = con.createChannel();
    			// 声明一个队列
    			channel.queueDeclare(QUEUE, false, false, false, null);
    			// 每个消费者发送确认收到消息之前,消息队列不发送下一个消息到消费者,一次只处理一个消息
    			// 限制发送给同一个消费者不超过1条消息
    			channel.basicQos(1);
    			// 发送50条消息 
    			for (int i = 0; i < 50; i++) {
    				// 消息内容
    				String msg = "work queue " + i;
    				// 发送消息
    				channel.basicPublish("", QUEUE, null, msg.getBytes());
    				try {
    					Thread.sleep(i * 20);
    				} catch (InterruptedException e) {
    					e.printStackTrace();
    				}
    			}
    			System.out.println("send success");
    		} catch (IOException e) {
    			e.printStackTrace();
    		} catch (TimeoutException e) {
    			e.printStackTrace();
    		} finally {
    			// 关闭连接
    			ConnectionUtils.close(channel, con);
    		}
    	}
    }

channel.basicQos(1)：限制发送给同一个消费者不超过1条消息，每个消费者发送确认收到消息之前，消息队列不发送下一个消息到消费者，保证一次只处理一个消息。

**改造消费者**

public class Recver1 {
    	
    	private static final String QUEUE = "test_work_queue";
    
    	public static void main(String[] args) throws IOException, TimeoutException {
    		// 获取连接
    		Connection con = ConnectionUtils.getConnection();
    		// 从连接中创建通道
    		final Channel channel = con.createChannel();
    		// 声明队列
    		channel.queueDeclare(QUEUE, false, false, false, null);
    		// 保证一次只分发一条消息
    		channel.basicQos(1);
    		// 创建消费者
    		Consumer consumer = new DefaultConsumer(channel) {
    			// 获取消息
    			@Override
    			public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
    					throws IOException {
    				String msg = new String(body, "utf-8");
    				System.out.println("Recver1接收到消息——" + msg);
    				// 模拟延迟,每2秒处理一个消息
    				try {
    					Thread.sleep(2000);
    				} catch (InterruptedException e) {
    					e.printStackTrace();
    				} finally {
    					// 手动确认消息
    					channel.basicAck(envelope.getDeliveryTag(), false);
    				}
    			}
    
    		};
    		// 监听队列
    		channel.basicConsume(QUEUE, false, consumer);
    	}
    }

改动的位置：

*  添加channel.basicQos(1)：保证一次只分发一条消息。
 *  channel.basicAck(envelope.getDeliveryTag(), false)：手动确认消息。false表示确认接收消息，true表示拒绝接收消息。
 *  channel.basicConsume(QUEUE, false, consumer)：设置自动应答为false。

消费者Recver2同样处理，不贴代码了。

依次运行消费者1、消费者2、生产者，控制台打印：

![70 10][]

![70 11][]

消费者Recver2还有两条打印没截上。。

现在rabbitmq采用的就是公平分发的模式来发送消息给消费者，消费者Recver2处理速度快自然收到的消息就多了。

**消息应答**

现在详细介绍一下channel.basicConsume的第二个参数autoAck，如果autoAck为true，就是自动应答模式，当消息队列发送给消费者后，队列就会将消息在内存中删除，如果此时消费者还没处理完消息就挂掉了，那么这个消息就丢失了。这并不是我们想看到的，我们希望一个消费者挂掉后，队列把这个消息发送给其他消费者进行处理，这就需要把autoAck设置为false。autoAck为false就是手动模式，当队列把消息发送给消费者后，只有消费者告诉rabbitmq已经处理完了才会删除内存中的消息。autoAck默认为false。

当autoAck为false时，对于RabbitMQ服务器端而言，队列中的消息分成了两部分：一部分是等待投递给消费者的消息；一部分是已经投递给消费者，但是还没有收到消费者ack信号的消息。如果服务器端一直没有收到消费者的ack信号，并且消费此消息的消费者已经断开连接，则服务器端会安排该消息重新进入队列，等待投递给下一个消费者（也可能还是原来的那个消费者），因为只有consumer向broker发送了ack，broker才会删除消息，所以此时broker并没有删除消息，如果消费者再次正常消费，依然可以获得消息。

消息收到未确认会怎么样？

如果应用程序接收了消息，因为bug忘记确认接收的话，消息在队列的状态会从“Ready”变为“Unacked”，如图：

![70 12][]

如果消息收到却未确认，rabbimq将不会再给这个消费者发送更多的消息了，因为rabbimq认为你没有准备好接收下一条消息。此条消息会一直保持Unacked的状态，直到消费者确认了消息，或者断开与rabbimq的连接，rabbimq会自动把消息改完Ready状态，分发给其他消费者。

**消息拒绝**

消息在确认之前，可以有两个选择：

选择1：断开与rabbimq的连接，这样rabbimq会重新把消息分派给另一个消费者；

选择2：拒绝rabbimq发送的消息使用channel.basicReject(long deliveryTag, boolean requeue)，参数1：消息的id；参数2：处理消息的方式，如果是true，rabbimq会重新分配这个消息给其他订阅者，如果设置成false的话，rabbimq会把消息发送到一个特殊的“死信”队列，用来存放被拒绝而不重新放入队列的消息。

**消息持久化**

消息应答的手动模式解决了消费者挂掉后消息会丢失的问题，但是如果rabbitmq挂掉了呢，消息仍然还会丢失。所以我们希望在重启rabbitmq后仍然可以拿到这些消息，这就需要用到消息持久化。

此时就需要把channel.queueDeclare()的第二个参数durable设置为true，即设置消息持久化。rabbitmq的队列拿到消息后会将消息保存到Erlang的数据库中，也可以理解为保存到磁盘上，重启rabbitmq会重新读取消息。

消息持久化的优点显而易见，但缺点也很明显，那就是性能，因为要写入硬盘要比写入内存性能较低很多，从而降低了服务器的吞吐量，尽管使用SSD硬盘可以使事情得到缓解，但他仍然吸干了Rabbit的性能，当消息成千上万条要写入磁盘的时候，性能是很低的。所以使用者要根据自己的情况，选择适合自己的方式。

现在我们把channel.queueDeclare()的durable从false改为true，并启动生产者和消费者，发现报错了。

java.io.IOException
    	at com.rabbitmq.client.impl.AMQChannel.wrap(AMQChannel.java:126)
    	at com.rabbitmq.client.impl.AMQChannel.wrap(AMQChannel.java:122)
    	at com.rabbitmq.client.impl.AMQChannel.exnWrappingRpc(AMQChannel.java:144)
    	at com.rabbitmq.client.impl.ChannelN.queueDeclare(ChannelN.java:953)
    	at com.rabbitmq.client.impl.recovery.AutorecoveringChannel.queueDeclare(AutorecoveringChannel.java:333)
    	at com.rabbitmq.work.fair.Sender.main(Sender.java:29)
    Caused by: com.rabbitmq.client.ShutdownSignalException: channel error; protocol method: #method<channel.close>(reply-code=406, reply-text=PRECONDITION_FAILED - inequivalent arg 'durable' for queue 'test_work_queue' in vhost '/vhost_test': received 'true' but current is 'false', class-id=50, method-id=10)
    	at com.rabbitmq.utility.ValueOrException.getValue(ValueOrException.java:66)
    	at com.rabbitmq.utility.BlockingValueOrException.uninterruptibleGetValue(BlockingValueOrException.java:36)
    	at com.rabbitmq.client.impl.AMQChannel$BlockingRpcContinuation.getReply(AMQChannel.java:494)
    	at com.rabbitmq.client.impl.AMQChannel.privateRpc(AMQChannel.java:288)
    	at com.rabbitmq.client.impl.AMQChannel.exnWrappingRpc(AMQChannel.java:138)
    	... 3 more
    Caused by: com.rabbitmq.client.ShutdownSignalException: channel error; protocol method: #method<channel.close>(reply-code=406, reply-text=PRECONDITION_FAILED - inequivalent arg 'durable' for queue 'test_work_queue' in vhost '/vhost_test': received 'true' but current is 'false', class-id=50, method-id=10)
    	at com.rabbitmq.client.impl.ChannelN.asyncShutdown(ChannelN.java:510)
    	at com.rabbitmq.client.impl.ChannelN.processAsync(ChannelN.java:346)
    	at com.rabbitmq.client.impl.AMQChannel.handleCompleteInboundCommand(AMQChannel.java:178)
    	at com.rabbitmq.client.impl.AMQChannel.handleFrame(AMQChannel.java:111)
    	at com.rabbitmq.client.impl.AMQConnection.readFrame(AMQConnection.java:670)
    	at com.rabbitmq.client.impl.AMQConnection.access$300(AMQConnection.java:48)
    	at com.rabbitmq.client.impl.AMQConnection$MainLoop.run(AMQConnection.java:597)
    	at java.lang.Thread.run(Thread.java:748)
    Exception in thread "main" com.rabbitmq.client.AlreadyClosedException: channel is already closed due to channel error; protocol method: #method<channel.close>(reply-code=406, reply-text=PRECONDITION_FAILED - inequivalent arg 'durable' for queue 'test_work_queue' in vhost '/vhost_test': received 'true' but current is 'false', class-id=50, method-id=10)
    	at com.rabbitmq.client.impl.AMQChannel.processShutdownSignal(AMQChannel.java:396)
    	at com.rabbitmq.client.impl.ChannelN.startProcessShutdownSignal(ChannelN.java:292)
    	at com.rabbitmq.client.impl.ChannelN.close(ChannelN.java:601)
    	at com.rabbitmq.client.impl.ChannelN.close(ChannelN.java:535)
    	at com.rabbitmq.client.impl.ChannelN.close(ChannelN.java:528)
    	at com.rabbitmq.client.impl.recovery.AutorecoveringChannel.close(AutorecoveringChannel.java:68)
    	at com.rabbitmq.util.ConnectionUtils.close(ConnectionUtils.java:44)
    	at com.rabbitmq.work.fair.Sender.main(Sender.java:52)

这是因为rabbitmq不允许重新定义一个已存在的队列，解决的办法是在Web控制台删除队列再重新声明或者声明一个新的队列即可。

[70]: /images/20220508/49ec9634c56c472fbe0bca19fee04720.png
[70 1]: /images/20220508/4ef26cf059754cf0b2c5a07fe0b02eec.png
[70 2]: /images/20220508/16dcb6a6f8974cf694de1ebc5dd88880.png
[http_localhost_15672]: http://localhost:15672
[70 3]: /images/20220508/a33bf3d4a08b4daa9844755027d4b1ff.png
[70 4]: /images/20220508/15024bed6bc64689b52f06ebc86017d5.png
[70 5]: /images/20220508/088e7e12dc8f4581a78fa69bc85e25f3.png
[70 6]: /images/20220508/c932270301a54b2c88824ed4d4d200b4.png
[70 7]: /images/20220508/4c4ad7853b1a406a83e9809a38f69e89.png
[70 8]: /images/20220508/26ae17344c11489a9a7bff8000c42bd7.png
[70 9]: /images/20220508/2b5f1624cac844a78659acbe2c872c4b.png
[70 10]: /images/20220508/81fcda0676834181a1b50e07eb578b7c.png
[70 11]: /images/20220508/6ec3a1b74643481a95500769081e2867.png
[70 12]: /images/20220508/208fc7c1fdb749f19d8b8788006a83d9.png