RabbitMQ简单应用の轮训分发

MQ连接工厂还是之前的那个Connection

 package com.mmr.rabbitmq.util;

 import java.io.IOException;

 import com.rabbitmq.client.Connection;

 import com.rabbitmq.client.ConnectionFactory;

 public class ConnectionUtils {

     /**

      * @desc 获取Mq 的链接

      * @author zp

      * @throws IOException

      * @date 2018-7-19

      */

     public static  Connection getConnection() throws IOException {

         // 1.定义一个链接工厂

         ConnectionFactory factroy = new ConnectionFactory();

         // 2.设置服务地址

         factroy.setHost("127.0.0.1");

         // 3.设置端口号

         factroy.setPort(5672);

         // 4.vhost  设置数据库

         factroy.setVirtualHost("vhtest");

         // 5.设置用户名

         factroy.setUsername("jerry");

         // 6. 设置密码

         factroy.setPassword("123456");

         // 7.返回链接

         return factroy.newConnection();

     }

 }

消息生产者类的定义Send

 package com.mmr.rabbitmq.work;

 import java.io.IOException;

 import com.mmr.rabbitmq.util.ConnectionUtils;

 import com.rabbitmq.client.Channel;

 import com.rabbitmq.client.Connection;

 public class Send {

     /*

      *         |--C1

      * P-------|--C2

      *         |--C3

      *

      * */

     private static final String QUEUE_NAME="test_work_queue";

     public static void main(String[] args) throws IOException, InterruptedException{

         // 获取链接

         Connection connection = ConnectionUtils.getConnection();

         // 获取通道

         Channel channel = connection.createChannel();

         // 声明队列

         channel.queueDeclare(QUEUE_NAME,false,false,false,null);

         for (int i = 0; i < 50; i++) {

             String msg = "hello "+i;

             channel.basicPublish("", QUEUE_NAME, null, msg.getBytes());

             System.out.println("send msg 的第"+i+"条");

             Thread.sleep(i*20);

         }

         channel.close();

         connection.close();

     }

 }

消息消费者累的定义 Recv1  Recv2

 package com.mmr.rabbitmq.work;

 import java.io.IOException;

 import com.mmr.rabbitmq.util.ConnectionUtils;

 import com.rabbitmq.client.Channel;

 import com.rabbitmq.client.Connection;

 import com.rabbitmq.client.Consumer;

 import com.rabbitmq.client.DefaultConsumer;

 import com.rabbitmq.client.Envelope;

 import com.rabbitmq.client.AMQP.BasicProperties;

 public class Recv1 {

     private static final String QUEUE_NAME="test_work_queue";

     public static void main(String[] args) throws IOException{

         // 获取链接

         Connection connection = ConnectionUtils.getConnection();

         //获取频道

         Channel channel = connection.createChannel();

         // 声明队列

         channel.queueDeclare(QUEUE_NAME, false, false, false, null);

         // 定义一个消费者

         Consumer consumer = new DefaultConsumer(channel){

             // 一旦有消息 就会触发这个方法  消息到达

             @Override

             public void handleDelivery(String consumerTag, Envelope envelope,

                     BasicProperties properties, byte[] body) throws IOException {

                 // TODO Auto-generated method stub

                 // 拿消息

                 String msg = new String(body,"utf-8");

                 //搭出来

                 System.out.println("[1]Recv msg:"+msg);

                 try {

                     Thread.sleep(2000);

                 } catch (Exception e) {

                     // TODO: handle exception

                     e.printStackTrace();

                 }finally{

                     System.out.println("[1] done");

                 }

             }

         };

         boolean autoAck = true;

         channel.basicConsume(QUEUE_NAME, autoAck,consumer);

     }

 }

 package com.mmr.rabbitmq.work;

 import java.io.IOException;

 import com.mmr.rabbitmq.util.ConnectionUtils;

 import com.rabbitmq.client.Channel;

 import com.rabbitmq.client.Connection;

 import com.rabbitmq.client.Consumer;

 import com.rabbitmq.client.DefaultConsumer;

 import com.rabbitmq.client.Envelope;

 import com.rabbitmq.client.AMQP.BasicProperties;

 public class Recv2 {

     private static final String QUEUE_NAME="test_work_queue";

     public static void main(String[] args) throws IOException{

         // 获取链接

         Connection connection = ConnectionUtils.getConnection();

         //获取频道

         Channel channel = connection.createChannel();

         // 声明队列

         channel.queueDeclare(QUEUE_NAME, false, false, false, null);

         // 定义一个消费者

         Consumer consumer = new DefaultConsumer(channel){

             // 一旦有消息 就会触发这个方法  消息到达

             @Override

             public void handleDelivery(String consumerTag, Envelope envelope,

                     BasicProperties properties, byte[] body) throws IOException {

                 // TODO Auto-generated method stub

                 // 拿消息

                 String msg = new String(body,"utf-8");

                 //搭出来

                 System.out.println("[2]Recv msg:"+msg);

                 try {

                     Thread.sleep(1000);

                 } catch (Exception e) {

                     // TODO: handle exception

                     e.printStackTrace();

                 }finally{

                     System.out.println("[2] done");

                 }

             }

         };

         boolean autoAck = true;

         channel.basicConsume(QUEUE_NAME, autoAck,consumer);

     }

 }

1.首先我们运行Recv1  Recv2 对消息进行监听

2.其次我们运行Send，开始生产消息。

3.最后得到的结果是：消费者1（都是偶数）和消费者2（都是奇数）处理消息是一样的

为什么会出现这种现象呢？

----这种方式叫做轮训分发（round-robin）结果就是不管谁忙谁闲，都不会多给一个消息，任务就是你一个我一个。