目錄

?

?

基本概念

代碼與實例

---

?

基本概念

簡單隊列的不足：

? ? ? ? ? ?1. 耦合性高；

? ? ? ? ? ?2. 如果生產(chǎn)者把生產(chǎn)隊列該了，消費者也要同時改；

?

Work Queues工作隊列，模型如下：

?

?

代碼與實例

程序運行截圖如下：

生產(chǎn)者：

兩個消費者如下：

在源碼中，消費者一個是1秒消費一次，一個是2秒消費一次，但從我們知道，RabbitMQ給他輪詢分發(fā)。平均分配

程序結(jié)構(gòu)如下：

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);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();Channel channel = connection.createChannel();channel.queueDeclare(QUEUE_NAME, false, false, false, null);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.basicConsume(QUEUE_NAME, true, 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();Channel channel = connection.createChannel();channel.queueDeclare(QUEUE_NAME, false, false, false, null);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[2] done");}}};channel.basicConsume(QUEUE_NAME, true, consumer);} }

源碼打包下載：

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

總結(jié)

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

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。