一個生產者對應多個消費者，但一條訊息只能有一個消費者獲得（可輪循獲取）！！！

生產者：

package com..workqueue;

import com..utils.RabbitConstant;
import com..utils.RabbitUtils;
import com.google.gson.Gson;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;

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

 
public class OrderSystem {
    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = RabbitUtils.getConnection();
        Channel channel = connection.createChannel();
        channel.queueDeclare(RabbitConstant.QUEUE_SMS, false, false, false, null
 
);
        for(int i = 1 ; i <= 10 ; i++) {
            SMS sms = new SMS("乘客" + i, "13900000" + i, "您的車票已預訂成功");
            String jsonSMS = new Gson().toJson(sms);
            channel.basicPublish("" , RabbitConstant.QUEUE_SMS , null , jsonSMS.getBytes());
        }
        System.out.println("傳送資料成功");
        channel.close();
        connection.close();
    }
}
 

3個消費者接收10條訊息：

（1）：

package com..workqueue;


import com..utils.RabbitConstant;
import com..utils.RabbitUtils;
import com.rabbitmq.client.*;

import java.io.IOException;

public class SMSSender1 {
    public static void main(String[] args) throws IOException {
        Connection connection = RabbitUtils.getConnection();
        final Channel channel = connection.createChannel();

        channel.queueDeclare(RabbitConstant.QUEUE_SMS, false, false, false, null);
        //如果不寫basicQos（1），則自動MQ會將所有請求平均傳送給所有消費者
        //basicQos,MQ不再對消費者一次傳送多個請求，而是消費者處理完一個訊息後（確認後），在從佇列中獲取一個新的
        channel.basicQos(1);//處理完一個取一個
        channel.basicConsume(RabbitConstant.QUEUE_SMS , false , new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String jsonSMS = new String(body);
                System.out.println("SMSSender1-簡訊傳送成功:" + jsonSMS);
                try {
                    Thread.sleep(10);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                channel.basicAck(envelope.getDeliveryTag() , false);
            }
        });
    }
}

後臺列印資訊：

（2）：

package com..workqueue;


import com..utils.RabbitConstant;
import com..utils.RabbitUtils;
import com.rabbitmq.client.*;

import java.io.IOException;

public class SMSSender2 {
    public static void main(String[] args) throws IOException {
        Connection connection = RabbitUtils.getConnection();
        final Channel channel = connection.createChannel();
        channel.queueDeclare(RabbitConstant.QUEUE_SMS, false, false, false, null);
        channel.basicQos(1);
        channel.basicConsume(RabbitConstant.QUEUE_SMS , false , new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String jsonSMS = new String(body);
                System.out.println("SMSSender2-簡訊傳送成功:" + jsonSMS);
                try {
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                channel.basicAck(envelope.getDeliveryTag() , false);
            }
        });
    }
}

後臺列印資訊：

（3）：

package com..workqueue;


import com..utils.RabbitConstant;
import com..utils.RabbitUtils;
import com.rabbitmq.client.*;

import java.io.IOException;

public class SMSSender3 {
    public static void main(String[] args) throws IOException {
        Connection connection = RabbitUtils.getConnection();
        final Channel channel = connection.createChannel();
        channel.queueDeclare(RabbitConstant.QUEUE_SMS, false, false, false, null);
        channel.basicQos(1);
        channel.basicConsume(RabbitConstant.QUEUE_SMS , false , new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String jsonSMS = new String(body);
                System.out.println("SMSSender3-簡訊傳送成功:" + jsonSMS);
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                channel.basicAck(envelope.getDeliveryTag() , false);
            }
        });
    }
}

後臺列印資訊：