RabbitMQ整合Spring Booot【公平队列】

客官°小女子只卖身不卖艺 2022-12-08 04:27 110阅读 0赞

producer：

package com.toov5.Producer;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection; import com.toov5.utils.MQConnectionUtils; public class Producer { // 队列名称 private static final String UEUE_NAME = "test_queue"; public static void main(String[] args) throws IOException, TimeoutException { // 创建新的连接 Connection connection = MQConnectionUtils.newConnection(); // 创建Channel Channel channel = connection.createChannel(); // 创建队列 channel.queueDeclare(UEUE_NAME, false, false, false, null); channel.basicQos(1); // 保证 取一个消费 队列给消费者发送消息时候 一个消息 for (int i = 0; i < 10; i++) { // 创建message String msg = "toov5_message"; System.out.println("生产者投递消息" + msg + i); // 生产者发送消息 channel.basicPublish("", UEUE_NAME, null, msg.getBytes()); } // 关闭通道和连接  channel.close(); connection.close(); } }

Consumer1

package com.toov5.Consumer;
    
    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.DefaultConsumer; import com.rabbitmq.client.Envelope; import com.toov5.utils.MQConnectionUtils; public class Consumer1 { //队列名称 private static final String QUEUE_NAME = "test_queue"; public static void main(String[] args) throws IOException, TimeoutException { System.out.println("消费者启动..........1"); //创建新的连接 Connection connection = MQConnectionUtils.newConnection(); //创建Channel final Channel channel = connection.createChannel(); // 消费者关联队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); channel.basicQos(1); DefaultConsumer defaultConsumerr = 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); try { //模拟应答等待时间 Thread.sleep(1000); } catch (Exception e) { }finally { channel.basicAck(envelope.getDeliveryTag(), false); //手动应答 告诉消息队列服务器 消费成功  } } }; //牵手模式设置 默认自动应答模式 true:自动应答模式 channel.basicConsume(QUEUE_NAME, false, defaultConsumerr);// fanse手动应答  } }

Consumer2

package com.toov5.Consumer;
    
    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.DefaultConsumer; import com.rabbitmq.client.Envelope; import com.toov5.utils.MQConnectionUtils; public class Consumer2 { //队列名称 private static final String QUEUE_NAME = "test_queue"; public static void main(String[] args) throws IOException, TimeoutException { System.out.println("消费者启动..........2"); //创建新的连接 Connection connection = MQConnectionUtils.newConnection(); //创建Channel final Channel channel = connection.createChannel(); // 消费者关联队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); channel.basicQos(1); DefaultConsumer defaultConsumerr = 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); try { //模拟应答等待时间 Thread.sleep(300); } catch (Exception e) { }finally { channel.basicAck(envelope.getDeliveryTag(), false); //手动应答 告诉消息队列服务器 消费成功  } } }; //牵手模式设置 默认自动应答模式 true:自动应答模式 channel.basicConsume(QUEUE_NAME, false, defaultConsumerr);// fanse手动应答  } }

运行结果：

睡眠少的(执行快的) 指定的多

![ab0d949635666f2df7cb0be0cc1b46f9.png][]

![fbc055e4fce3b3403b125799e8509fa5.png][]

![9ac509ddbcc421ca00cec961b0653c11.png][]

** 注意 每个消费者 必须要应答 一下！ 队列服务器没有收到应答 就不会发送下一个给消费者~**

[ab0d949635666f2df7cb0be0cc1b46f9.png]: /images/20221123/94989117295446eb9c13cd586865fc04.png
[fbc055e4fce3b3403b125799e8509fa5.png]: /images/20221123/aa83517b15af40798274cc988b60b0f1.png
[9ac509ddbcc421ca00cec961b0653c11.png]: /images/20221123/5fb7676c41874e88a3d341498473f635.png