目錄

?

?

基本概念

代碼與實例

---

?

基本概念

當某些客戶端處理比較強的時候，就多發數據讓其處理，當某些客戶端處理一般的時候，就少發數據讓其處理。

主要是讓消費者處理完后，回信息給RabbitMQ，然后RabbitMQ才會發送下一個。

使用basicQos(perfetch = 1)

注意：使用公平分發必須關閉自動應答ack改成手動。

?

代碼與實例

程序運行截圖如下：

生產者：

兩個消費者，其中一個的效率是另外一個的2倍（X2的效率）

源碼如下：

Send.java

package work;import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import util.ConnectionUtils;import java.io.IOException; import java.util.concurrent.TimeoutException;public class Send {private static final String QUEUE_NAME = "test_work_queue";public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {Connection connection = ConnectionUtils.getConnect();Channel channel = connection.createChannel();//聲明隊列channel.queueDeclare(QUEUE_NAME, false, false, false, null);//每個消費者發送確認消息之前，消息隊列不發送下一個消息到消費者，一次只處理一個//限制發送給一個消費者不超過一條int prefetchCount = 1;channel.basicQos(prefetchCount);for(int i = 0; i < 50; i++){String msg = "Hello World : " + i;System.out.println("send msg : " + msg);channel.basicPublish("", QUEUE_NAME, null, msg.getBytes());Thread.sleep(i * 20);}channel.close();connection.close();} }

Recv1.java

package work;import com.rabbitmq.client.*; import util.ConnectionUtils;import java.io.IOException; import java.util.concurrent.TimeoutException;public class Recv1 {private static final String QUEUE_NAME = "test_work_queue";public static void main(String[] args) throws IOException, TimeoutException {Connection connection = ConnectionUtils.getConnect();final Channel channel = connection.createChannel();channel.queueDeclare(QUEUE_NAME, false, false, false, null);//保證一次只發一個channel.basicQos(1);System.out.println("recv1 running");Consumer consumer = new DefaultConsumer(channel){@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {String msg = new String(body, "utf-8");System.out.println("Recv[1] msg is : " + msg);try {Thread.sleep(2000);} catch (InterruptedException e) {e.printStackTrace();} finally {System.out.println("Recv[1] done");/*關閉自動應答，手動處理*/channel.basicAck(envelope.getDeliveryTag(), false);}}};channel.basicConsume(QUEUE_NAME, false, consumer);} }

Recv2.java

package work;import com.rabbitmq.client.*; import util.ConnectionUtils;import java.io.IOException; import java.util.concurrent.TimeoutException;public class Recv2 {private static final String QUEUE_NAME = "test_work_queue";public static void main(String[] args) throws IOException, TimeoutException {Connection connection = ConnectionUtils.getConnect();final Channel channel = connection.createChannel();channel.queueDeclare(QUEUE_NAME, false, false, false, null);channel.basicQos(1);System.out.println("recv2 running");Consumer consumer = new DefaultConsumer(channel){@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {String msg = new String(body, "utf-8");System.out.println("Recv[2] msg is : " + msg);try {Thread.sleep(1000);} catch (InterruptedException e) {e.printStackTrace();} finally {System.out.println("Recv[1] done");channel.basicAck(envelope.getDeliveryTag(), false);}}};channel.basicConsume(QUEUE_NAME, false, consumer);} }

源碼下載地址：

https://github.com/fengfanchen/Java/tree/master/FairDispatch

總結

以上是生活随笔為你收集整理的Java笔记-使用RabbitMQ的Java接口实现Fair dispatch(公平分发)的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。