Java使用RabbitMQ之公平分发

发送消息：

package org.study.workfair;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import org.junit.Test;
import org.study.utils.ConnectionUtils;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class Sender {
    public static final String QUEUE_NAME = "test_simple_queue";

    @Test
    public void send() throws IOException, TimeoutException, InterruptedException {
        // 获取连接
        Connection conn = ConnectionUtils.getConnection();
        // 获取通道
        Channel channel = conn.createChannel();
        //创建队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        //每个消费者发送确认消息前，只发送一条消息
        channel.basicQos(1);
        String msg = "hello rabbitmq!";

        for (int i = 0; i < 50; i++) {
            String tempStr = i + " " + msg;
            //发送消息
            channel.basicPublish("", QUEUE_NAME, null, tempStr.getBytes());
            System.out.println("[send] msg " + i + ": " + msg);
            Thread.sleep(100);
        }

        channel.close();
        conn.close();
    }
}

接受消息：

package org.study.workfair;

import com.rabbitmq.client.*;
import org.junit.Test;
import org.study.utils.ConnectionUtils;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class Recv {
    public static final String QUEUE_NAME = "test_simple_queue";

    @Test
    public void recv() throws IOException, TimeoutException, InterruptedException {
        Connection conn = ConnectionUtils.getConnection();
        Channel channel = conn.createChannel();
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        channel.basicQos(1);

        //定义消费者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            //重写获取到达消息
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
//                super.handleDelivery(consumerTag, envelope, properties, body);
                String msg = new String(body, "utf-8");
                System.out.println("[1] recv: " + msg);

                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }finally {
                    System.out.println("[1] done!");
                    channel.basicAck(envelope.getDeliveryTag(),false);
                }
            }
        };

        while (true) {
            //监听队列
            channel.basicConsume(QUEUE_NAME, false, consumer);
            Thread.sleep(100);
        }


    }
}