Java 整合 RabbitMQ

一、Java项目创建并整合RabbitMQ

1、创建Maven项目

2、添加依赖

官方地址： https://www.rabbitmq.com/java-client.html

依赖地址： https://mvnrepository.com/artifact/com.rabbitmq/amqp-client/5.10.0

<dependencies>
    <dependency>
        <groupId>com.rabbitmq</groupId>
        <artifactId>amqp-client</artifactId>
        <version>5.10.0</version>
    </dependency>
</dependencies>

二、RabbitMQ简单队列实战

⽂档：https://www.rabbitmq.com/tutorials/tutorial-one-java.html

1、消息生产者

package com.xdclass.simple;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.nio.charset.StandardCharsets;

public class Send {
    private final static String QUEUE_NAME = "hello";

    public static void main(String[] argv) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("192.168.216.130");
        factory.setUsername("admin");
        factory.setPassword("password");
        factory.setVirtualHost("/dev");
        factory.setPort(5672);
        try (
                //JDK7语法 或⾃动关闭 connnection 和channel
                //创建连接
                Connection connection = factory.newConnection();
                //创建信道
                Channel channel = connection.createChannel()) {
            /**
             * 队列名称
             * 持久化配置： mq重启后还在
             * 是否独占：只能有⼀个消费者监听队列；当connection关闭是否删除队列，⼀般是false，发布订阅是独占
             * ⾃动删除: 当没有消费者的时候，⾃动删除掉，⼀般是false
             * 其他参数
             *
             * 队列不存在则会⾃动创建，如果存在则不会覆盖，所以此时的时候需要注意属性
             */
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            String message = "Hello World!222222";
            /**
             * 参数说明：
             * 交换机名称：不写则是默认的交换机，那路由键需要和队列名称⼀样才可以被路由，
             * 路由键名称
             * 配置信息
             * 发送的消息数据：字节数组
             */
            channel.basicPublish("", QUEUE_NAME, null, message.getBytes(StandardCharsets.UTF_8));
            System.out.println(" [x] Sent '" + message + "'");
        }
    }
}

2、消息消费者

会⼀直监听队列

package com.xdclass.simple;

import com.rabbitmq.client.*;

import java.io.IOException;

public class Recv {
    private final static String QUEUE_NAME = "hello";

    public static void main(String[] argv) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("192.168.216.130");
        factory.setUsername("admin");
        factory.setPassword("password");
        factory.setVirtualHost("/dev");
        factory.setPort(5672);
        //消费者⼀般不增加⾃动关闭
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        System.out.println(" [*] Waiting for messages.To exit press CTRL + C");
        //回调方法，下⾯两种都⾏
        Consumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope,
                                       AMQP.BasicProperties properties, byte[] body) throws IOException {
                // consumerTag 是固定的 可以做此会话的名字，deliveryTag 每次接收消息 +1
                System.out.println("consumerTag 消息标识 = " + consumerTag);
                //可以获取交换机，路由键等System.out.println("envelope元数据 = "+envelope);
                System.out.println("properties配置信息 = " + properties);
                System.out.println("body=" + new String(body, "utf-8"));
            }
        };
        channel.basicConsume(QUEUE_NAME, true, consumer);


        //也可以用下面这种
        /*
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "UTF-8");
            System.out.println(" [x] Received '" + message + "'");
        };
        //⾃动确认消息
        channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> {
        });
        */
    }
}

三、RabbitMQ工作队列—轮询策略

1、工作队列

• 消息生产能力大于消费能力，增加多几个消费节点
• 和简单队列类似，增加多个几个消费节点，处于竞争关系
• 默认策略： round robin 轮询

2、生产者代码

package com.xdclass.work.rr;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.nio.charset.StandardCharsets;

public class Send {
    private final static String QUEUE_NAME = "work_mq_rr";

    public static void main(String[] argv) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("192.168.216.130");
        factory.setUsername("admin");
        factory.setPassword("password");
        factory.setVirtualHost("/dev");
        factory.setPort(5672);
        try (
                //JDK7语法 或⾃动关闭 connnection 和channel
                //创建连接
                Connection connection = factory.newConnection();
                //创建信道
                Channel channel = connection.createChannel()) {

            channel.queueDeclare(QUEUE_NAME, false, false, false, null);

            for (int i = 0; i < 10; i++) {
                String message = "Hello World! i==" + i;
                channel.basicPublish("", QUEUE_NAME, null, message.getBytes(StandardCharsets.UTF_8));
                System.out.println(" [x] Sent '" + message + "'");
            }
        }
    }
}

3、消费者代码，延迟1s消费

public class Recv1 {
    private final static String QUEUE_NAME = "work_mq_rr";

    public static void main(String[] argv) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("192.168.216.130");
        factory.setUsername("admin");
        factory.setPassword("password");
        factory.setVirtualHost("/dev");
        factory.setPort(5672);

        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        System.out.println(" [*]Waiting for messages. To exit press CTRL+C");

        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            //模拟消费缓慢
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            String message = new String(delivery.getBody(), "UTF-8");
            System.out.println("[x] Received '" + message + "'");
            //⼿工确认消息消费，不是多条确认
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
        };
        //关闭⾃动确认消息
        channel.basicConsume(QUEUE_NAME, false, deliverCallback, consumerTag -> {
        });
    }
}

4、消费者代码，延迟3s消费

public class Recv2 {
    private final static String QUEUE_NAME = "work_mq_rr";

    public static void main(String[] argv) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("192.168.216.130");
        factory.setUsername("admin");
        factory.setPassword("password");
        factory.setVirtualHost("/dev");
        factory.setPort(5672);

        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        System.out.println(" [*]Waiting for messages. To exit press CTRL+C");

        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            //模拟消费缓慢
            try {
                TimeUnit.SECONDS.sleep(3);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            String message = new String(delivery.getBody(), "UTF-8");
            System.out.println("[x] Received '" + message + "'");
            //⼿工确认消息消费，不是多条确认
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
        };
        //关闭⾃动确认消息
        channel.basicConsume(QUEUE_NAME, false, deliverCallback, consumerTag -> {
        });
    }
}

5、轮询策略验证

• 先启动两个消费者，再启动生产者，消息会平分到两个消费端
• 缺点：存在部分节点消费过快，部分节点消费慢，导致不能合理处理消息

四、RabbitMQ工作队列—公平策略

1、公平策略验证

• 修改消费者策略
• 解决消费者能力消费不足的问题，降低消费时间问题