RabbitMQ简单应用の公平分发（fair dipatch）

公平分发（fair dipatch）和轮询分发其实基本一致，只是每次分发的机制变了，由原来的平均分配到现在每次只处理一条消息

1.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();
    }
}

2.消息生产者Send，这里的变化是声明了“每个消费者发送确认消息之前，消息队列不发送下一个消息到消费者，一次只处理一个消息” channel.basicQos(intnum);

package com.mmr.rabbitmq.workfair;

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);
        /*
         * 每个消费者发送确认消息之前，消息队列不发送下一个消息到消费者，一次只处理一个消息
         * 
         * 限制发送给同一个消费者只能发送一条
         * */
        int prefetchCount =1;
        channel.basicQos(prefetchCount);
        
        
        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();
    }
}

3.消息处理者（消费者）Recv1 Recv2，这里的区别在于：

（1）每次只处理1条消息channel.basicQos(1);

（2）并且在消息处理完之后会手动返回回执单 channel.basicAck(envelope.getDeliveryTag(), false);

（3）最后将之前的自动应答true改为false boolean autoAck = false;

package com.mmr.rabbitmq.workfair;

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();
        
        //获取频道
        
        final Channel channel = connection.createChannel();
        
        // 声明队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        
        // 关闭自动应答
        channel.basicQos(1); // 保证每次只被分发一个
        
        // 定义一个消费者
        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");
                    // 手动回执
                    channel.basicAck(envelope.getDeliveryTag(), false);
                }
            }
        };
        // boolean autoAck = true; // 自动应答改为false
        boolean autoAck = false;
        channel.basicConsume(QUEUE_NAME, autoAck,consumer);
        
    }
}

package com.mmr.rabbitmq.workfair;

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();
        
        //获取频道
        
        final Channel channel = connection.createChannel();
        
        // 声明队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        
        // 保证每次只接收一条消息
        channel.basicQos(1);
        
        // 定义一个消费者
        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");
                    // 手动回执
                    channel.basicAck(envelope.getDeliveryTag(), false);
                }
            }
        };
        // boolean autoAck = true; // 自动应答改为false
        boolean autoAck = false;
        channel.basicConsume(QUEUE_NAME, autoAck,consumer);
        
    }
}