文章目錄

• 消息隊列之輪詢分發
• • 1、什麼是輪詢
  • 2、輪詢代碼

消息隊列之輪詢分發

1、什麼是輪詢

一個生産者對應多個消費者，生産者發送多次消息，是采用輪詢的機制，公平的分給每一個消費者。

2、輪詢代碼

//發送消息
public class Send {

   private static final String  QUEUE_NAME= "test_work_queue";

   public static void main(String[] args) throws IOException, TimeoutException {
       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);
       }
       channel.close();
       connection.close();
   }
}
           

//接受者1
public class MQGet {

    private static final String  QUEUE_NAME= "test_work_queue";
    public static void main(String[] args) throws IOException, TimeoutException {
        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, AMQP.BasicProperties properties, byte[] body) throws IOException {
                super.handleDelivery(consumerTag, envelope, properties, body);
                String msg = new String(body,"utf-8");
                System.out.println("接收消息:"+msg);
            }
        };
        channel.basicConsume(QUEUE_NAME,true,consumer);
    }
}
           

//接收者2
public class MQGet2 {

    private static final String  QUEUE_NAME= "test_work_queue";
    public static void main(String[] args) throws IOException, TimeoutException {
        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, AMQP.BasicProperties properties, byte[] body) throws IOException {
                super.handleDelivery(consumerTag, envelope, properties, body);
                String msg = new String(body,"utf-8");
                System.out.println("接收消息:"+msg);
            }
        };
        channel.basicConsume(QUEUE_NAME,true,consumer);
    }
}
           

//接收者3
public class MQGet3 {

    private static final String  QUEUE_NAME= "test_work_queue";
    public static void main(String[] args) throws IOException, TimeoutException {
        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, AMQP.BasicProperties properties, byte[] body) throws IOException {
                super.handleDelivery(consumerTag, envelope, properties, body);
                String msg = new String(body,"utf-8");
                System.out.println("接收消息:"+msg);
            }
        };
        channel.basicConsume(QUEUE_NAME,true,consumer);
    }
}
           

每當發送一個消息給伺服器，伺服器都會按照順序輪流分發給3個消費者