目錄

?

?

基本概念

代碼與實(shí)例

---

?

基本概念

簡(jiǎn)單隊(duì)列的不足：

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

? ? ? ? ? ?2. 如果生產(chǎn)者把生產(chǎn)隊(duì)列該了，消費(fèi)者也要同時(shí)改；

?

Work Queues工作隊(duì)列，模型如下：

?

?

代碼與實(shí)例

程序運(yùn)行截圖如下：

生產(chǎn)者：

兩個(gè)消費(fèi)者如下：

在源碼中，消費(fèi)者一個(gè)是1秒消費(fèi)一次，一個(gè)是2秒消費(fèi)一次，但從我們知道，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();//聲明隊(duì)列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");//消費(fèi)者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");//消費(fèi)者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(轮询分发)的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

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