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）结果就是不管谁忙谁闲，都不会多给一个消息，任务就是你一个我一个。