RabbitMQ整合Spring Booot【公平隊列】

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手動應答  } }

運行結果：

睡眠少的（執行快的） 指定的多

 注意 每一個消費者 必需要應答 一下！ 隊列服務器沒有收到應答 就不會發送下一個給消費者~